"""code-to10 v6 A8 第二轮：audit_loop_driver 剩余子集缺口补测。 本文件补驱动相关测试子集（test_audit_loop_driver + test_v6_a8_audit_loop_driver_gaps + wiring + server_audit_loop） 覆盖率实测后仍缺的 59 行中可达的部分；不可达的两行单列 在文件尾注释与交回报告里，不写成通过： - validate_ledger_event:792：changed 非空时 (added or changed) 恒真，该错误分支逻辑上不可达。 - apply_correction:1393：original/corrected 经映射后恒为 pass/fail，该校验分支逻辑上不可达。 只加测试，不改生产代码；依赖缺失用 sys.modules 注入 ImportError 模拟（不 mock 被测函数本身）。 """ import sys import unittest from contextlib import contextmanager from pathlib import Path try: import cloud.audit_loop_driver as driver from cloud.audit_loop_driver import ( _check_first_gate, _map_conditional, _root_cause_signals, _triage_third_failure, after_fix, apply_correction, confirm_rebuild, decide_audit, emit, new_loop_state, validate_ledger_event, ) except ImportError: # discover -s cloud/tests 时把项目根目录补进 sys.path sys.path.insert(0, str(Path(__file__).resolve().parents[2])) import cloud.audit_loop_driver as driver from cloud.audit_loop_driver import ( _check_first_gate, _map_conditional, _root_cause_signals, _triage_third_failure, after_fix, apply_correction, confirm_rebuild, decide_audit, emit, new_loop_state, validate_ledger_event, ) @contextmanager def _missing_pushback_module(): """让 audit_pushback 的 import 在函数内抛 ImportError， 走本地镜像回落分支；用完恢复原模块对象。""" key = "cloud.audit_pushback" had = key in sys.modules old = sys.modules.get(key) sys.modules[key] = None try: yield finally: if had: sys.modules[key] = old else: sys.modules.pop(key, None) def _fail_verdict(**overrides): verdict = { "source": "audit", "verdict": "fail", "verdict_text": "audit fail", "failures": [{"check": "ci_green", "detail": "ci red"}], "passed_items": [], } 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 class FirstGateAcceptanceTestCase(unittest.TestCase): def test_missing_test_and_missing_and_failed_acceptance(self): gate = _check_first_gate( ["t1", "t2"], [{"id": "a1"}, "a2"], {"t1": True}, {"a1": False}) self.assertFalse(gate["gate_pass"]) self.assertEqual(gate["missing"], ["test:t2", "acceptance:a2"]) self.assertEqual(gate["failed"], ["acceptance:a1"]) class DependencyFallbackTestCase(unittest.TestCase): def test_validate_test_changes_falls_back_to_local(self): with _missing_pushback_module(): violations = driver._validate_test_changes( [{"action": "delete", "test_id": "t1"}], False, []) self.assertEqual(len(violations), 2) def test_classify_root_cause_falls_back_to_local(self): rounds = [[{"root_cause": "contract"}], [{"root_cause": "contract"}]] with _missing_pushback_module(): self.assertEqual(driver._classify_root_cause(rounds), "contract") class RootCauseSignalsTestCase(unittest.TestCase): def test_classified_contract_rounds_signal(self): rounds = [[{"root_cause": "contract"}], [{"root_cause": "contract"}]] contract, impl = _root_cause_signals(_fail_verdict(), [], rounds) self.assertTrue(contract) self.assertFalse(impl) def test_structural_kind_and_core_touch_signal(self): failures = [{"check": "x", "structural_kind": "tech_stack"}, {"check": "y", "touches_core_structure": True}] contract, impl = _root_cause_signals(_fail_verdict(), failures, []) self.assertFalse(contract) self.assertTrue(impl) def test_repeated_clause_in_two_rounds_signals_contract(self): failures = [{"check": "x", "root_cause": "contract", "evidence": {"contract_clause": "C1"}}] rounds = [[{"evidence": {"contract_clause": "C1"}}], [{"evidence": {"contract_clause": "C1"}}]] contract, impl = _root_cause_signals(_fail_verdict(), failures, rounds) self.assertTrue(contract) class TriageCapReasonTestCase(unittest.TestCase): def test_too_many_failures_lists_cap_reason(self): state = new_loop_state("m1", "v1") failures = [{"check": "f%d" % i} for i in range(5)] triage = _triage_third_failure( state, _fail_verdict(broken_passed_items=None), failures, [{"check": "p1"}], []) self.assertEqual(triage["verdict"], "normal_pushback") self.assertTrue(any("exceeds near-complete cap" in r for r in triage["reasons"])) class LedgerValidationBranchesTestCase(unittest.TestCase): def test_non_dict_payload(self): self.assertEqual(validate_ledger_event("audit_verdict", "x"), ["ledger payload must be a dict"]) def test_enum_violation(self): errors = validate_ledger_event("audit_verdict", { "module_id": "m1", "contract_version": "v1", "verdict": "bogus", "source": "audit", "failures": []}) self.assertTrue(any("invalid verdict" in e for e in errors)) def test_rebuild_no_below_one(self): errors = validate_ledger_event("audit_rebuild", { "module_id": "m1", "contract_version": "v1", "rebuild_no": 0, "root_cause": "implementation", "status": "decided", "retained_assets": {}}) self.assertIn("rebuild_no must be >= 1", errors) def test_test_change_names_nothing(self): errors = validate_ledger_event("audit_test_change", { "module_id": "m1", "contract_version": "v1", "test_changes": {"added": [], "modified": [], "deleted": []}, "auditor": "a", "contract_refrozen": False, "reason": "none"}) self.assertIn("test_changes must name at least one change", errors) def test_refreeze_without_ref(self): errors = validate_ledger_event("audit_test_change", { "module_id": "m1", "contract_version": "v1", "test_changes": {"added": ["t9"], "modified": [], "deleted": []}, "auditor": "a", "contract_refrozen": True, "reason": "ok"}) self.assertIn("refreeze_ref required when contract_refrozen", errors) def test_meta_review_misjudgment_pairing(self): base = {"original_entry_id": "e1", "original_verdict": "fail", "reviewer": "b", "findings": [], "triggered_process": "re-review"} errors = validate_ledger_event("audit_meta_review", dict( base, misjudgment_found=True, corrected_verdict=None)) self.assertIn("corrected_verdict required when misjudgment_found", errors) errors = validate_ledger_event("audit_meta_review", dict( base, misjudgment_found=False, corrected_verdict="pass")) self.assertIn("corrected_verdict must be None without a misjudgment", errors) def test_rebuild_notice_empty_content(self): errors = validate_ledger_event("audit_rebuild_notice", { "module_id": "m1", "contract_version": "v1", "rebuild_no": 1, "notification_content": "", "notified_party": "user", "confirmation_status": "pending"}) self.assertIn("notification_content must be non-empty", errors) class ConditionalReviewAndSinkTestCase(unittest.TestCase): def test_re_review_result_mapped_to_fail(self): mapped = _map_conditional({ "verdict": "pass", "re_review_result": {"verdict": "conditional", "note": "x"}, }) self.assertEqual(mapped["re_review_result"]["verdict"], "fail") self.assertEqual( mapped["re_review_result"]["original_verdict"], "conditional") def test_emit_calls_sink(self): seen = [] event = emit("audit_verdict", { "module_id": "m1", "contract_version": "v1", "verdict": "pass", "source": "audit", "failures": []}, sink=seen.append) self.assertEqual(seen, [event]) class DecideAuditGuardsTestCase(unittest.TestCase): def test_non_loop_state_and_non_dict_verdict(self): with self.assertRaises(TypeError): decide_audit("not-a-state", _pass_verdict()) with self.assertRaises(TypeError): decide_audit(new_loop_state("m1", "v1"), "not-a-dict") def test_unknown_source(self): with self.assertRaises(ValueError): decide_audit(new_loop_state("m1", "v1"), _fail_verdict(source="mystery")) def test_qa_failed_links_not_rerun_adds_failure(self): state = new_loop_state("m1", "v1") action = decide_audit(state, _fail_verdict( qa_failed_links=["https://example.invalid/broken"], qa_links_rerun_passed=False)) checks = [f.get("check") for f in action.pushback_package["failures"]] self.assertIn("qa_failed_links_rerun", checks) def test_pass_with_version_change_resets(self): state = new_loop_state("m1", "v1") for _ in range(2): state = decide_audit(state, _fail_verdict()).next_state action = decide_audit(state, _pass_verdict(contract_version="v2")) self.assertEqual(action.kind, "pass") self.assertEqual(action.next_state.contract_version, "v2") self.assertEqual(action.next_state.failure_count, 0) def test_rebuilding_gate_bool_form(self): state = new_loop_state("m1", "v1") for _ in range(3): state = decide_audit(state, _fail_verdict()).next_state rebuilding = decide_audit( state, _fail_verdict(retained_tests=["t1"])).next_state self.assertEqual(rebuilding.status, "rebuilding") action = decide_audit(rebuilding, _pass_verdict( retained_tests_passed=True)) self.assertEqual(action.kind, "pass") # bool 门表单为 False 时门失败，pass 被门失败项挡下。 action = decide_audit(rebuilding, _pass_verdict( retained_tests_passed=False)) self.assertEqual(action.kind, "pushback") checks = [f.get("check") for f in action.pushback_package["failures"]] self.assertIn("rebuild_first_gate", checks) def test_fourth_failure_with_contract_signal_refreezes(self): state = new_loop_state("m1", "v1") for _ in range(3): state = decide_audit(state, _fail_verdict()).next_state action = decide_audit(state, _fail_verdict( root_cause_hint="contract")) self.assertEqual(action.kind, "refreeze") class AfterFixAndCorrectionGuardsTestCase(unittest.TestCase): def test_after_fix_non_dict_and_non_dict_reaudit(self): state = new_loop_state("m1", "v1") with self.assertRaises(TypeError): after_fix(state, "not-a-dict") with self.assertRaises(TypeError): after_fix(state, {"reaudit_verdict": 123}) def test_correction_non_dict(self): with self.assertRaises(TypeError): apply_correction(new_loop_state("m1", "v1"), "not-a-dict") def test_correction_same_verdict_rejected(self): with self.assertRaises(ValueError): apply_correction(new_loop_state("m1", "v1"), { "original_verdict": "fail", "corrected_verdict": "fail", "reviewer": "b", "original_entry_id": "e1", "reason": "same"}) def test_correction_without_reviewer(self): with self.assertRaises(ValueError): apply_correction(new_loop_state("m1", "v1"), { "original_verdict": "fail", "corrected_verdict": "pass", "original_entry_id": "e1", "reason": "no reviewer"}) def test_correction_unknown_source(self): with self.assertRaises(ValueError): apply_correction(new_loop_state("m1", "v1"), { "original_verdict": "fail", "corrected_verdict": "pass", "reviewer": "b", "original_entry_id": "e1", "reason": "x", "source": "mystery"}) class ConfirmRebuildBranchesTestCase(unittest.TestCase): def _pending_state(self): state = new_loop_state("m1", "v1") for _ in range(3): state = decide_audit( state, _fail_verdict(source="acceptance")).next_state action = decide_audit(state, _fail_verdict(source="acceptance")) self.assertEqual(action.kind, "pending_user_confirmation") return action.next_state def test_confirm_non_loop_state(self): with self.assertRaises(TypeError): confirm_rebuild("not-a-state", user_confirmed=True) def test_confirm_quota_blocked_keeps_pending(self): pending = self._pending_state() action = confirm_rebuild(pending, user_confirmed=True, quota_check=lambda estimate: False) self.assertEqual(action.kind, "quota_blocked") self.assertIs(action.next_state, pending) self.assertTrue(action.next_state.pending_confirmation) def test_confirm_budget_exhausted_escalates(self): pending = self._pending_state() pending.rebuild_no = 1 # MAX_REBUILDS_PER_CONTRACT_VERSION action = confirm_rebuild(pending, user_confirmed=True) self.assertEqual(action.kind, "escalate") self.assertEqual(action.escalation_target, "coordinator") self.assertEqual(action.next_state.rebuild_no, 1) def test_confirm_cost_threshold_escalates_to_user(self): pending = self._pending_state() action = confirm_rebuild(pending, user_confirmed=True, cost_threshold=100, cost_estimate=500) self.assertEqual(action.kind, "escalate") self.assertEqual(action.escalation_target, "user") self.assertEqual(action.cost_estimate, 500) if __name__ == "__main__": unittest.main()