"""Release gate composition on planted inputs shaped like the real engines' outputs. The hard governance rules live here.""" from __future__ import annotations import pytest from sensorflow.studio2.release_gate import GO, NO_GO, REVIEW, ReleaseGate def make_gate(registry): return ReleaseGate(registry) def test_all_pass_is_go_pending_approval(registry, passing_safety, passing_seqeval, clean_shift): gate = make_gate(registry) d = gate.evaluate(passing_safety, passing_seqeval, clean_shift, agentic_outcome={"outcome": "CONTINUE_INVESTIGATION", "policy_version": "p", "failure_id": "f", "severity": "S1"}, closed_loop={"scenario_id": "s", "verdict": "METRIC_ONLY"}, hardware_matrix={"status": "PASS", "matrix_id": "m", "n_combinations": 27, "insufficient": []}) assert d["status"] == GO assert d["blocking_conditions"] == [] # GO is a recommendation, never a deployment assert d["human_approval_required"] is True assert d["deployment_authorized"] is False assert d["evidence_completeness"] == 1.0 def test_regression_verdict_forces_no_go(registry, passing_safety, regressed_seqeval, clean_shift): d = make_gate(registry).evaluate(passing_safety, regressed_seqeval, clean_shift) assert d["status"] == NO_GO assert any("sequential regression confirmed" in c for c in d["blocking_conditions"]) assert "pedestrian|night" in " ".join(d["blocking_conditions"]) def test_blocked_safety_forces_no_go(registry, blocked_safety, passing_seqeval, clean_shift): d = make_gate(registry).evaluate(blocked_safety, passing_seqeval, clean_shift) assert d["status"] == NO_GO assert any("safety gate blocked: safety" in c for c in d["blocking_conditions"]) def test_missing_subsystem_degrades_to_review_with_named_gap( registry, passing_safety, passing_seqeval): """A missing required input can never produce a silent GO.""" d = make_gate(registry).evaluate(passing_safety, passing_seqeval, None) assert d["status"] == REVIEW assert any("distribution shift report unavailable" in g for g in d["degraded_inputs"]) assert d["evidence_completeness"] < 1.0 def test_all_missing_is_review_never_go(registry): d = make_gate(registry).evaluate(None, None, None) assert d["status"] == REVIEW assert len(d["degraded_inputs"]) == 3 assert d["evidence_completeness"] == 0.0 assert d["confidence"] <= 0.3 def test_optional_subsystem_absence_is_surfaced_not_silent( registry, passing_safety, passing_seqeval, clean_shift): """agentic / closed loop / hardware unavailable -> named unresolved question and no GO (graceful degradation, never a quiet pass).""" d = make_gate(registry).evaluate(passing_safety, passing_seqeval, clean_shift) assert d["status"] == REVIEW joined = " ".join(d["unresolved_questions"]) assert "agentic policy outcome unavailable" in joined assert "closed-loop verdict unavailable" in joined assert "hardware gate matrix unavailable" in joined def test_inconclusive_seqeval_maps_to_review(registry, passing_safety, clean_shift): inconclusive = {"run_id": "seq-x", "decision": "INSUFFICIENT_EVIDENCE", "stopping_reason": "escalation_exhausted", "samples_used": 9000, "affected_strata": []} d = make_gate(registry).evaluate(passing_safety, inconclusive, clean_shift) assert d["status"] == REVIEW assert any("INSUFFICIENT_EVIDENCE" in q for q in d["unresolved_questions"]) def test_shift_with_recall_impact_forces_review(registry, passing_safety, passing_seqeval, bad_shift): d = make_gate(registry).evaluate(passing_safety, passing_seqeval, bad_shift) assert d["status"] == REVIEW assert any("RCA" in q for q in d["unresolved_questions"]) def test_agentic_stop_ship_forces_no_go(registry, passing_safety, passing_seqeval, clean_shift): d = make_gate(registry).evaluate( passing_safety, passing_seqeval, clean_shift, agentic_outcome={"outcome": "AUTOMATIC_STOP_SHIP", "severity": "S4", "policy_version": "p", "failure_id": "f"}) assert d["status"] == NO_GO def test_decision_records_full_evidence_tuple_and_policy_version( registry, passing_safety, passing_seqeval, clean_shift): d = make_gate(registry).evaluate(passing_safety, passing_seqeval, clean_shift, context={"candidate": "model-v9"}) t = d["evidence_tuple"] assert t["release_policy_version"] assert t["safety_gates"]["evidence_package_id"] == "sep-eval-test" assert t["sequential_regression"]["seqeval_run_id"] == "seq-test" assert t["distribution_shift"]["shift_run_id"] == "eval-test" assert t["context"] == {"candidate": "model-v9"} # policy is content-hash versioned and registered pols = registry.list("policies") assert any(p["policy_version"] == d["policy_version"] for p in pols) # decision persisted assert registry.get("decisions", d["entity_id"])["status"] == d["status"] # ------------------------------------------------------------------ approval def _go_decision(registry, passing_safety, passing_seqeval, clean_shift): return make_gate(registry).evaluate( passing_safety, passing_seqeval, clean_shift, agentic_outcome={"outcome": "CONTINUE_INVESTIGATION", "policy_version": "p", "failure_id": "f", "severity": "S1"}, closed_loop={"scenario_id": "s", "verdict": "METRIC_ONLY"}, hardware_matrix={"status": "PASS", "matrix_id": "m", "n_combinations": 27, "insufficient": []}) def test_go_is_not_deployment_until_separately_approved( registry, passing_safety, passing_seqeval, clean_shift): gate = make_gate(registry) d = _go_decision(registry, passing_safety, passing_seqeval, clean_shift) assert d["status"] == GO and d["deployment_authorized"] is False approved = gate.approve(d["entity_id"], "warda", "reviewed evidence package") assert approved["deployment_authorized"] is True assert approved["approval"]["approver"] == "warda" # the approval is a first-class recorded entity assert len(registry.list("approvals")) == 1 def test_only_go_decisions_can_be_approved(registry, passing_safety, regressed_seqeval, clean_shift): gate = make_gate(registry) d = gate.evaluate(passing_safety, regressed_seqeval, clean_shift) assert d["status"] == NO_GO with pytest.raises(ValueError): gate.approve(d["entity_id"], "warda", "trying anyway") def test_approval_requires_named_approver_and_rationale( registry, passing_safety, passing_seqeval, clean_shift): gate = make_gate(registry) d = _go_decision(registry, passing_safety, passing_seqeval, clean_shift) with pytest.raises(ValueError): gate.approve(d["entity_id"], " ", "reason") with pytest.raises(ValueError): gate.approve(d["entity_id"], "warda", "") def test_hardware_critical_failure_blocks_despite_everything_else_passing( registry, passing_safety, passing_seqeval, clean_shift): matrix = {"status": "FAIL_CRITICAL", "matrix_id": "m", "n_combinations": 27, "critical_failures": [ {"combination_label": "san_francisco × versal-ai-edge × LiDAR-Gen2", "failed_checks": ["recall 0.71 < 0.8"]}], "insufficient": []} d = make_gate(registry).evaluate(passing_safety, passing_seqeval, clean_shift, hardware_matrix=matrix) assert d["status"] == NO_GO assert any("critical hardware combination" in c for c in d["blocking_conditions"])