Spaces:
Paused
Paused
File size: 6,376 Bytes
4b5fa98 | 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 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 | """API lifecycle over an isolated router: bank -> run -> query ->
attribution -> consequence -> HITL -> flywheel -> gate."""
from __future__ import annotations
import pytest
from fastapi import FastAPI
from fastapi.testclient import TestClient
from sensorflow.rotr.api import router
SPEC_QUERY = ("failed to yield to pedestrian at uncontrolled intersection "
"during low visibility")
@pytest.fixture()
def client():
app = FastAPI()
app.include_router(router)
return TestClient(app)
@pytest.fixture()
def run(client):
r = client.post("/api/rotr/runs",
json={"n_scenarios": 28, "seed": 7,
"model_version": "stack-v1"})
assert r.status_code == 200, r.text
return r.json()
class TestLifecycle:
def test_health_and_rules(self, client):
assert client.get("/api/rotr/health").json()["status"] == "ok"
rules = client.get("/api/rotr/rules").json()
assert len(rules["rules"]) == 6
assert "ILLUSTRATIVE" in rules["config_note"]
def test_bank_endpoints(self, client):
r = client.post("/api/rotr/banks",
json={"n_scenarios": 14, "seed": 3,
"model_version": "stack-v1"})
assert r.status_code == 200
bank = r.json()
assert client.get(f"/api/rotr/banks/{bank['bank_id']}").status_code == 200
sid = bank["scenarios"][0]["scenario_id"]
sc = client.get(f"/api/rotr/banks/{bank['bank_id']}/scenarios/{sid}").json()
assert sc["ego"] and sc["actual_context"]
assert client.post("/api/rotr/banks",
json={"model_version": "nope"}).status_code == 422
def test_run_and_metrics(self, client, run):
assert run["metrics"]["rotr_recall"] == 1.0
assert run["metrics"]["false_accusation_rate"] == 0.0
assert run["gate"]["outcome"] == "NO_GO"
got = client.get(f"/api/rotr/runs/{run['run_id']}").json()
assert got["run_id"] == run["run_id"]
assert client.get("/api/rotr/runs").json()
def test_violation_listing_and_taxonomy_filters(self, client, run):
rid = run["run_id"]
allv = client.get(f"/api/rotr/runs/{rid}/violations").json()
assert allv["n_results"] > 0
vru = client.get(f"/api/rotr/runs/{rid}/violations",
params={"vulnerability": "VRU"}).json()
assert 0 < vru["n_results"] < allv["n_results"]
assert all(x["taxonomy"]["vulnerability"] == "VRU"
for x in vru["results"])
def test_spec_structured_query(self, client, run):
r = client.post("/api/rotr/query",
json={"run_id": run["run_id"], "text": SPEC_QUERY})
assert r.status_code == 200
out = r.json()
assert out["query"]["actor"] == "pedestrian"
assert out["query"]["visibility"] == "low"
assert out["n_results"] >= 1
for item in out["results"]:
assert item["taxonomy"]["legality"] == "YIELD"
assert item["environment"]["visibility"] == "low"
assert item["cluster_id"]
assert item["provenance"]["model_version"] == "stack-v1"
def test_attribution_matrix_and_consequence(self, client, run):
rid = run["run_id"]
matrix = client.get(f"/api/rotr/runs/{rid}/attribution").json()
assert matrix["rows"]
row = matrix["rows"][0]
assert set(row["layers"]) == {
"perception", "prediction", "planning", "localization", "map",
"control", "policy_rule", "data_label"}
vid = row["violation_id"]
detail = client.get(f"/api/rotr/runs/{rid}/violations/{vid}").json()
assert detail["attribution"]["primary_layer"]
cf = client.get(
f"/api/rotr/runs/{rid}/violations/{vid}/consequence").json()
assert cf["planner_evaluation"]["observed_trajectory"]
assert cf["planner_evaluation"]["corrected_trajectory"]
assert "scenario_geometry" in cf
def test_clusters_and_gate(self, client, run):
rid = run["run_id"]
clusters = client.get(f"/api/rotr/runs/{rid}/clusters").json()
assert clusters and clusters[0]["count"] >= 1
gate = client.get(f"/api/rotr/runs/{rid}/gate").json()
assert gate["outcome"] in ("GO", "NO_GO")
policy = client.get("/api/rotr/stopship/policy").json()
assert policy["policy_version"]
class TestHITLAndFlywheel:
def test_validate_then_guard_then_override(self, client, run):
rid = run["run_id"]
queue = client.get(f"/api/rotr/runs/{rid}/hitl").json()
assert queue and queue[0]["status"] == "PENDING"
review = queue[0]
r = client.post("/api/rotr/hitl/action",
json={"run_id": rid, "review_id": review["review_id"],
"action": "VALIDATE", "actor": "reviewer",
"notes": "confirmed"})
assert r.status_code == 200
cand = r.json()["candidate"]
assert cand["dataset_role"] == "REGRESSION"
suite = client.get("/api/rotr/flywheel/suite").json()
assert any(m["candidate_id"] == cand["candidate_id"]
for m in suite["members"])
promote = client.post("/api/rotr/flywheel/promote",
json={"candidate_id": cand["candidate_id"],
"actor": "sneaky"})
assert promote.status_code == 409 # contamination guard
override = client.post("/api/rotr/flywheel/override",
json={"candidate_id": cand["candidate_id"],
"actor": "safety-lead",
"reason": "governed exception"})
assert override.status_code == 200
promote2 = client.post("/api/rotr/flywheel/promote",
json={"candidate_id": cand["candidate_id"],
"actor": "lead"})
assert promote2.status_code == 200
assert promote2.json()["guard_state"] == "OVERRIDDEN"
class TestDocs:
def test_architecture_doc_served(self, client):
r = client.get("/api/rotr/docs/architecture")
assert r.status_code == 200
assert "Right-of-the-Road" in r.json()["markdown"]
|