Spaces:
Paused
Paused
File size: 3,513 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 | """Stage state machine: enforced ordering + unknown acknowledgment."""
from __future__ import annotations
import pytest
from sensorflow.rca.models import (Investigation, StageOrderError,
UnknownsNotAcknowledgedError, make_finding,
STAGES)
def _inv() -> Investigation:
return Investigation.new("t", "a", "b", "STATISTICAL_NOISE", seed=1)
def test_initial_state():
inv = _inv()
assert len(inv.stages) == 13
assert inv.stages[0].status == "in_progress"
assert all(s.status == "pending" for s in inv.stages[1:])
def test_cannot_skip_ahead():
inv = _inv()
with pytest.raises(StageOrderError):
inv.complete_stage(5)
# Even the immediately-next stage is blocked until stage 0 completes.
with pytest.raises(StageOrderError):
inv.complete_stage(1)
def test_ordered_completion_advances():
inv = _inv()
inv.complete_stage(0)
assert inv.stages[0].status == "complete"
assert inv.stages[1].status == "in_progress"
inv.complete_stage(1)
assert inv.stages[2].status == "in_progress"
def test_critical_unknowns_block_without_ack():
inv = _inv()
inv.upsert_auto_findings("comparison_validity", [
make_finding("comparison_validity", "CV_UNKNOWN:x", "x unknown",
"UNKNOWN", "CRITICAL")])
with pytest.raises(UnknownsNotAcknowledgedError):
inv.complete_stage(0)
stage = inv.complete_stage(0, acknowledge_unknowns=True,
ack_note="shipping anyway, tracked in JIRA-123")
assert stage.status == "complete_with_unknowns"
assert stage.acknowledged_unknowns
assert stage.ack_note == "shipping anyway, tracked in JIRA-123"
# The acknowledgment is recorded in the event log ("proceeding with
# unknowns" is auditable).
kinds = [e["kind"] for e in inv.events]
assert "unknowns_acknowledged" in kinds
def test_non_critical_unknowns_do_not_block():
inv = _inv()
inv.upsert_auto_findings("comparison_validity", [
make_finding("comparison_validity", "CV_X", "warn unknown",
"UNKNOWN", "WARN")])
stage = inv.complete_stage(0)
assert stage.status == "complete"
def test_reopen():
inv = _inv()
inv.complete_stage(0)
inv.reopen_stage(0)
assert inv.stages[0].status == "in_progress"
with pytest.raises(StageOrderError):
inv.complete_stage(1)
def test_serialization_roundtrip():
inv = _inv()
inv.upsert_auto_findings("comparison_validity", [
make_finding("comparison_validity", "CV_UNKNOWN:x", "x", "UNKNOWN",
"CRITICAL")])
inv.complete_stage(0, acknowledge_unknowns=True)
d = inv.to_json_dict()
inv2 = Investigation.from_json_dict(d)
assert inv2.stages[0].status == "complete_with_unknowns"
assert inv2.findings[0].code == "CV_UNKNOWN:x"
assert inv2.scenario_cause == "STATISTICAL_NOISE"
def test_training_mode_hides_cause():
inv = Investigation.new("t", "a", "b", "FEATURE_SKEW", seed=1,
training_mode=True)
assert "scenario_cause" not in inv.to_dict()
inv.revealed = True
assert inv.to_dict()["scenario_cause"] == "FEATURE_SKEW"
def test_stage_keys_match_spec_order():
keys = [s["key"] for s in STAGES]
assert keys[0] == "comparison_validity"
assert keys[6] == "statistical_significance"
assert keys[11] == "root_cause_scoring"
assert keys[12] == "recommendations_report"
|