""" Tests for infra.api (Stage 3b). The real logic lives in infra.api.handlers (pure, no fastapi) — that is what this file covers thoroughly. The FastAPI app in app.py is thin wiring; it gets a couple of smoke tests, skipped when fastapi/httpx aren't installed. """ import pytest from infra.api import handlers from infra.api.errors import ApiError from infra.service import OrgStateService @pytest.fixture def svc(): service = OrgStateService(":memory:") yield service service.close() def _degrading_body(entity_id="w1", metric="m"): """A request-shaped observation list with one clearly degrading entity.""" series = [100.0] * 20 + [100.0 + j * 6 for j in range(20)] return [ {"entity_id": entity_id, "day": f"2026-01-{i + 1:02d}", "values": {metric: v}} for i, v in enumerate(series) ] # the logistics 'team' entity type uses metric names we don't control here, so # the handler-level tests use the logistics vertical's real config via a body # built from the adapter. yaml is needed to load that config. yaml = pytest.importorskip("yaml") from verticals.logistics import adapter # noqa: E402 def _warehouse_body(): obs = adapter.load_warehouse_observations() return [{"entity_id": o.entity_id, "day": o.day, "values": o.values} for o in obs] # --- health / tenants ---------------------------------------------------- def test_health_reports_ok(svc): out = handlers.health(svc) assert out["status"] == "ok" assert out["db"]["backend"] == "sqlite" def test_register_and_list_tenants(svc): handlers.register_tenant(svc, {"tenant_id": "acme", "name": "ACME"}) assert handlers.get_tenant(svc, "acme")["name"] == "ACME" assert handlers.list_tenants(svc)["tenants"][0]["tenant_id"] == "acme" def test_register_tenant_missing_fields_is_bad_request(svc): with pytest.raises(ApiError) as e: handlers.register_tenant(svc, {"name": "no id"}) assert e.value.code == "bad_request" and e.value.status == 400 def test_register_tenant_unknown_vertical_is_not_found(svc): with pytest.raises(ApiError) as e: handlers.register_tenant(svc, {"tenant_id": "x", "name": "X", "vertical": "teleportation"}) assert e.value.status == 404 def test_get_unknown_tenant_is_not_found(svc): with pytest.raises(ApiError) as e: handlers.get_tenant(svc, "ghost") assert e.value.code == "not_found" and e.value.status == 404 # --- calibration --------------------------------------------------------- def test_calibrate_returns_per_metric_scales(svc): handlers.register_tenant(svc, {"tenant_id": "acme", "name": "ACME"}) out = handlers.calibrate(svc, "acme", { "vertical": "logistics", "entity_type": "warehouse", "observations": _warehouse_body(), }) assert set(out["metrics"]) == {"backlog", "picking_rate", "shipment_delay"} assert all(m["scale"] > 0 for m in out["metrics"].values()) assert handlers.list_calibrations(svc, "acme")["calibrations"] def test_calibrate_unknown_tenant_is_not_found(svc): with pytest.raises(ApiError) as e: handlers.calibrate(svc, "ghost", { "vertical": "logistics", "entity_type": "warehouse", "observations": _warehouse_body(), }) assert e.value.status == 404 def test_calibrate_unknown_entity_type_is_not_found(svc): handlers.register_tenant(svc, {"tenant_id": "acme", "name": "ACME"}) with pytest.raises(ApiError) as e: handlers.calibrate(svc, "acme", { "vertical": "logistics", "entity_type": "spaceship", "observations": _warehouse_body(), }) assert e.value.status == 404 def test_calibrate_rejects_malformed_observations(svc): handlers.register_tenant(svc, {"tenant_id": "acme", "name": "ACME"}) with pytest.raises(ApiError) as e: handlers.calibrate(svc, "acme", { "vertical": "logistics", "entity_type": "warehouse", "observations": [{"entity_id": "w1"}], # no day, no values }) assert e.value.code == "bad_request" # --- runs ---------------------------------------------------------------- def test_run_analysis_without_calibration_is_conflict(svc): handlers.register_tenant(svc, {"tenant_id": "acme", "name": "ACME"}) with pytest.raises(ApiError) as e: handlers.run_analysis(svc, "acme", { "vertical": "logistics", "entity_type": "warehouse", "observations": _warehouse_body(), }) assert e.value.code == "conflict" and e.value.status == 409 def test_calibrate_then_run_then_query(svc): handlers.register_tenant(svc, {"tenant_id": "acme", "name": "ACME"}) body = {"vertical": "logistics", "entity_type": "warehouse", "observations": _warehouse_body()} handlers.calibrate(svc, "acme", body) run = handlers.run_analysis(svc, "acme", body) assert run["n_issues"] >= 1 fetched = handlers.get_run(svc, run["run_id"]) assert fetched["status"] == "completed" issues = handlers.get_run_issues(svc, run["run_id"])["issues"] assert issues[0]["entity_id"] == "warehouse_north" assert handlers.get_run_decisions(svc, run["run_id"])["decisions"] def test_list_runs_paginates(svc): handlers.register_tenant(svc, {"tenant_id": "acme", "name": "ACME"}) body = {"vertical": "logistics", "entity_type": "warehouse", "observations": _warehouse_body()} handlers.calibrate(svc, "acme", body) for _ in range(3): handlers.run_analysis(svc, "acme", body) page1 = handlers.list_runs(svc, "acme", limit=2, offset=0) page2 = handlers.list_runs(svc, "acme", limit=2, offset=2) assert page1["count"] == 2 and page2["count"] == 1 ids = {r["run_id"] for r in page1["runs"]} | {r["run_id"] for r in page2["runs"]} assert len(ids) == 3 # no overlap, full coverage def test_list_runs_clamps_limit(svc): handlers.register_tenant(svc, {"tenant_id": "acme", "name": "ACME"}) out = handlers.list_runs(svc, "acme", limit=10_000) assert out["limit"] == handlers.MAX_PAGE def test_get_unknown_run_is_not_found(svc): with pytest.raises(ApiError) as e: handlers.get_run(svc, "run_does_not_exist") assert e.value.status == 404 # --- FastAPI shell smoke (skipped if fastapi/httpx absent) ---------------- def test_fastapi_app_smoke(): pytest.importorskip("fastapi") pytest.importorskip("httpx") from fastapi.testclient import TestClient from infra.api import create_app app = create_app(":memory:") client = TestClient(app) assert client.get("/health").json()["status"] == "ok" # POST /tenants is the v1 bootstrap (still open) — the response is the # newly registered tenant row. r = client.post("/tenants", json={"tenant_id": "acme", "name": "ACME"}) assert r.status_code == 201 # Everything else needs an API key. Mint one through the service # (operator path — same way the first key is bootstrapped in prod). svc = app.state.svc raw_key = svc.create_api_key("acme", name="smoke").raw headers = {"Authorization": f"Bearer {raw_key}"} assert client.get("/tenants/acme", headers=headers).json()["name"] == "ACME" # Missing the header is a 401, not a 404 — auth is the first gate. assert client.get("/tenants/acme").status_code == 401 # Cross-tenant probe gets 403 (path tenant mismatches the key's tenant). assert client.get("/tenants/ghost", headers=headers).status_code == 403 # OpenAPI schema is generated assert "/tenants/{tenant_id}/runs" in client.get("/openapi.json").json()["paths"]