host
488aedbcad
(cherry picked from commit 1db02545f74c96b3b92be4fc32954659cec1f814) Reconcile-conflict: A14b E402 imports referencing cloud.handler_core (does not exist here; core methods live in server.py) — kept HEAD imports with incoming's noqa: E402 markers, dropped the unused core_mod import (verified unused elsewhere in both files).
573 lines
26 KiB
Python
573 lines
26 KiB
Python
"""Audit loop driver: pure decision logic for the project-level
|
|
audit rework loop (track 1 of the production built-in loop).
|
|
|
|
Pins the driver interface the next (wiring) track must use
|
|
verbatim, and the counting rules shared by #244 / #241 / #247:
|
|
pushbacks cover failures 1-3, the 4th failure (> 3, never >= 3)
|
|
rebuilds, rebuilds are 1 per module per contract version, the
|
|
3rd failure is triaged, a pass clears failures but not budgets,
|
|
a contract-version change clears everything, quota blocks consume
|
|
nothing, a correction to pass does not unblock, and ledger events
|
|
match the #241 field table.
|
|
|
|
Style follows the other cloud tests: unittest, stdlib only, import
|
|
with a sys.path fallback so the file also runs standalone.
|
|
"""
|
|
|
|
import sys
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
try:
|
|
from cloud.audit_loop_driver import (
|
|
ACTION_KINDS,
|
|
LEDGER_EVENT_TYPES,
|
|
apply_correction,
|
|
confirm_rebuild,
|
|
decide_audit,
|
|
emit,
|
|
new_loop_state,
|
|
validate_ledger_event,
|
|
after_fix,
|
|
)
|
|
except ImportError:
|
|
sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
|
|
from cloud.audit_loop_driver import (
|
|
ACTION_KINDS,
|
|
LEDGER_EVENT_TYPES,
|
|
apply_correction,
|
|
confirm_rebuild,
|
|
decide_audit,
|
|
emit,
|
|
new_loop_state,
|
|
validate_ledger_event,
|
|
after_fix,
|
|
)
|
|
|
|
|
|
def fail_verdict(**overrides):
|
|
verdict = {
|
|
"source": "audit",
|
|
"verdict": "fail",
|
|
"verdict_text": "audit fail: ci red",
|
|
"failures": [{"check": "ci_green", "detail": "ci red"}],
|
|
"passed_items": [],
|
|
"previous_artifacts": [{"path": "app.py", "sha256": "a" * 64}],
|
|
"reproduction_steps": ["run ci"],
|
|
}
|
|
verdict.update(overrides)
|
|
return verdict
|
|
|
|
|
|
def pass_verdict(**overrides):
|
|
verdict = {
|
|
"source": "audit",
|
|
"verdict": "pass",
|
|
"verdict_text": "audit pass",
|
|
"failures": [],
|
|
"passed_items": [],
|
|
}
|
|
verdict.update(overrides)
|
|
return verdict
|
|
|
|
|
|
def near_complete_verdict(**overrides):
|
|
verdict = fail_verdict(
|
|
failures=[{"check": "edge-case"}],
|
|
passed_items=[{"check": "p1"}, {"check": "p2"}, {"check": "p3"}],
|
|
broken_passed_items=[],
|
|
verdict_text="audit fail: one edge case left",
|
|
)
|
|
verdict.update(overrides)
|
|
return verdict
|
|
|
|
|
|
def drive(state, verdict, **kwargs):
|
|
return decide_audit(state, verdict, **kwargs)
|
|
|
|
|
|
class PushbackCountingTestCase(unittest.TestCase):
|
|
def test_fail_pushbacks_and_blocks(self):
|
|
state = new_loop_state("m1", "v1")
|
|
action = drive(state, fail_verdict())
|
|
self.assertEqual(action.kind, "pushback")
|
|
self.assertTrue(action.merge_blocked)
|
|
self.assertTrue(action.release_blocked)
|
|
self.assertEqual(action.next_state.failure_count, 1)
|
|
self.assertEqual(action.next_state.pushback_round, 1)
|
|
self.assertEqual(action.next_state.status, "pushback")
|
|
package = action.pushback_package
|
|
self.assertEqual(package["audit_verdict_text"], "audit fail: ci red")
|
|
self.assertEqual(package["failures"][0]["check"], "ci_green")
|
|
self.assertEqual(package["reproduction_steps"], ["run ci"])
|
|
# Full re-audit criteria, never just the failed items.
|
|
self.assertIn("ci_green", package["reaudit_criteria"])
|
|
self.assertIn("audit_signer_separation", package["reaudit_criteria"])
|
|
self.assertEqual(package["previous_artifacts"][0]["path"], "app.py")
|
|
self.assertEqual(
|
|
[e["event_type"] for e in action.ledger_events],
|
|
["audit_verdict", "audit_pushback_round"])
|
|
|
|
def test_exactly_three_still_pushes_back(self):
|
|
state = new_loop_state("m1", "v1")
|
|
for _ in range(3):
|
|
action = drive(state, fail_verdict())
|
|
state = action.next_state
|
|
self.assertEqual(action.kind, "pushback")
|
|
self.assertEqual(action.next_state.failure_count, 3)
|
|
self.assertEqual(action.next_state.pushback_round, 3)
|
|
|
|
def test_fourth_failure_rebuilds(self):
|
|
state = new_loop_state("m1", "v1")
|
|
for _ in range(3):
|
|
state = drive(state, fail_verdict()).next_state
|
|
action = drive(state, fail_verdict(retained_tests=["t1", "t2"]))
|
|
self.assertEqual(action.kind, "rebuild")
|
|
self.assertEqual(action.rebuild_no, 1)
|
|
self.assertEqual(action.next_state.rebuild_no, 1)
|
|
self.assertEqual(action.next_state.failure_count, 0)
|
|
self.assertEqual(action.next_state.pushback_round, 0)
|
|
self.assertEqual(action.next_state.status, "rebuilding")
|
|
self.assertEqual(action.next_state.retained_tests, ["t1", "t2"])
|
|
retained = action.pushback_package["retained_assets"]
|
|
self.assertEqual(retained["tests"], ["t1", "t2"])
|
|
self.assertTrue(retained["contract_unchanged"])
|
|
self.assertTrue(retained["design_unchanged"])
|
|
|
|
def test_rebuild_budget_exhausted_escalates(self):
|
|
state = new_loop_state("m1", "v1")
|
|
for _ in range(3):
|
|
state = drive(state, fail_verdict()).next_state
|
|
state = drive(state, fail_verdict()).next_state # rebuild 1
|
|
self.assertEqual(state.rebuild_no, 1)
|
|
for _ in range(3):
|
|
state = drive(state, fail_verdict()).next_state
|
|
action = drive(state, fail_verdict())
|
|
self.assertEqual(action.kind, "escalate")
|
|
self.assertEqual(action.escalation_target, "coordinator")
|
|
self.assertEqual(action.next_state.rebuild_no, 1)
|
|
self.assertEqual(action.next_state.status, "escalated")
|
|
self.assertFalse(action.escalation_grants_user_acceptance)
|
|
self.assertFalse(hasattr(action, "tick_acceptance"))
|
|
# Full history, not just the last round.
|
|
texts = [entry.get("verdict_text")
|
|
for entry in action.escalation_history]
|
|
self.assertIn("audit fail: ci red", texts)
|
|
self.assertGreaterEqual(len(action.escalation_history), 4)
|
|
|
|
def test_audit_and_acceptance_counts_combine_but_stay_separate(self):
|
|
state = new_loop_state("m1", "v1")
|
|
state = drive(state, fail_verdict()).next_state
|
|
state = drive(state, fail_verdict(source="acceptance")).next_state
|
|
action = drive(state, fail_verdict())
|
|
self.assertEqual(action.next_state.failure_count, 3)
|
|
self.assertEqual(action.next_state.audit_fail_count, 2)
|
|
self.assertEqual(action.next_state.acceptance_fail_count, 1)
|
|
|
|
def test_pass_clears_failures_keeps_rebuild_budget(self):
|
|
state = new_loop_state("m1", "v1")
|
|
for _ in range(3):
|
|
state = drive(state, fail_verdict()).next_state
|
|
state = drive(state, fail_verdict()).next_state # rebuild 1
|
|
state = drive(state, fail_verdict()).next_state # failure 1, new episode
|
|
action = drive(state, pass_verdict())
|
|
self.assertEqual(action.kind, "pass")
|
|
self.assertFalse(action.merge_blocked)
|
|
self.assertEqual(action.next_state.failure_count, 0)
|
|
self.assertEqual(action.next_state.pushback_round, 0)
|
|
self.assertEqual(action.next_state.rebuild_no, 1)
|
|
self.assertEqual(action.next_state.status, "passed")
|
|
|
|
def test_contract_version_change_clears_everything(self):
|
|
state = new_loop_state("m1", "v1")
|
|
for _ in range(3):
|
|
state = drive(state, fail_verdict()).next_state
|
|
state = drive(state, fail_verdict()).next_state # rebuild 1
|
|
state = drive(state, fail_verdict()).next_state # failure 1, v1
|
|
action = drive(state, fail_verdict(contract_version="v2"))
|
|
self.assertEqual(action.kind, "pushback")
|
|
self.assertEqual(action.next_state.contract_version, "v2")
|
|
self.assertEqual(action.next_state.failure_count, 1)
|
|
self.assertEqual(action.next_state.pushback_round, 1)
|
|
self.assertEqual(action.next_state.rebuild_no, 0)
|
|
self.assertFalse(action.next_state.targeted_used)
|
|
|
|
|
|
class ThirdFailureTriageTestCase(unittest.TestCase):
|
|
def _two_failures(self):
|
|
state = new_loop_state("m1", "v1")
|
|
for _ in range(2):
|
|
state = drive(state, fail_verdict()).next_state
|
|
return state
|
|
|
|
def test_structural_third_failure_early_rebuilds(self):
|
|
state = self._two_failures()
|
|
action = drive(state, fail_verdict(
|
|
failures=[{"check": "arch",
|
|
"structural_kind": "architecture_boundary"}],
|
|
verdict_text="audit fail: architecture boundary crossed"))
|
|
self.assertEqual(action.kind, "rebuild")
|
|
self.assertEqual(action.rebuild_no, 1)
|
|
assessments = [e for e in action.ledger_events
|
|
if e["event_type"] == "audit_third_failure_assessment"]
|
|
self.assertEqual(len(assessments), 1)
|
|
self.assertEqual(assessments[0]["payload"]["determination"], "early_rebuild")
|
|
self.assertEqual(assessments[0]["payload"]["failure_total"], 3)
|
|
self.assertEqual(assessments[0]["payload"]["max_extra_rounds"], 0)
|
|
|
|
def test_contract_root_third_failure_goes_to_refreeze(self):
|
|
state = self._two_failures()
|
|
action = drive(state, fail_verdict(root_cause_hint="contract"))
|
|
self.assertEqual(action.kind, "refreeze")
|
|
self.assertEqual(action.next_state.status, "refreeze")
|
|
self.assertEqual(action.next_state.rebuild_no, 0)
|
|
self.assertEqual(action.compat_hint, "audit_wiring_event:escalate_refreeze")
|
|
# No rebuild event, and no assessment row forced into the
|
|
# wrong determination (see #247: early_refreeze has no
|
|
# assessment determination).
|
|
kinds = [e["event_type"] for e in action.ledger_events]
|
|
self.assertNotIn("audit_rebuild", kinds)
|
|
self.assertNotIn("audit_third_failure_assessment", kinds)
|
|
|
|
def test_near_complete_gets_one_targeted_round_then_rebuilds(self):
|
|
state = self._two_failures()
|
|
action = drive(state, near_complete_verdict())
|
|
self.assertEqual(action.kind, "pushback")
|
|
self.assertEqual(action.scope, "targeted")
|
|
self.assertEqual(action.targeted_items, ["edge-case"])
|
|
self.assertTrue(action.next_state.targeted_used)
|
|
self.assertEqual(action.next_state.pushback_round, 3)
|
|
assessments = [e for e in action.ledger_events
|
|
if e["event_type"] == "audit_third_failure_assessment"]
|
|
self.assertEqual(assessments[0]["payload"]["determination"],
|
|
"continuation_round")
|
|
self.assertEqual(assessments[0]["payload"]["max_extra_rounds"], 1)
|
|
# One more failure is the 4th: rebuild, not a 4th pushback.
|
|
action = drive(action.next_state, fail_verdict())
|
|
self.assertEqual(action.kind, "rebuild")
|
|
|
|
def test_targeted_budget_not_replenished_by_pass(self):
|
|
state = self._two_failures()
|
|
state = drive(state, near_complete_verdict()).next_state
|
|
self.assertTrue(state.targeted_used)
|
|
state = drive(state, pass_verdict()).next_state
|
|
self.assertTrue(state.targeted_used)
|
|
for _ in range(2):
|
|
state = drive(state, fail_verdict()).next_state
|
|
action = drive(state, near_complete_verdict())
|
|
self.assertEqual(action.kind, "pushback")
|
|
self.assertEqual(action.scope, "normal")
|
|
self.assertIsNone(action.targeted_items)
|
|
|
|
def test_unknown_broken_passed_disqualifies_near_complete(self):
|
|
state = self._two_failures()
|
|
# broken_passed_items absent (unknown, not []) -> normal.
|
|
action = drive(state, fail_verdict(
|
|
failures=[{"check": "edge-case"}],
|
|
passed_items=[{"check": "p1"}, {"check": "p2"}, {"check": "p3"}]))
|
|
self.assertEqual(action.scope, "normal")
|
|
|
|
|
|
class RebuildGateAndAcceptanceTestCase(unittest.TestCase):
|
|
def _rebuilding_state(self):
|
|
state = new_loop_state("m1", "v1")
|
|
for _ in range(3):
|
|
state = drive(state, fail_verdict()).next_state
|
|
return drive(state, fail_verdict(retained_tests=["t1", "t2"])).next_state
|
|
|
|
def test_old_test_gate_missing_result_counts_as_failure(self):
|
|
state = self._rebuilding_state()
|
|
action = after_fix(state, {"reaudit_verdict": pass_verdict()})
|
|
self.assertEqual(action.kind, "pushback")
|
|
self.assertEqual(action.next_state.failure_count, 1)
|
|
self.assertEqual(action.next_state.status, "pushback")
|
|
checks = [item.get("check") for item in action.pushback_package["failures"]]
|
|
self.assertIn("rebuild_first_gate", checks)
|
|
|
|
def test_old_test_gate_failing_result_counts_as_failure(self):
|
|
state = self._rebuilding_state()
|
|
action = after_fix(state, {
|
|
"reaudit_verdict": pass_verdict(),
|
|
"retained_test_results": {"t1": True, "t2": False},
|
|
})
|
|
self.assertEqual(action.kind, "pushback")
|
|
self.assertEqual(action.next_state.failure_count, 1)
|
|
|
|
def test_old_test_gate_passing_allows_reaudit_pass(self):
|
|
state = self._rebuilding_state()
|
|
action = after_fix(state, {
|
|
"reaudit_verdict": pass_verdict(),
|
|
"retained_test_results": {"t1": True, "t2": True},
|
|
})
|
|
self.assertEqual(action.kind, "pass")
|
|
self.assertEqual(action.next_state.status, "passed")
|
|
|
|
def _acceptance_pending(self):
|
|
state = new_loop_state("m1", "v1")
|
|
for _ in range(3):
|
|
state = drive(state, fail_verdict(source="acceptance")).next_state
|
|
return drive(state, fail_verdict(source="acceptance"))
|
|
|
|
def test_acceptance_rebuild_unconfirmed_does_not_start(self):
|
|
action = self._acceptance_pending()
|
|
self.assertEqual(action.kind, "pending_user_confirmation")
|
|
self.assertTrue(action.next_state.pending_confirmation)
|
|
self.assertEqual(action.next_state.rebuild_no, 0)
|
|
self.assertEqual(action.next_state.status, "pending_user_confirmation")
|
|
notices = [e for e in action.ledger_events
|
|
if e["event_type"] == "audit_rebuild_notice"]
|
|
self.assertEqual(notices[0]["payload"]["confirmation_status"], "pending")
|
|
|
|
def test_acceptance_rebuild_confirmed_then_starts(self):
|
|
pending = self._acceptance_pending().next_state
|
|
action = confirm_rebuild(pending, user_confirmed=True)
|
|
self.assertEqual(action.kind, "rebuild")
|
|
self.assertEqual(action.rebuild_no, 1)
|
|
self.assertEqual(action.next_state.status, "rebuilding")
|
|
self.assertFalse(action.next_state.pending_confirmation)
|
|
notices = [e for e in action.ledger_events
|
|
if e["event_type"] == "audit_rebuild_notice"]
|
|
self.assertEqual(notices[0]["payload"]["confirmation_status"], "confirmed")
|
|
self.assertEqual(notices[0]["payload"]["confirmed_by"], "user")
|
|
|
|
def test_acceptance_rebuild_rejected_escalates_without_budget(self):
|
|
pending = self._acceptance_pending().next_state
|
|
action = confirm_rebuild(pending, user_confirmed=False)
|
|
self.assertEqual(action.kind, "escalate")
|
|
self.assertEqual(action.escalation_target, "coordinator")
|
|
self.assertEqual(action.next_state.rebuild_no, 0)
|
|
|
|
def test_confirm_without_pending_raises(self):
|
|
state = new_loop_state("m1", "v1")
|
|
with self.assertRaises(ValueError):
|
|
confirm_rebuild(state, user_confirmed=True)
|
|
|
|
|
|
class AfterFixTestCase(unittest.TestCase):
|
|
def test_after_fix_pass_clears(self):
|
|
state = new_loop_state("m1", "v1")
|
|
state = drive(state, fail_verdict()).next_state
|
|
action = after_fix(state, {"reaudit_verdict": pass_verdict()})
|
|
self.assertEqual(action.kind, "pass")
|
|
self.assertEqual(action.next_state.failure_count, 0)
|
|
|
|
def test_after_fix_failing_reaudit_goes_to_next_pushback(self):
|
|
state = new_loop_state("m1", "v1")
|
|
state = drive(state, fail_verdict()).next_state
|
|
action = after_fix(state, {"reaudit_verdict": fail_verdict()})
|
|
self.assertEqual(action.kind, "pushback")
|
|
self.assertEqual(action.next_state.failure_count, 2)
|
|
self.assertEqual(action.next_state.pushback_round, 2)
|
|
|
|
def test_after_fix_diff_only_raises(self):
|
|
state = new_loop_state("m1", "v1")
|
|
state = drive(state, fail_verdict()).next_state
|
|
with self.assertRaises(ValueError):
|
|
after_fix(state, {"diff": "--- a/x\n+++ b/x\n"})
|
|
with self.assertRaises(ValueError):
|
|
after_fix(state, {"diff_only": True,
|
|
"reaudit_verdict": pass_verdict()})
|
|
|
|
|
|
class CorrectionTestCase(unittest.TestCase):
|
|
def test_correction_to_pass_does_not_unblock(self):
|
|
state = new_loop_state("m1", "v1")
|
|
for _ in range(2):
|
|
state = drive(state, fail_verdict()).next_state
|
|
action = apply_correction(state, {
|
|
"original_verdict": "fail", "corrected_verdict": "pass",
|
|
"reviewer": "auditor-b", "original_entry_id": "e1",
|
|
"reason": "second look: evidence was sufficient",
|
|
})
|
|
# Counting view: effective count is 0 ...
|
|
self.assertEqual(action.next_state.failure_count, 0)
|
|
# ... but the gate view is still blocked until a fresh pass.
|
|
self.assertTrue(action.merge_blocked)
|
|
self.assertTrue(action.release_blocked)
|
|
self.assertEqual(action.next_state.status, "blocked")
|
|
self.assertTrue(action.requires_reaudit)
|
|
self.assertEqual(action.correction_action, "schedule_reaudit")
|
|
self.assertEqual(action.ledger_events[0]["event_type"], "audit_meta_review")
|
|
# Only a fresh passing verdict unblocks.
|
|
action = drive(action.next_state, pass_verdict())
|
|
self.assertEqual(action.kind, "pass")
|
|
self.assertFalse(action.merge_blocked)
|
|
|
|
def test_correction_to_fail_reblocks_at_once(self):
|
|
state = drive(new_loop_state("m1", "v1"), pass_verdict()).next_state
|
|
action = apply_correction(state, {
|
|
"original_verdict": "pass", "corrected_verdict": "fail",
|
|
"reviewer": "auditor-b", "original_entry_id": "e1",
|
|
"reason": "missed a failing check",
|
|
})
|
|
self.assertEqual(action.kind, "pushback")
|
|
self.assertTrue(action.merge_blocked)
|
|
self.assertEqual(action.next_state.failure_count, 1)
|
|
self.assertEqual(action.correction_action, "retrigger_pushback_and_reblock")
|
|
|
|
|
|
class QuotaAndCostTestCase(unittest.TestCase):
|
|
def test_quota_block_consumes_nothing(self):
|
|
state = new_loop_state("m1", "v1")
|
|
for _ in range(2):
|
|
state = drive(state, fail_verdict()).next_state
|
|
action = drive(state, fail_verdict(),
|
|
quota_check=lambda estimate: False)
|
|
self.assertEqual(action.kind, "quota_blocked")
|
|
self.assertIs(action.next_state, state)
|
|
self.assertEqual(action.next_state.failure_count, 2)
|
|
self.assertEqual(action.next_state.pushback_round, 2)
|
|
|
|
def test_quota_block_at_rebuild_consumes_no_rebuild(self):
|
|
state = new_loop_state("m1", "v1")
|
|
for _ in range(3):
|
|
state = drive(state, fail_verdict()).next_state
|
|
action = drive(state, fail_verdict(),
|
|
quota_check=lambda estimate: False)
|
|
self.assertEqual(action.kind, "quota_blocked")
|
|
self.assertEqual(action.next_state.rebuild_no, 0)
|
|
self.assertEqual(action.next_state.failure_count, 3)
|
|
|
|
def test_quota_check_exception_fails_closed(self):
|
|
state = new_loop_state("m1", "v1")
|
|
def _boom(estimate):
|
|
raise RuntimeError("quota service down")
|
|
action = drive(state, fail_verdict(), quota_check=_boom)
|
|
self.assertEqual(action.kind, "quota_blocked")
|
|
self.assertEqual(action.next_state.failure_count, 0)
|
|
|
|
def test_cost_threshold_escalates_rebuild_to_user(self):
|
|
state = new_loop_state("m1", "v1")
|
|
for _ in range(3):
|
|
state = drive(state, fail_verdict()).next_state
|
|
action = drive(state, fail_verdict(cost_estimate=500),
|
|
cost_threshold=100)
|
|
self.assertEqual(action.kind, "escalate")
|
|
self.assertEqual(action.escalation_target, "user")
|
|
self.assertEqual(action.next_state.rebuild_no, 0)
|
|
self.assertEqual(action.cost_estimate, 500)
|
|
self.assertGreaterEqual(len(action.escalation_history), 4)
|
|
self.assertFalse(action.escalation_grants_user_acceptance)
|
|
|
|
def test_cost_threshold_not_applied_to_pushback(self):
|
|
# #244 scope preserved: the threshold gates rebuild steps,
|
|
# a fix pushback is quota-checked only.
|
|
state = new_loop_state("m1", "v1")
|
|
action = drive(state, fail_verdict(cost_estimate=500),
|
|
cost_threshold=100)
|
|
self.assertEqual(action.kind, "pushback")
|
|
|
|
|
|
class TestChangeTestCase(unittest.TestCase):
|
|
def test_test_deletion_violation_does_not_advance(self):
|
|
state = new_loop_state("m1", "v1")
|
|
state = drive(state, fail_verdict()).next_state
|
|
action = drive(state, fail_verdict(test_changes=[
|
|
{"action": "delete", "test_id": "t1"}]))
|
|
self.assertEqual(action.kind, "escalate")
|
|
self.assertTrue(action.violations)
|
|
self.assertEqual(action.next_state.failure_count, 1)
|
|
self.assertEqual(action.next_state.pushback_round, 1)
|
|
|
|
def test_test_deletion_allowed_with_refreeze_and_audit(self):
|
|
state = new_loop_state("m1", "v1")
|
|
action = drive(state, fail_verdict(test_changes=[
|
|
{"action": "delete", "test_id": "t1",
|
|
"separate_audit_id": "audit-9"}],
|
|
refreeze_approved=True, separate_audit_ids=["audit-9"]))
|
|
self.assertEqual(action.kind, "pushback")
|
|
self.assertEqual(action.violations, [])
|
|
|
|
|
|
class LedgerEventTestCase(unittest.TestCase):
|
|
def test_all_seven_event_types_validate(self):
|
|
cases = {
|
|
"audit_verdict": {"module_id": "m1", "contract_version": "v1",
|
|
"verdict": "pass", "source": "audit", "failures": []},
|
|
"audit_pushback_round": {"module_id": "m1", "contract_version": "v1",
|
|
"round_no": 1, "trigger_items": ["ci_green"],
|
|
"pushback_package": {}, "round_status": "pushed"},
|
|
"audit_rebuild": {"module_id": "m1", "contract_version": "v1",
|
|
"rebuild_no": 1, "root_cause": "implementation",
|
|
"status": "decided", "retained_assets": {}},
|
|
"audit_third_failure_assessment": {
|
|
"module_id": "m1", "contract_version": "v1",
|
|
"determination": "continuation_round", "basis": {},
|
|
"failure_total": 3, "max_extra_rounds": 1},
|
|
"audit_test_change": {"module_id": "m1", "contract_version": "v1",
|
|
"test_changes": {"added": ["t9"], "modified": [],
|
|
"deleted": []},
|
|
"auditor": "auditor-a",
|
|
"contract_refrozen": False, "reason": "add only"},
|
|
"audit_meta_review": {"original_entry_id": "e1",
|
|
"original_verdict": "fail",
|
|
"corrected_verdict": "pass",
|
|
"misjudgment_found": True,
|
|
"reviewer": "auditor-b", "findings": [],
|
|
"triggered_process": "re-review"},
|
|
"audit_rebuild_notice": {"module_id": "m1", "contract_version": "v1",
|
|
"rebuild_no": 1,
|
|
"notification_content": "notice text",
|
|
"notified_party": "user",
|
|
"confirmation_status": "pending"},
|
|
}
|
|
for event_type, payload in cases.items():
|
|
self.assertEqual(validate_ledger_event(event_type, payload), [],
|
|
event_type)
|
|
event = emit(event_type, payload)
|
|
self.assertEqual(event["event_type"], event_type)
|
|
self.assertEqual(set(LEDGER_EVENT_TYPES), set(cases))
|
|
|
|
def test_unknown_event_type_and_missing_field_rejected(self):
|
|
self.assertTrue(validate_ledger_event("nope", {}))
|
|
with self.assertRaises(ValueError):
|
|
emit("audit_verdict", {"module_id": "m1"})
|
|
with self.assertRaises(ValueError):
|
|
emit("audit_verdict", {"module_id": "m1", "contract_version": "v1",
|
|
"verdict": "pass", "source": "audit"})
|
|
|
|
def test_conditional_maps_to_fail_and_keeps_original(self):
|
|
event = emit("audit_verdict", {
|
|
"module_id": "m1", "contract_version": "v1",
|
|
"verdict": "conditional", "source": "audit", "failures": []})
|
|
self.assertEqual(event["payload"]["verdict"], "fail")
|
|
self.assertEqual(event["payload"]["extra"]["original_verdict"],
|
|
"conditional")
|
|
# A driver verdict of conditional is a fail, not a pass.
|
|
action = drive(new_loop_state("m1", "v1"),
|
|
fail_verdict(verdict="conditional"))
|
|
self.assertEqual(action.kind, "pushback")
|
|
|
|
def test_test_change_hard_gate_in_emit(self):
|
|
payload = {"module_id": "m1", "contract_version": "v1",
|
|
"test_changes": {"added": [], "modified": ["t1"],
|
|
"deleted": []},
|
|
"auditor": "auditor-a", "contract_refrozen": False,
|
|
"reason": "drop a test"}
|
|
with self.assertRaises(ValueError):
|
|
emit("audit_test_change", payload)
|
|
payload["contract_refrozen"] = True
|
|
payload["refreeze_ref"] = "freeze-2"
|
|
emit("audit_test_change", payload) # no raise
|
|
|
|
def test_rebuild_notice_confirmed_needs_confirmer(self):
|
|
payload = {"module_id": "m1", "contract_version": "v1",
|
|
"rebuild_no": 1, "notification_content": "notice",
|
|
"notified_party": "user", "confirmation_status": "confirmed"}
|
|
with self.assertRaises(ValueError):
|
|
emit("audit_rebuild_notice", payload)
|
|
payload["confirmed_by"] = "user"
|
|
emit("audit_rebuild_notice", payload) # no raise
|
|
|
|
def test_action_kinds_are_bounded(self):
|
|
for kind in ACTION_KINDS:
|
|
self.assertIn(kind, ("pass", "pushback", "rebuild", "refreeze",
|
|
"escalate", "pending_user_confirmation",
|
|
"quota_blocked"))
|
|
action = drive(new_loop_state("m1", "v1"), fail_verdict())
|
|
self.assertIn(action.kind, ACTION_KINDS)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|