Spaces:
Paused
Paused
File size: 4,312 Bytes
953f0c4 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 | """REST surface: run lifecycle, evidence, attribution, policy, trajectories."""
from __future__ import annotations
import pytest
from fastapi import FastAPI
from fastapi.testclient import TestClient
from sensorflow.seqeval.api import router
from tests.test_seqeval.conftest import FAST_POLICY
@pytest.fixture(scope="module")
def client(seq_env):
app = FastAPI()
app.include_router(router)
return TestClient(app)
@pytest.fixture(scope="module")
def finished_run(seq_env, client):
body = {
"population_id": seq_env["meta"]["population_id"],
"baseline": {"model_version": "api-base-v1"},
"candidate": {"model_version": "api-cand-v1",
"effects": {"pedestrian|night": -0.03}},
"policy": {**FAST_POLICY, "safety_floor": 1500},
"sync": True,
}
r = client.post("/api/seqeval/runs", json=body)
assert r.status_code == 200, r.text
return r.json()
def test_unknown_population_404(client):
r = client.post("/api/seqeval/runs", json={
"population_id": "pop-nope",
"baseline": {"model_version": "a"},
"candidate": {"model_version": "b"}})
assert r.status_code == 404
def test_run_state_shape(client, finished_run):
rid = finished_run["run_id"]
r = client.get(f"/api/seqeval/runs/{rid}")
assert r.status_code == 200
state = r.json()
assert state["status"] == "done"
assert state["decision"] in ("REGRESSION", "PASS", "INSUFFICIENT_EVIDENCE")
assert state["gate"] in ("block", "allow", "expand_or_report")
assert state["budget"]["samples_used"] > 0
assert state["budget"]["planned_total"] > 0
assert state["plan"]["plan_hash"]
assert state["sanity"]["ok"] is True
# per-node decision table for the dashboard
keys = {n["node"] for n in state["nodes"]}
assert "overall" in keys
assert any(k.startswith("stratum:") for k in keys)
node = next(n for n in state["nodes"] if n["node"] == "overall")
assert "delta_ci" in node and "e_regression" in node and "decision" in node
def test_trajectories_include_boundaries(client, finished_run):
rid = finished_run["run_id"]
state = client.get(f"/api/seqeval/runs/{rid}").json()
traj = state["trajectories"]
assert "stratum:pedestrian|night" in traj
t = traj["stratum:pedestrian|night"]
assert len(t["points"]) >= 2
p = t["points"][-1]
for f in ("n", "log_e_regression", "log_e_pass", "delta_lower", "delta_upper"):
assert f in p
assert t["boundaries"]["log_e_regression"] > 0 # the stopping boundary to plot
# evidence must be increasing in samples for a real regression
assert t["points"][-1]["n"] >= t["points"][0]["n"]
def test_evidence_endpoint(client, finished_run):
rid = finished_run["run_id"]
r = client.get(f"/api/seqeval/runs/{rid}/evidence")
assert r.status_code == 200
body = r.json()
assert body["records"], "expected ledger records"
assert body["lineage"]["plan_hash"]
assert set(body["required_fields"]).issubset(set(body["records"][0].keys()))
def test_attribution_endpoint(client, finished_run):
rid = finished_run["run_id"]
r = client.get(f"/api/seqeval/runs/{rid}/attribution")
assert r.status_code == 200
body = r.json()
assert body["decision"] == "REGRESSION"
assert "stratum:pedestrian|night" in body["affected_strata"]
row = body["regression_map"][0]
for f in ("baseline_value", "candidate_value", "abs_delta", "rel_delta",
"delta_ci", "n", "n_effective", "decision"):
assert f in row
def test_runs_listing(client, finished_run):
r = client.get("/api/seqeval/runs")
assert r.status_code == 200
assert any(run["run_id"] == finished_run["run_id"] for run in r.json()["runs"])
def test_policy_endpoint(client):
r = client.get("/api/seqeval/policy")
assert r.status_code == 200
body = r.json()
assert body["default_policy"]["delta_margin"] == 0.005
assert body["default_policy"]["alpha"] == 0.05
assert "REGRESSION" in body["decision_semantics"]
assert "anytime-valid" in body["test_method"]
def test_404_on_unknown_run(client):
assert client.get("/api/seqeval/runs/seq-nope").status_code == 404
assert client.get("/api/seqeval/runs/seq-nope/evidence").status_code == 404
|