Spaces:
Paused
Paused
File size: 7,185 Bytes
0523c2e | 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 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 | """API lifecycle via TestClient: create -> diagnostics -> stage completion ->
scoreboard -> decision tree -> experiments -> report -> reveal."""
from __future__ import annotations
def _create(client, **kw):
body = {"cause": "FEATURE_SKEW", "seed": 7}
body.update(kw)
res = client.post("/api/rca/investigations", json=body)
assert res.status_code == 200, res.text
return res.json()
def test_meta_routes(rca_client):
causes = rca_client.get("/api/rca/causes").json()["causes"]
assert len(causes) == 8
stages = rca_client.get("/api/rca/stages").json()["stages"]
assert len(stages) == 13
def test_create_and_list(rca_client):
inv = _create(rca_client)
assert inv["scenario_cause"] == "FEATURE_SKEW"
assert inv["claims"]["offline_delta_pp"] > 4.0
assert inv["claims"]["shadow_delta_pp"] < -1.0
listed = rca_client.get("/api/rca/investigations").json()["investigations"]
assert any(r["id"] == inv["id"] for r in listed)
def test_training_mode_hides_cause_until_reveal(rca_client):
inv = _create(rca_client, training_mode=True, cause=None, seed=123)
assert "scenario_cause" not in inv
got = rca_client.get(f"/api/rca/investigations/{inv['id']}").json()
assert "scenario_cause" not in got
reveal = rca_client.post(f"/api/rca/investigations/{inv['id']}/reveal").json()
assert reveal["cause"] in {
"TRUE_MODEL_REGRESSION", "DISTRIBUTION_SHIFT", "FEATURE_SKEW",
"SERVING_MISMATCH", "LABEL_LATENCY", "SAMPLING_BIAS",
"STATISTICAL_NOISE", "OFFLINE_CONTAMINATION"}
assert reveal["explanation"]
got = rca_client.get(f"/api/rca/investigations/{inv['id']}").json()
assert got["scenario_cause"] == reveal["cause"]
def test_all_diagnostic_endpoints(rca_client):
inv = _create(rca_client)
for key in ("comparison_validity", "offline_audit", "population_validation",
"distribution_shift", "conditional_performance",
"paired_comparison", "statistical_significance",
"feature_parity", "serving_parity", "shadow_traffic",
"label_integrity"):
res = rca_client.get(
f"/api/rca/investigations/{inv['id']}/diagnostics/{key}")
assert res.status_code == 200, f"{key}: {res.text}"
body = res.json()
assert body["stage"] == key
assert body["data"]
assert isinstance(body["findings"], list) and body["findings"]
res = rca_client.get(
f"/api/rca/investigations/{inv['id']}/diagnostics/nope")
assert res.status_code == 404
def test_stage_ordering_enforced_via_api(rca_client):
inv = _create(rca_client)
res = rca_client.post(
f"/api/rca/investigations/{inv['id']}/stages/3/complete", json={})
assert res.status_code == 409
res = rca_client.post(
f"/api/rca/investigations/{inv['id']}/stages/0/complete", json={})
assert res.status_code == 200
def test_unknown_ack_flow_via_api(rca_client):
inv = _create(rca_client) # FEATURE_SKEW -> CV_UNKNOWN:feature_pipeline_version
iid = inv["id"]
rca_client.get(f"/api/rca/investigations/{iid}/diagnostics/comparison_validity")
res = rca_client.post(
f"/api/rca/investigations/{iid}/stages/0/complete", json={})
assert res.status_code == 409
assert "UNKNOWN" in res.json()["detail"]
res = rca_client.post(
f"/api/rca/investigations/{iid}/stages/0/complete",
json={"acknowledge_unknowns": True, "note": "pipeline team paged"})
assert res.status_code == 200
assert res.json()["stage"]["status"] == "complete_with_unknowns"
def test_record_finding_and_scoreboard(rca_client):
inv = _create(rca_client)
iid = inv["id"]
res = rca_client.post(f"/api/rca/investigations/{iid}/findings", json={
"stage": "feature_parity", "title": "Confirmed with pipeline team",
"status": "MISMATCH", "severity": "CRITICAL",
"detail": "fp-2.5 deployed online only",
"code": "FP_FEATURE_SKEW:human_confirmed"})
assert res.status_code == 200
board = rca_client.get(f"/api/rca/investigations/{iid}/scoreboard").json()
assert board["rows"][0]["hypothesis"] == "FEATURE_SKEW"
# The human finding contributes to the score via the same rule table.
top = board["rows"][0]
assert any(e["finding_id"] and e["code"] == "FP_FEATURE_SKEW:human_confirmed"
for e in top["evidence_for"])
res = rca_client.post(f"/api/rca/investigations/{iid}/scoreboard/assess",
json={"hypothesis": "FEATURE_SKEW",
"confidence": "HIGH", "note": "confirmed"})
assert res.status_code == 200
row = next(r for r in res.json()["rows"] if r["hypothesis"] == "FEATURE_SKEW")
assert row["human_confidence"] == "HIGH"
assert row["auto_confidence"] # both assessments are kept
def test_tree_experiments_report(rca_client):
inv = _create(rca_client)
iid = inv["id"]
tree = rca_client.get(f"/api/rca/investigations/{iid}/decision-tree").json()
assert tree["conclusion"] == "FEATURE_SKEW"
assert tree["path"]
exp = rca_client.get(f"/api/rca/investigations/{iid}/experiments").json()
assert exp["minimum_additional_evidence"]
rep = rca_client.get(f"/api/rca/investigations/{iid}/report").json()
assert rep["executive_finding"]["conclusion"] == "FEATURE_SKEW"
assert rep["markdown"].startswith("# RCA Report")
assert len(rep["hypothesis_ranking"]) == 8
assert len(rep["stage_summaries"]) == 11
assert rep["remediation"]["containment"]
def test_persistence_roundtrip(rca_client):
"""Data survives the CSV round trip: fresh store caches cleared, then the
scoreboard still ranks correctly."""
from sensorflow.rca import store
from sensorflow.rca import api as rca_api
inv = _create(rca_client)
store._cache.clear()
store._data_cache.clear()
rca_api._battery_cache.clear()
board = rca_client.get(
f"/api/rca/investigations/{inv['id']}/scoreboard").json()
assert board["rows"][0]["hypothesis"] == "FEATURE_SKEW"
def test_recorded_only_scoreboard_evolves(rca_client):
"""The working-hypothesis banner must not leak unvisited stages: with no
diagnostics run, all 8 hypotheses stay in play; evidence narrows it."""
inv = _create(rca_client)
iid = inv["id"]
board = rca_client.get(
f"/api/rca/investigations/{iid}/scoreboard?recorded_only=true").json()
assert len(board["working_hypothesis_set"]) == 8
for key in ("feature_parity", "offline_audit", "shadow_traffic",
"label_integrity", "serving_parity"):
rca_client.get(f"/api/rca/investigations/{iid}/diagnostics/{key}")
board = rca_client.get(
f"/api/rca/investigations/{iid}/scoreboard?recorded_only=true").json()
ws = board["working_hypothesis_set"]
assert "FEATURE_SKEW" in ws
assert "OFFLINE_CONTAMINATION" not in ws # ruled out by clean audit
assert len(ws) < 8
def test_404s(rca_client):
assert rca_client.get("/api/rca/investigations/nope").status_code == 404
assert rca_client.get(
"/api/rca/investigations/nope/scoreboard").status_code == 404
|