Spaces:
Paused
Paused
| """Live verification of every /api/safety/* endpoint through the real app | |
| (in-process fastapi TestClient against app_backend.app — no running servers | |
| are touched).""" | |
| from __future__ import annotations | |
| import pytest | |
| from fastapi.testclient import TestClient | |
| LENIENT = { | |
| "scenario_quality": {"min_geometric_pass_rate": 0.05}, | |
| "coverage": {"min_coverage_rate": 0.0, "min_production_weighted_coverage": 0.0, | |
| "min_samples": 5, "max_ci_width": 0.99}, | |
| } | |
| def client(): | |
| from app_backend import app | |
| return TestClient(app) | |
| def api_env(mega_env, fresh_safety_root, tmp_path): | |
| """Isolated safety root + labeleval store with one dataset, plus mega runs.""" | |
| from sensorflow.evaluation import synthetic | |
| from sensorflow.evaluation.records import reset_store | |
| store = reset_store(tmp_path / "eval") | |
| ds = synthetic.generate_dataset(store, num_sequences=2, frames_per_sequence=6, | |
| seed=23) | |
| synthetic.generate_labels(store, ds) | |
| return {"mega": mega_env, "store": store, "ds": ds} | |
| def test_odd_endpoints(client, api_env): | |
| assert client.get("/api/safety/odd/taxonomy").status_code == 200 | |
| run_id = api_env["mega"]["good"].run_id | |
| r = client.get("/api/safety/odd/coverage", | |
| params={"run": run_id, "dims": "weather,lighting", | |
| "min_samples": 10_000}) | |
| assert r.status_code == 200 | |
| cov = r.json() | |
| assert cov["run_id"] == run_id and cov["gaps"] | |
| gap = cov["gaps"][0] | |
| r = client.post("/api/safety/odd/fill-gap", | |
| json={"run": run_id, "cell": gap["cell"], "num_sequences": 1, | |
| "frames_per_sequence": 10}) | |
| assert r.status_code == 200 | |
| assert r.json()["objects_added"] > 0 | |
| assert client.get("/api/safety/odd/coverage", | |
| params={"run": "eval-nope"}).status_code == 404 | |
| def test_gates_and_evidence_endpoints(client, api_env): | |
| good = api_env["mega"]["good"].run_id | |
| bad = api_env["mega"]["bad"].run_id | |
| r = client.get("/api/safety/gates/policy") | |
| assert r.status_code == 200 and "coverage" in r.json()["policy"] | |
| r = client.post("/api/safety/gates/policy", | |
| json={"overrides": {"safety": {"max_csi_increase_ratio": 0.05}}}) | |
| assert r.status_code == 200 | |
| assert r.json()["policy"]["safety"]["max_csi_increase_ratio"] == 0.05 | |
| r = client.post("/api/safety/gates/evaluate", | |
| json={"candidate_run": bad, "baseline_run": good, | |
| "policy_overrides": LENIENT}) | |
| assert r.status_code == 200 | |
| res = r.json() | |
| assert res["decision"] == "BLOCKED" | |
| assert "regression" in res["blocking_gates"] | |
| r = client.get(f"/api/safety/gates/result/{bad}") | |
| assert r.status_code == 200 and r.json()["decision"] == "BLOCKED" | |
| r = client.get(f"/api/safety/evidence/{bad}") | |
| assert r.status_code == 200 | |
| assert r.json()["package_id"] == f"sep-{bad}" | |
| r = client.get(f"/api/safety/evidence/{bad}", params={"format": "markdown"}) | |
| assert r.status_code == 200 | |
| assert "# Safety Evidence Package" in r.json()["markdown"] | |
| assert client.get("/api/safety/evidence/eval-none").status_code == 404 | |
| def test_ssam_endpoints(client, api_env): | |
| r = client.post("/api/safety/ssam/analyze", | |
| json={"scenario": "crossing", "seed": 0, "reaction_delay_s": 0.9}) | |
| assert r.status_code == 200 | |
| body = r.json() | |
| assert body["aggregate"]["num_conflicts"] >= 1 | |
| assert body["generated"]["simulated"] is True | |
| # explicit trajectories path | |
| states = [{"t": round(i * 0.1, 3), "x": 10.0 * i * 0.1, "y": 0.0, | |
| "speed": 10.0, "heading": 0.0} for i in range(21)] | |
| still = [{"t": round(i * 0.1, 3), "x": 30.0, "y": 0.0, "speed": 0.0, | |
| "heading": 0.0} for i in range(21)] | |
| r = client.post("/api/safety/ssam/analyze", json={"trajectories": [ | |
| {"vehicle_id": "A", "states": states}, | |
| {"vehicle_id": "B", "states": still}]}) | |
| assert r.status_code == 200 | |
| assert r.json()["aggregate"]["num_conflicts"] == 1 | |
| run_id = api_env["mega"]["good"].run_id | |
| r = client.get("/api/safety/ssam/summary", params={"run": run_id}) | |
| assert r.status_code == 200 | |
| assert r.json()["aggregate_csi"] > 0 | |
| def test_calibration_endpoints(client, api_env): | |
| r = client.post("/api/safety/calibration/validate", | |
| json={"mode": "miscalibrated", "rotation_offset_deg": 2.0}) | |
| assert r.status_code == 200 | |
| assert r.json()["status"] == "MISCALIBRATED" | |
| r = client.get("/api/safety/calibration/status") | |
| assert r.status_code == 200 and r.json()["status"] == "MISCALIBRATED" | |
| r = client.post("/api/safety/calibration/validate", json={"mode": "clean"}) | |
| assert r.json()["status"] == "CALIBRATED" | |
| def test_discrepancy_endpoints(client, api_env): | |
| ds_id = api_env["ds"].dataset_id | |
| r = client.post("/api/safety/discrepancy/mine", | |
| json={"dataset_id": ds_id, | |
| "profile": {"extra_miss_rate": 0.5}}) | |
| assert r.status_code == 200 | |
| report = r.json() | |
| assert report["totals"]["discrepancies"] > 0 | |
| r = client.get("/api/safety/discrepancy/summary", params={"dataset_id": ds_id}) | |
| assert r.status_code == 200 | |
| assert r.json()["dataset_id"] == ds_id | |
| r = client.get("/api/safety/discrepancy/summary") | |
| assert r.status_code == 200 and len(r.json()["datasets"]) == 1 | |
| def test_scenario_endpoints(client, api_env): | |
| ds_id = api_env["ds"].dataset_id | |
| # seed content through discrepancy mining | |
| client.post("/api/safety/discrepancy/mine", | |
| json={"dataset_id": ds_id, "profile": {"extra_miss_rate": 0.5}}) | |
| r = client.get("/api/safety/scenarios", params={"source": "discrepancy"}) | |
| assert r.status_code == 200 | |
| body = r.json() | |
| assert body["counts"]["total"] > 0 | |
| assert all(s["source"] == "discrepancy" for s in body["scenarios"]) | |
| r = client.post("/api/safety/scenarios/populate") | |
| assert r.status_code == 200 | |
| r = client.post("/api/safety/scenarios/export", json={"source": "discrepancy"}) | |
| assert r.status_code == 200 | |
| bundle = r.json() | |
| assert bundle["bundle_format"] == "sensorflow-scenario-bundle/v1" | |
| assert bundle["count"] == len(bundle["scenarios"]) > 0 | |
| def test_semantic_search_endpoint(client, api_env): | |
| r = client.post("/api/safety/semantic-search", | |
| json={"concept": "hesitant vulnerable road user near lane edge", | |
| "target": "containers", "k": 8, "use_llm": False}) | |
| assert r.status_code == 200 | |
| body = r.json() | |
| assert body["results"] | |
| assert body["stage2"]["provider"] == "offline_deterministic" | |
| assert body["results"][0]["explanations"]["stage2_reasoning"] | |
| r = client.post("/api/safety/semantic-search", | |
| json={"concept": "night rain", "target": "containers", | |
| "filters": {"lighting": "night"}, "use_llm": False}) | |
| assert all(x["lighting"] == "night" for x in r.json()["results"]) | |
| def test_consensus_endpoint(client, api_env): | |
| store, ds = api_env["store"], api_env["ds"] | |
| from sensorflow.evaluation.graders import grade_annotation | |
| frames = {f.frame_id: f for f in store.where("frames", dataset_id=ds.dataset_id)} | |
| for ann in [a for a in store.where("annotations", dataset_id=ds.dataset_id) | |
| if a.bbox_3d][:6]: | |
| grade_annotation(store, ann, frames[ann.frame_id]) | |
| r = client.get("/api/safety/consensus/summary", | |
| params={"dataset_id": ds.dataset_id, "examples": 3}) | |
| assert r.status_code == 200 | |
| body = r.json() | |
| assert "kendall_tau_confidence_vs_consensus" in body["statistics"] | |
| assert body["examples"] | |
| ex = body["examples"][0] | |
| assert "consensus_score_vector" in ex and "mbr" in ex | |
| assert client.get("/api/safety/consensus/summary", | |
| params={"dataset_id": "ds-none"}).status_code == 404 | |