Spaces:
Paused
Paused
File size: 8,160 Bytes
d4d5551 | 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 169 170 171 172 173 174 175 176 177 178 | """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"])
|