| """API tests. Run: python -m unittest discover -s tests -v |
| |
| Covers: (1) published figures reproduced through the HTTP layer, |
| (2) API output identical to calling the reference modules directly, |
| (3) input validation, (4) HTTP behaviour, (5) rate limiting / cost guards, |
| (6) privacy (bodies never logged), (7) a live-socket smoke test. |
| """ |
| import contextlib |
| import io |
| import json |
| import os |
| import sys |
| import threading |
| import unittest |
| import urllib.error |
| import urllib.request |
|
|
| sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..")) |
|
|
| from app.guard import ConcurrencyGate, RateLimiter |
| from app.wsgi import make_app |
| from core import rmti_core as core |
| from oncology import rmti_hcc as hcc |
|
|
|
|
| def call(app, method, path, body=None, headers=None, raw=None, ctype="application/json", query=""): |
| data = raw if raw is not None else (json.dumps(body).encode() if body is not None else b"") |
| env = {"REQUEST_METHOD": method, "PATH_INFO": path, "QUERY_STRING": query, |
| "REMOTE_ADDR": "203.0.113.9", "CONTENT_LENGTH": str(len(data)), |
| "wsgi.input": io.BytesIO(data)} |
| if method == "POST" or raw is not None: |
| env["CONTENT_TYPE"] = ctype |
| for k, v in (headers or {}).items(): |
| env["HTTP_" + k.upper().replace("-", "_")] = v |
| cap = {} |
|
|
| def sr(status, hdrs): |
| cap["status"], cap["headers"] = int(status.split()[0]), dict(hdrs) |
|
|
| out = b"".join(app(env, sr)) |
| parsed = None |
| if out and cap["headers"].get("Content-Type", "").startswith("application/json"): |
| parsed = json.loads(out) |
| return cap["status"], parsed, cap["headers"], out |
|
|
|
|
| def fresh_app(**kw): |
| kw.setdefault("log", False) |
| kw.setdefault("limiter", RateLimiter(per_client_rpm=6000, per_client_burst=10000, |
| global_rpm=60000, global_burst=100000)) |
| return make_app(**kw) |
|
|
|
|
| class PublishedFigures(unittest.TestCase): |
| def setUp(self): |
| self.app = fresh_app() |
|
|
| def assess(self, **kw): |
| s, j, _, _ = call(self.app, "POST", "/v1/disaster/assess", kw) |
| self.assertEqual(s, 200, j) |
| return j["result"] |
|
|
| def test_three_primary_cities(self): |
| cases = [ |
| (dict(N=4_500_000, L=0.8, E_sel=0.85, V=0.75, rho_now=0.20, rho_after=0.58, C=4), |
| 0.408, "Tier 4 - Severe", 0.214, "Tier 3 - High", 47.5, 0.480, 0.252), |
| (dict(N=16_000_000, L=0.9, E_sel=0.65, V=0.8, rho_now=0.20, rho_after=0.55, C=5), |
| 0.374, "Tier 4 - Severe", 0.211, "Tier 3 - High", 43.8, 0.720, 0.405), |
| (dict(N=1_100_000, L=0.875, E_sel=0.8, V=0.8, rho_now=0.25, rho_after=0.48, C=4), |
| 0.420, "Tier 4 - Severe", 0.291, "Tier 3 - High", 30.7, 0.492, 0.341), |
| ] |
| for kw, rn, tn, ra, ta, red, ean, eaa in cases: |
| r = self.assess(**kw) |
| self.assertAlmostEqual(r["R_now"], rn, 3) |
| self.assertEqual(r["tier_now"], tn) |
| self.assertAlmostEqual(r["R_after"], ra, 3) |
| self.assertEqual(r["tier_after"], ta) |
| self.assertEqual(r["risk_reduction_pct"], red) |
| self.assertAlmostEqual(r["EAL_now"], ean, 3) |
| self.assertAlmostEqual(r["EAL_after"], eaa, 3) |
|
|
| def test_lilongwe_equity_rate(self): |
| r = self.assess(N=1_100_000, L=0.875, E_sel=0.8, V=0.8, rho_now=0.25, C=4) |
| self.assertAlmostEqual(r["equity_now_per_1M"], 0.4474, 4) |
|
|
| def test_E_alias(self): |
| a = self.assess(N=1_100_000, L=0.875, E=0.8, V=0.8, rho_now=0.25) |
| b = self.assess(N=1_100_000, L=0.875, E_sel=0.8, V=0.8, rho_now=0.25) |
| self.assertEqual(a, b) |
|
|
| def test_cohort_endpoint(self): |
| s, j, h, _ = call(self.app, "GET", "/v1/disaster/cohort") |
| self.assertEqual(s, 200) |
| self.assertEqual(j["count"], 8) |
| by = {r["name"]: r for r in j["ranked"]} |
| self.assertAlmostEqual(by["Bamako"]["R_now"], 0.408, 3) |
| self.assertAlmostEqual(by["Lilongwe"]["R_after"], 0.291, 3) |
| for name in ("Nairobi", "Dar es Salaam", "Lusaka"): |
| self.assertNotIn("R_after", by[name]) |
| self.assertAlmostEqual(by["Nairobi"]["R_now"], 0.196, 3) |
| self.assertEqual(by["Lagos"]["priority_in_tier"], 1) |
| self.assertIn("max-age", h["Cache-Control"]) |
|
|
| def test_oncology_rounding_regression(self): |
| """0.30*0.65*0.60*(1-0.50) = 0.0585 must round HALF_UP to 0.059 (not 0.058).""" |
| s, j, _, _ = call(self.app, "POST", "/v1/oncology/assess", |
| dict(N=100_000, L=0.30, E=0.65, V=0.60, rho_now=0.20, rho_after=0.50)) |
| self.assertEqual(s, 200, j) |
| self.assertAlmostEqual(j["result"]["R_now"], 0.094, 3) |
| self.assertAlmostEqual(j["result"]["R_after"], 0.059, 3) |
| self.assertEqual(j["result"]["tier_now"], "2 Moderate") |
|
|
| def test_oncology_albi_cohort_v(self): |
| s, j, _, _ = call(self.app, "POST", "/v1/oncology/assess", |
| dict(N=29, L=0.3, E=0.6, albi_grade_counts={"1": 10, "2": 18, "3": 1}, rho_now=0.2)) |
| self.assertEqual(s, 200, j) |
| self.assertAlmostEqual(j["inputs"]["V"], 0.5224, 3) |
| self.assertEqual(j["provenance"]["V_source"], "albi_grade_counts") |
|
|
| def test_oncology_bclc_stage_maps_to_C(self): |
| s, j, _, _ = call(self.app, "POST", "/v1/oncology/assess", |
| dict(N=1000, L=0.3, E=0.6, V=0.5, rho_now=0.2, bclc_stage="BCLC_A")) |
| self.assertEqual(j["inputs"]["C"], 2) |
| self.assertAlmostEqual(j["result"]["EAL_now"], 0.3 * 0.30 * 0.8, 3) |
|
|
| def test_multi_matches_reference(self): |
| hz = [ |
| {"name": "coastal flood", "L": 0.90, "rho_hazard": 0.20, "C": 5, "categories": { |
| "social": {"E": 0.70, "V": 0.80, "rho": 0.20}, |
| "economic": {"E": 0.60, "V": 0.75, "rho": 0.25}, |
| "environmental": {"E": 0.65, "V": 0.70, "rho": 0.20}}}, |
| {"name": "pluvial", "L": 0.85, "rho_hazard": 0.28, "C": 4, "categories": { |
| "social": {"E": 0.55, "V": 0.70, "rho": 0.30}, |
| "economic": {"E": 0.50, "V": 0.65, "rho": 0.30}, |
| "environmental": {"E": 0.60, "V": 0.68, "rho": 0.25}}}] |
| w = {"social": 0.4, "economic": 0.3, "environmental": 0.3} |
| s, j, _, _ = call(self.app, "POST", "/v1/multi", {"N": 16_000_000, "hazards": hz, "weights": w}) |
| self.assertEqual(s, 200, j) |
| self.assertAlmostEqual(j["result"]["R"], 0.578, 3) |
| self.assertAlmostEqual(j["result"]["EAL"], 1.179, 3) |
| self.assertAlmostEqual(j["result"]["equity_per_1M"], 0.0737, 4) |
| self.assertIn("not field-validated", j["warning"]) |
| s, j, _, _ = call(self.app, "POST", "/v1/multi", |
| {"hazards": hz, "weights": {"social": 0.5, "economic": 0.3, "environmental": 0.3}}) |
| self.assertEqual((s, j["error"]["code"]), (400, "invalid_weights")) |
|
|
| def test_exposure_helper(self): |
| s, j, _, _ = call(self.app, "GET", "/v1/exposure", query="N=1100000") |
| self.assertEqual(j["E_log"], 0.8) |
| s, j, _, _ = call(self.app, "GET", "/v1/exposure", query="N=100000&domain=oncology") |
| self.assertEqual(j["bounds"], {"N_min": 100, "N_max": 5_000_000}) |
|
|
| def test_framework_tiers(self): |
| s, j, _, _ = call(self.app, "GET", "/v1/framework") |
| self.assertEqual([t["upper_bound_exclusive"] for t in j["tiers"]], [0.05, 0.15, 0.35, 0.60, None]) |
| self.assertEqual(j["consequence_scale_C"]["4"], 0.75) |
|
|
|
|
| class MatchesReferenceModules(unittest.TestCase): |
| def test_oncology_rank_equals_direct_call(self): |
| app = fresh_app() |
| items = [{"label": n, **{k: v for k, v in r.items() if k != "incidence"}} |
| for n, r in hcc.PUBLISHED_COHORTS.items()] |
| s, j, _, _ = call(app, "POST", "/v1/oncology/rank", {"items": items}) |
| self.assertEqual(s, 200, j) |
| direct = hcc.rank_cohort_hcc(hcc._scoreable(hcc.PUBLISHED_COHORTS)) |
| self.assertEqual([r["name"] for r in j["ranked"]], [r["name"] for r in direct]) |
| for a, b in zip(j["ranked"], direct): |
| for k in ("R_now", "R_after", "tier_now", "tier_after", "EAL_now", "priority_in_tier"): |
| self.assertEqual(a[k], b[k], (a["name"], k)) |
|
|
| def test_disaster_rank_equals_core(self): |
| app = fresh_app() |
| rows = {"A": dict(N=1e6, L=.5, E_sel=.5, V=.5, rho_now=.2, rho_after=.5, C=3), |
| "B": dict(N=2e6, L=.9, E_sel=.9, V=.9, rho_now=.1, C=5)} |
| items = [{"label": k, **v} for k, v in rows.items()] |
| s, j, _, _ = call(app, "POST", "/v1/disaster/rank", {"items": items}) |
| direct = core.rank_cohort(rows) |
| self.assertEqual([r["name"] for r in j["ranked"]], [r["name"] for r in direct]) |
| self.assertEqual(j["ranked"][0]["name"], "B") |
|
|
|
|
| class Validation(unittest.TestCase): |
| def setUp(self): |
| self.app = fresh_app() |
| self.good = dict(N=1000, L=0.5, E_sel=0.5, V=0.5, rho_now=0.2) |
|
|
| def expect(self, path, body, code, status=400): |
| s, j, _, _ = call(self.app, "POST", path, body) |
| self.assertEqual((s, j["error"]["code"]), (status, code), j) |
| return j |
|
|
| def test_unknown_field_rejected_and_identifier_warning(self): |
| j = self.expect("/v1/oncology/assess", |
| dict(N=10, L=.3, E=.5, V=.5, rho_now=.2, patient_name="X"), "unknown_field") |
| self.assertIn("identifiers", j["error"]["message"]) |
| self.expect("/v1/disaster/assess", {**self.good, "mrn": "123"}, "unknown_field") |
|
|
| def test_ranges_and_types(self): |
| self.expect("/v1/disaster/assess", {**self.good, "L": 1.2}, "out_of_range") |
| self.expect("/v1/disaster/assess", {**self.good, "V": -0.1}, "out_of_range") |
| self.expect("/v1/disaster/assess", {**self.good, "N": 0}, "out_of_range") |
| self.expect("/v1/disaster/assess", {**self.good, "L": True}, "invalid_type") |
| self.expect("/v1/disaster/assess", {**self.good, "L": "0.5"}, "invalid_type") |
| self.expect("/v1/disaster/assess", {**self.good, "C": 4.5}, "invalid_type") |
| self.expect("/v1/disaster/assess", {**self.good, "C": 6}, "out_of_range") |
| bad = dict(self.good); del bad["V"] |
| self.expect("/v1/disaster/assess", bad, "missing_field") |
| self.expect("/v1/disaster/assess", {**self.good, "E": 0.4}, "conflicting_fields") |
|
|
| def test_rho_now_one_with_rho_after(self): |
| self.expect("/v1/disaster/assess", {**self.good, "rho_now": 1.0, "rho_after": 1.0}, "invalid_value") |
| s, j, _, _ = call(self.app, "POST", "/v1/disaster/assess", {**self.good, "rho_now": 1.0}) |
| self.assertEqual(s, 200) |
|
|
| def test_nan_infinity_and_bad_json(self): |
| for raw in (b'{"N": NaN}', b'{"N": Infinity}', b'{bad', b'\xff\xfe'): |
| s, j, _, _ = call(self.app, "POST", "/v1/disaster/assess", raw=raw) |
| self.assertEqual((s, j["error"]["code"]), (400, "invalid_json")) |
| s, j, _, _ = call(self.app, "POST", "/v1/disaster/assess", raw=b'[1,2]') |
| self.assertEqual(j["error"]["code"], "invalid_body") |
|
|
| def test_oncology_vulnerability_source_exclusive(self): |
| base = dict(N=10, L=.3, E=.5, rho_now=.2) |
| self.expect("/v1/oncology/assess", base, "invalid_vulnerability") |
| self.expect("/v1/oncology/assess", {**base, "V": .5, "albi_grade": 2}, "invalid_vulnerability") |
| self.expect("/v1/oncology/assess", {**base, "albi_grade": 4}, "out_of_range") |
| self.expect("/v1/oncology/assess", {**base, "albi_grade_counts": {"1": 0}}, "invalid_value") |
| self.expect("/v1/oncology/assess", {**base, "albi_grade_counts": {"9": 1}}, "invalid_value") |
| self.expect("/v1/oncology/assess", {**base, "V": .5, "C": 3, "bclc_stage": "BCLC_A"}, "conflicting_fields") |
| self.expect("/v1/oncology/assess", {**base, "V": .5, "bclc_stage": "nope"}, "invalid_value") |
|
|
| def test_oncology_E_defaults_to_log_helper(self): |
| s, j, _, _ = call(self.app, "POST", "/v1/oncology/assess", dict(N=100_000, L=.3, V=.5, rho_now=.2)) |
| self.assertEqual(s, 200) |
| self.assertAlmostEqual(j["inputs"]["E"], hcc.exposure_from_population(100_000)) |
| self.assertIn("log-normalised", j["provenance"]["E_source"]) |
|
|
| def test_rank_constraints(self): |
| row = {"label": "a", **self.good} |
| self.expect("/v1/disaster/rank", {"items": []}, "invalid_body") |
| self.expect("/v1/disaster/rank", {"items": [row, row]}, "duplicate_label") |
| self.expect("/v1/disaster/rank", {"items": [{**self.good}]}, "invalid_label") |
| self.expect("/v1/disaster/rank", {"items": [{**row, "label": "x" * 61}]}, "invalid_label") |
| many = [{**row, "label": f"c{i}"} for i in range(101)] |
| self.expect("/v1/disaster/rank", {"items": many}, "too_many_items") |
| ok = [{**row, "label": f"c{i}"} for i in range(100)] |
| s, j, _, _ = call(self.app, "POST", "/v1/disaster/rank", {"items": ok}) |
| self.assertEqual((s, j["count"]), (200, 100)) |
|
|
| def test_notices_present(self): |
| s, j, _, _ = call(self.app, "POST", "/v1/oncology/assess", |
| dict(N=10, L=.3, E=.5, V=.5, rho_now=.2)) |
| self.assertIn("Not a medical device", j["notice"]) |
| self.assertIn("patient-identifiable", j["notice"]) |
| s, j, _, _ = call(self.app, "POST", "/v1/disaster/assess", self.good) |
| self.assertIn("not forecasts", j["notice"]) |
|
|
|
|
| class HttpBehaviour(unittest.TestCase): |
| def setUp(self): |
| self.app = fresh_app() |
|
|
| def test_basic_routes(self): |
| self.assertEqual(call(self.app, "GET", "/health")[0], 200) |
| self.assertEqual(call(self.app, "GET", "/")[0], 200) |
| s, _, h, body = call(self.app, "GET", "/docs") |
| self.assertEqual(s, 200) |
| self.assertIn("text/html", h["Content-Type"]) |
| self.assertIn(b"swagger-ui", body) |
| s, j, _, _ = call(self.app, "GET", "/openapi.json") |
| self.assertEqual(j["openapi"], "3.0.3") |
| self.assertIn("/v1/oncology/assess", j["paths"]) |
| self.assertEqual(call(self.app, "GET", "/v1/framework/")[0], 200) |
|
|
| def test_404_405(self): |
| self.assertEqual(call(self.app, "GET", "/nope")[0], 404) |
| s, j, h, _ = call(self.app, "GET", "/v1/disaster/assess") |
| self.assertEqual((s, h["Allow"]), (405, "POST")) |
| self.assertEqual(call(self.app, "POST", "/v1/framework", {})[0], 405) |
|
|
| def test_cors_and_security_headers(self): |
| s, _, h, out = call(self.app, "OPTIONS", "/v1/disaster/assess") |
| self.assertEqual((s, out), (204, b"")) |
| self.assertEqual(h["Access-Control-Allow-Origin"], "*") |
| s, _, h, _ = call(self.app, "GET", "/v1/framework") |
| self.assertEqual(h["X-Content-Type-Options"], "nosniff") |
|
|
| def test_head(self): |
| s, _, h, out = call(self.app, "HEAD", "/v1/framework") |
| self.assertEqual((s, out), (200, b"")) |
| self.assertGreater(int(h["Content-Length"]), 10) |
|
|
| def test_body_limits_and_content_type(self): |
| s, j, _, _ = call(self.app, "POST", "/v1/disaster/assess", raw=b" " * 40000) |
| self.assertEqual((s, j["error"]["code"]), (413, "payload_too_large")) |
| s, j, _, _ = call(self.app, "POST", "/v1/disaster/assess", raw=b"{}", ctype="text/plain") |
| self.assertEqual(s, 415) |
| env = {"REQUEST_METHOD": "POST", "PATH_INFO": "/v1/disaster/assess", "QUERY_STRING": "", |
| "CONTENT_TYPE": "application/json", "wsgi.input": io.BytesIO(b"{}"), "REMOTE_ADDR": "1.1.1.1"} |
| got = {} |
| self.app(env, lambda st, h: got.setdefault("s", st)) |
| self.assertTrue(got["s"].startswith("411")) |
|
|
| def test_query_validation(self): |
| self.assertEqual(call(self.app, "GET", "/v1/exposure")[0], 400) |
| self.assertEqual(call(self.app, "GET", "/v1/exposure", query="N=abc")[0], 400) |
| self.assertEqual(call(self.app, "GET", "/v1/exposure", query="N=5&x=1")[0], 400) |
| self.assertEqual(call(self.app, "GET", "/v1/exposure", query="N=5&domain=zzz")[0], 400) |
| self.assertEqual(call(self.app, "GET", "/v1/exposure", query="N=" + "9" * 600)[0], 400) |
|
|
| def test_api_key_when_configured(self): |
| app = fresh_app(api_key="s3cret") |
| body = dict(N=10, L=.3, E_sel=.5, V=.5, rho_now=.2) |
| self.assertEqual(call(app, "POST", "/v1/disaster/assess", body)[0], 401) |
| self.assertEqual(call(app, "POST", "/v1/disaster/assess", body, headers={"X-API-Key": "bad"})[0], 401) |
| self.assertEqual(call(app, "POST", "/v1/disaster/assess", body, headers={"X-API-Key": "s3cret"})[0], 200) |
| self.assertEqual(call(app, "GET", "/v1/framework")[0], 200) |
|
|
| def test_unexpected_error_is_generic(self): |
| from app import routes |
| orig = routes.framework |
| try: |
| def boom(): |
| raise RuntimeError("secret internals") |
| from app import wsgi |
| wsgi.ROUTES[("GET", "/v1/framework")] = (boom, "get") |
| s, j, _, out = call(fresh_app(), "GET", "/v1/framework") |
| self.assertEqual((s, j["error"]["code"]), (500, "internal_error")) |
| self.assertNotIn(b"secret", out) |
| finally: |
| wsgi.ROUTES[("GET", "/v1/framework")] = (orig, "get") |
|
|
|
|
| class CostGuards(unittest.TestCase): |
| def test_per_client_limit_and_retry_after(self): |
| t = [0.0] |
| lim = RateLimiter(per_client_rpm=60, per_client_burst=3, global_rpm=6000, global_burst=1000, |
| clock=lambda: t[0]) |
| app = make_app(limiter=lim, log=False) |
| codes = [call(app, "GET", "/v1/framework")[0] for _ in range(5)] |
| self.assertEqual(codes, [200, 200, 200, 429, 429]) |
| s, j, h, _ = call(app, "GET", "/v1/framework") |
| self.assertEqual(j["error"]["code"], "rate_limited") |
| self.assertGreaterEqual(int(h["Retry-After"]), 1) |
| t[0] += 2.0 |
| self.assertEqual(call(app, "GET", "/v1/framework")[0], 200) |
|
|
| def test_health_is_not_limited(self): |
| lim = RateLimiter(per_client_rpm=1, per_client_burst=1, global_rpm=1, global_burst=1, |
| clock=lambda: 0.0) |
| app = make_app(limiter=lim, log=False) |
| self.assertTrue(all(call(app, "GET", "/health")[0] == 200 for _ in range(20))) |
|
|
| def test_global_limit_protects_the_free_allowance(self): |
| t = [0.0] |
| lim = RateLimiter(per_client_rpm=6000, per_client_burst=1000, global_rpm=60, global_burst=5, |
| clock=lambda: t[0]) |
| app = make_app(limiter=lim, log=False) |
| codes = [] |
| for i in range(8): |
| env_ip = f"198.51.100.{i}" |
| codes.append(call(app, "GET", "/v1/framework", headers={"X-Forwarded-For": f"6.6.6.6, {env_ip}"})[0]) |
| self.assertEqual(codes.count(200), 5) |
| self.assertEqual(codes.count(429), 3) |
|
|
| def test_forged_xff_prefix_cannot_evade_client_limit(self): |
| t = [0.0] |
| lim = RateLimiter(per_client_rpm=60, per_client_burst=2, global_rpm=6000, global_burst=1000, |
| clock=lambda: t[0]) |
| app = make_app(limiter=lim, log=False) |
| codes = [call(app, "GET", "/v1/framework", |
| headers={"X-Forwarded-For": f"10.0.0.{i}, 203.0.113.50"})[0] for i in range(4)] |
| self.assertEqual(codes, [200, 200, 429, 429]) |
|
|
| def test_client_table_is_bounded(self): |
| lim = RateLimiter(max_clients=100, clock=lambda: 0.0, global_burst=10**6, global_rpm=10**6) |
| for i in range(1000): |
| lim.check(f"ip{i}") |
| self.assertLessEqual(len(lim._clients), 100) |
|
|
| def test_concurrency_gate_returns_503(self): |
| gate = ConcurrencyGate(1) |
| app = fresh_app(gate=gate) |
| self.assertTrue(gate.acquire()) |
| s, j, h, _ = call(app, "GET", "/v1/framework") |
| self.assertEqual((s, j["error"]["code"], h["Retry-After"]), (503, "busy", "2")) |
| gate.release() |
| self.assertEqual(call(app, "GET", "/v1/framework")[0], 200) |
|
|
|
|
| class Privacy(unittest.TestCase): |
| def test_request_bodies_and_queries_are_never_logged(self): |
| app = make_app(limiter=RateLimiter(per_client_rpm=6000, per_client_burst=10000, |
| global_rpm=60000, global_burst=100000), log=True) |
| sentinel = "SENTINEL-PATIENT-12345" |
| buf = io.StringIO() |
| with contextlib.redirect_stderr(buf): |
| call(app, "POST", "/v1/oncology/assess", |
| dict(N=10, L=.3, E=.5, V=.5, rho_now=.2, patient_name=sentinel)) |
| call(app, "POST", "/v1/oncology/rank", |
| {"items": [{"label": sentinel, "N": 10, "L": .3, "E": .5, "V": .5, "rho_now": .2}]}) |
| call(app, "GET", "/v1/exposure", query=f"N=5&{sentinel}=1") |
| logged = buf.getvalue() |
| self.assertNotIn(sentinel, logged) |
| self.assertIn("POST /v1/oncology/assess 400", logged) |
| self.assertIn("POST /v1/oncology/rank 200", logged) |
|
|
|
|
| class LiveServer(unittest.TestCase): |
| """Real sockets, real server class, real HTTP client.""" |
|
|
| @classmethod |
| def setUpClass(cls): |
| from wsgiref.simple_server import make_server |
| from app.server import QuietHandler, ThreadedServer |
| cls.srv = make_server("127.0.0.1", 0, fresh_app(), server_class=ThreadedServer, |
| handler_class=QuietHandler) |
| cls.port = cls.srv.server_address[1] |
| threading.Thread(target=cls.srv.serve_forever, daemon=True).start() |
|
|
| @classmethod |
| def tearDownClass(cls): |
| cls.srv.shutdown() |
| cls.srv.server_close() |
|
|
| def req(self, path, body=None): |
| r = urllib.request.Request(f"http://127.0.0.1:{self.port}{path}", |
| data=None if body is None else json.dumps(body).encode(), |
| headers={"Content-Type": "application/json"}) |
| with urllib.request.urlopen(r, timeout=5) as resp: |
| return resp.status, json.loads(resp.read()) |
|
|
| def test_roundtrip(self): |
| self.assertEqual(self.req("/health")[1]["status"], "ok") |
| s, j = self.req("/v1/disaster/assess", dict(N=1_100_000, L=.875, E_sel=.8, V=.8, |
| rho_now=.25, rho_after=.48, C=4)) |
| self.assertEqual((s, j["result"]["R_after"]), (200, 0.291)) |
| with self.assertRaises(urllib.error.HTTPError) as cm: |
| self.req("/v1/disaster/assess", {"nope": 1}) |
| self.assertEqual(cm.exception.code, 400) |
|
|
| def test_parallel_requests(self): |
| results = [] |
|
|
| def go(): |
| results.append(self.req("/v1/oncology/cohorts")[0]) |
| ts = [threading.Thread(target=go) for _ in range(20)] |
| [t.start() for t in ts] |
| [t.join() for t in ts] |
| self.assertEqual(results, [200] * 20) |
|
|
|
|
| if __name__ == "__main__": |
| unittest.main() |
|
|
