From 31edc5cda986dba38edd8defc4b7770d9da4ebb7 Mon Sep 17 00:00:00 2001 From: host Date: Mon, 2 Nov 2026 03:38:25 +0000 Subject: [PATCH] code-to10 v6 A8 driver round 2: remaining subset gaps (first gate acceptance, fallbacks, signals, ledger branches, guards, confirm branches) (cherry picked from commit e3cac9f3e62f01c49fca9b7cbf80431a4d55b8ad) --- .../test_v6_a8_audit_loop_driver_gaps2.py | 363 ++++++++++++++++++ 1 file changed, 363 insertions(+) create mode 100644 cloud/tests/test_v6_a8_audit_loop_driver_gaps2.py diff --git a/cloud/tests/test_v6_a8_audit_loop_driver_gaps2.py b/cloud/tests/test_v6_a8_audit_loop_driver_gaps2.py new file mode 100644 index 0000000..d147264 --- /dev/null +++ b/cloud/tests/test_v6_a8_audit_loop_driver_gaps2.py @@ -0,0 +1,363 @@ +"""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()