diff --git a/src/api/server.py b/src/api/server.py index 22e4057..29caf75 100644 --- a/src/api/server.py +++ b/src/api/server.py @@ -40,6 +40,7 @@ def build_api_dependencies() -> dict: STATIC_DIR = Path(__file__).resolve().parents[1] / "web" +MAX_TRAINING_LIMIT = 100 def route_request( @@ -80,7 +81,7 @@ def route_request( status = _get_single(query, "status") skill_tag = _get_single(query, "skill_tag") instrument_id = _get_optional_int(query, "instrument_id") - limit = _get_int(query, "limit", 100) + limit = _get_limit(query, "limit", 100, max_value=MAX_TRAINING_LIMIT) return 200, { "items": deps["training_cases"]( status=status, @@ -106,7 +107,7 @@ def route_request( if method == "GET" and path == "/training/progress": user_id = _get_single(query, "user_id", "local") skill_tag = _get_single(query, "skill_tag") - limit = _get_int(query, "limit", 100) + limit = _get_limit(query, "limit", 100, max_value=MAX_TRAINING_LIMIT) return 200, {"items": deps["training_progress"](user_id=user_id, skill_tag=skill_tag, limit=limit)} return 404, {"error": "not_found"} @@ -120,11 +121,11 @@ class APIRequestHandler(BaseHTTPRequestHandler): self._serve_static(parsed.path) return try: - status, payload = route_request(parsed.path, parse_qs(parsed.query), dependencies=self.dependencies) + status, payload = route_request(parsed.path, parse_qs(parsed.query, keep_blank_values=True), dependencies=self.dependencies) except ValueError as exc: status, payload = 400, {"error": str(exc)} - except Exception as exc: - status, payload = 500, {"error": "internal_server_error", "detail": str(exc)} + except Exception: + status, payload = 500, {"error": "internal_server_error"} self._send_json(status, payload) def do_POST(self) -> None: @@ -140,8 +141,8 @@ class APIRequestHandler(BaseHTTPRequestHandler): ) except ValueError as exc: status, payload = 400, {"error": str(exc)} - except Exception as exc: - status, payload = 500, {"error": "internal_server_error", "detail": str(exc)} + except Exception: + status, payload = 500, {"error": "internal_server_error"} self._send_json(status, payload) def _send_json(self, status: int, payload: dict) -> None: @@ -206,6 +207,19 @@ def _get_int(query: dict, key: str, default: int) -> int: raise ValueError(f"invalid_int_query_param:{key}") from exc +def _get_limit(query: dict, key: str, default: int, max_value: int) -> int: + value = _get_single(query, key) + if value is None: + return default + try: + parsed = int(value) + except ValueError as exc: + raise ValueError(f"invalid_limit_query_param:{key}") from exc + if parsed < 1 or parsed > max_value: + raise ValueError(f"invalid_limit_query_param:{key}") + return parsed + + def _get_single(query: dict, key: str, default: Optional[str] = None) -> Optional[str]: values = query.get(key) if not values: diff --git a/tests/unit/test_training_api_m4.py b/tests/unit/test_training_api_m4.py new file mode 100644 index 0000000..d6d0900 --- /dev/null +++ b/tests/unit/test_training_api_m4.py @@ -0,0 +1,127 @@ +import io +import json +import unittest + +from src.api.server import APIRequestHandler + + +def _dependencies(cases=None, progress=None) -> dict: + return { + "training_cases": cases or (lambda **kwargs: []), + "training_progress": progress or (lambda **kwargs: []), + } + + +class _GetHandlerHarness(APIRequestHandler): + def __init__(self, path: str, dependencies: dict) -> None: + self.path = path + self.wfile = io.BytesIO() + self.dependencies = dependencies + self.status = None + self.response_headers = {} + + def send_response(self, code, message=None): # noqa: D401 + self.status = code + + def send_header(self, keyword, value): + self.response_headers[keyword] = value + + def end_headers(self): + return None + + def log_message(self, format, *args): + return None + + def json_body(self): + return json.loads(self.wfile.getvalue().decode("utf-8")) + + +class TrainingApiM4Tests(unittest.TestCase): + def test_training_cases_rejects_invalid_limits(self) -> None: + calls = [] + + def list_cases(**kwargs): + calls.append(kwargs) + return [] + + for path in ( + "/training/cases?limit=", + "/training/cases?limit=0", + "/training/cases?limit=-1", + "/training/cases?limit=101", + "/training/cases?limit=not-a-number", + ): + with self.subTest(path=path): + handler = _GetHandlerHarness(path, _dependencies(cases=list_cases)) + + handler.do_GET() + + self.assertEqual(handler.status, 400) + self.assertEqual(handler.json_body(), {"error": "invalid_limit_query_param:limit"}) + + self.assertEqual(calls, []) + + def test_training_progress_rejects_invalid_limits(self) -> None: + calls = [] + + def get_progress(**kwargs): + calls.append(kwargs) + return [] + + for path in ( + "/training/progress?limit=", + "/training/progress?limit=0", + "/training/progress?limit=-1", + "/training/progress?limit=101", + "/training/progress?limit=not-a-number", + ): + with self.subTest(path=path): + handler = _GetHandlerHarness(path, _dependencies(progress=get_progress)) + + handler.do_GET() + + self.assertEqual(handler.status, 400) + self.assertEqual(handler.json_body(), {"error": "invalid_limit_query_param:limit"}) + + self.assertEqual(calls, []) + + def test_training_limits_accept_upper_bound(self) -> None: + case_calls = [] + progress_calls = [] + + def list_cases(**kwargs): + case_calls.append(kwargs) + return [] + + def get_progress(**kwargs): + progress_calls.append(kwargs) + return [] + + cases_handler = _GetHandlerHarness("/training/cases?limit=100", _dependencies(cases=list_cases)) + progress_handler = _GetHandlerHarness("/training/progress?limit=100", _dependencies(progress=get_progress)) + + cases_handler.do_GET() + progress_handler.do_GET() + + self.assertEqual(cases_handler.status, 200) + self.assertEqual(progress_handler.status, 200) + self.assertEqual(case_calls[0]["limit"], 100) + self.assertEqual(progress_calls[0]["limit"], 100) + + def test_unknown_exception_response_hides_internal_detail(self) -> None: + def list_cases(**kwargs): + raise RuntimeError("raw internal failure marker") + + handler = _GetHandlerHarness("/training/cases", _dependencies(cases=list_cases)) + + handler.do_GET() + + body = handler.json_body() + serialized = json.dumps(body) + self.assertEqual(handler.status, 500) + self.assertEqual(body, {"error": "internal_server_error"}) + self.assertNotIn("raw internal failure marker", serialized) + + +if __name__ == "__main__": + unittest.main()