128 lines
3.9 KiB
Python
128 lines
3.9 KiB
Python
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()
|