"""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 # noqa: E402 from app.wsgi import make_app # noqa: E402 from core import rmti_core as core # noqa: E402 from oncology import rmti_hcc as hcc # noqa: E402 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"): # baseline-only, as in the paper self.assertNotIn("R_after", by[name]) self.assertAlmostEqual(by["Nairobi"]["R_now"], 0.196, 3) self.assertEqual(by["Lagos"]["priority_in_tier"], 1) # highest EAL in Tier 4 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) # paper's Lilongwe walkthrough 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) # fine without rho_after 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) # trailing slash 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) # reads stay open 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 # tokens refill 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): # 8 different clients 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]) # keyed on the proxy-appended entry 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)) # rejected 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()