tests/test_rca/test_api.py

"""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