""" Tests for Stage 66 — GET /tenants/{tid}/backfill + SDK method. HTTP exposure of Stage 65's point-in-time analysis. Same auth pattern as preview (Stage 45) and same read-only contract (no DB writes). The SDK method roundtrips through the route. """ import pytest pytest.importorskip("fastapi") pytest.importorskip("httpx") pytest.importorskip("yaml") from fastapi.testclient import TestClient from infra import OrgStateService from infra.api import create_app from orgstate_client import Client, ClientError def _seed_calibrated(dbfile, tenant_id="acme"): """Register + ingest 30 days of warehouse observations + calibrate. Gives backfill 24 eligible cutoffs (30 - 7 baseline + 1 = ~24 depending on how the math falls).""" from verticals import get_vertical_config, get_vertical_observation_loader svc = OrgStateService(dbfile) try: svc.register_tenant(tenant_id, tenant_id, vertical="logistics") cfg = get_vertical_config("logistics").entity_type("warehouse") obs = get_vertical_observation_loader( "logistics")()["warehouse"] svc.ingest_observations(tenant_id, "warehouse", [ {"entity_id": o.entity_id, "day": o.day, "values": o.values} for o in obs ]) svc.calibrate_and_store(tenant_id, obs, cfg, vertical="logistics") finally: svc.close() def _bootstrap(tmp_path): dbfile = str(tmp_path / "bf_http.sqlite3") _seed_calibrated(dbfile, "acme") _seed_calibrated(dbfile, "globex") svc = OrgStateService(dbfile) try: keys = { "acme_ro": svc.create_api_key("acme", role="readonly").raw, "globex_ro": svc.create_api_key("globex", role="readonly").raw, "admin": svc.create_admin_key().raw, } finally: svc.close() return dbfile, keys def _client(dbfile): return TestClient(create_app(dbfile)) def _auth(k): return {"Authorization": f"Bearer {k}"} # --- HTTP route ----------------------------------------------------- def test_returns_results_array(tmp_path): dbfile, keys = _bootstrap(tmp_path) client = _client(dbfile) r = client.get( "/tenants/acme/backfill?entity_type=warehouse", headers=_auth(keys["acme_ro"]), ) assert r.status_code == 200 body = r.json() assert body["tenant_id"] == "acme" assert body["entity_type"] == "warehouse" assert body["vertical"] == "logistics" assert "results" in body assert body["n_points"] == len(body["results"]) def test_each_result_has_summary_fields(tmp_path): dbfile, keys = _bootstrap(tmp_path) client = _client(dbfile) body = client.get( "/tenants/acme/backfill?entity_type=warehouse", headers=_auth(keys["acme_ro"]), ).json() for r in body["results"]: assert {"as_of_day", "n_states", "n_issues", "n_decisions", "top_severity", "top_issues"} <= r.keys() def test_from_until_filters_narrow_range(tmp_path): """The route accepts `from`/`until` as the URL keys (Python keywords avoided via alias).""" dbfile, keys = _bootstrap(tmp_path) client = _client(dbfile) body = client.get( "/tenants/acme/backfill?entity_type=warehouse" "&from=2026-02-01&until=2026-02-15", headers=_auth(keys["acme_ro"]), ).json() for r in body["results"]: assert "2026-02-01" <= r["as_of_day"] <= "2026-02-15" assert body["from_day"] == "2026-02-01" assert body["until_day"] == "2026-02-15" def test_step_days_skips_cutoffs(tmp_path): """step_days=7 → far fewer cutoffs than the default daily.""" dbfile, keys = _bootstrap(tmp_path) client = _client(dbfile) daily = client.get( "/tenants/acme/backfill?entity_type=warehouse", headers=_auth(keys["acme_ro"]), ).json() weekly = client.get( "/tenants/acme/backfill?entity_type=warehouse&step_days=7", headers=_auth(keys["acme_ro"]), ).json() assert weekly["n_points"] < daily["n_points"] assert weekly["step_days"] == 7 # --- THE central contract: no DB writes via HTTP -------------------- def test_http_backfill_does_not_write_runs(tmp_path): """The contract from Stage 65 carries through the route.""" dbfile, keys = _bootstrap(tmp_path) svc = OrgStateService(dbfile) try: before = svc.db.query_one( "SELECT COUNT(*) AS n FROM runs" )["n"] finally: svc.close() client = _client(dbfile) r = client.get( "/tenants/acme/backfill?entity_type=warehouse", headers=_auth(keys["acme_ro"]), ) assert r.status_code == 200 svc = OrgStateService(dbfile) try: after = svc.db.query_one( "SELECT COUNT(*) AS n FROM runs" )["n"] finally: svc.close() assert before == after # --- auth ------------------------------------------------------------ def test_no_key_401(tmp_path): dbfile, _ = _bootstrap(tmp_path) client = _client(dbfile) r = client.get("/tenants/acme/backfill?entity_type=warehouse") assert r.status_code == 401 def test_cross_tenant_403(tmp_path): dbfile, keys = _bootstrap(tmp_path) client = _client(dbfile) r = client.get( "/tenants/globex/backfill?entity_type=warehouse", headers=_auth(keys["acme_ro"]), ) assert r.status_code == 403 def test_admin_can_backfill_any_tenant(tmp_path): dbfile, keys = _bootstrap(tmp_path) client = _client(dbfile) r = client.get( "/tenants/globex/backfill?entity_type=warehouse", headers=_auth(keys["admin"]), ) assert r.status_code == 200 # --- error paths --------------------------------------------------- def test_missing_entity_type_returns_422(tmp_path): dbfile, keys = _bootstrap(tmp_path) client = _client(dbfile) r = client.get("/tenants/acme/backfill", headers=_auth(keys["acme_ro"])) assert r.status_code == 422 def test_unknown_entity_type_returns_404(tmp_path): dbfile, keys = _bootstrap(tmp_path) client = _client(dbfile) r = client.get( "/tenants/acme/backfill?entity_type=ghost", headers=_auth(keys["acme_ro"]), ) assert r.status_code == 404 # --- SDK ----------------------------------------------------------- def test_sdk_backfill_roundtrips(tmp_path): """SDK Client.backfill() returns same payload as the HTTP route.""" dbfile, keys = _bootstrap(tmp_path) c = Client(base_url="http://test", tenant_id="acme", api_key=keys["acme_ro"], transport=TestClient(create_app(dbfile))) out = c.backfill("warehouse") assert out["tenant_id"] == "acme" assert out["entity_type"] == "warehouse" assert "results" in out assert out["n_points"] == len(out["results"]) def test_sdk_backfill_passes_date_filters(tmp_path): dbfile, keys = _bootstrap(tmp_path) c = Client(base_url="http://test", tenant_id="acme", api_key=keys["acme_ro"], transport=TestClient(create_app(dbfile))) out = c.backfill( "warehouse", from_day="2026-02-01", until_day="2026-02-15", ) for r in out["results"]: assert "2026-02-01" <= r["as_of_day"] <= "2026-02-15" def test_sdk_backfill_step_days(tmp_path): """SDK passes step_days to the route — weekly sampling reduces the result count.""" dbfile, keys = _bootstrap(tmp_path) c = Client(base_url="http://test", tenant_id="acme", api_key=keys["acme_ro"], transport=TestClient(create_app(dbfile))) daily = c.backfill("warehouse") weekly = c.backfill("warehouse", step_days=7) assert weekly["n_points"] < daily["n_points"] def test_sdk_backfill_unknown_entity_type_raises(tmp_path): dbfile, keys = _bootstrap(tmp_path) c = Client(base_url="http://test", tenant_id="acme", api_key=keys["acme_ro"], transport=TestClient(create_app(dbfile))) with pytest.raises(ClientError) as ei: c.backfill("ghost") assert ei.value.status_code == 404