"""v7 机制第 6 项：生产证据 fail-closed live 接线定向测试。 只测这一项，不碰其它 7 项，不改阈值（阈值仍 test_default_unapproved，见 cloud/audit_scorecard.py， 本文件不测阈值）、不改 #238 auto_apply、不改 #247 “纠正为 pass 本身不解封”、不改 v1.1 契约。 被测对象（cloud/gate_verification.py）： - LiveEvidenceReader 协议：get_checks(repo, ref) / get_branch_protection(repo, branch)，返回结构含状态 与原始证据；生产代码里绝无真实 HTTP 调用、真实 token 读取、真实 GitHub API 调用（本文件以 AST 断 言钉住这一点）。 - verify_live_evidence(claimed, reader, repo, ref)：无 reader、无 repo/ref、reader 失败、reader 返回与声称 不一致时 verified 一律 False；声称 CI 全绿但假 reader 返回失败/缺失时记 mismatch。 - live_gate_decision：verified=False 时返回 409 语义 （allowed=False、http_status=409、status="blocked"、 blocked=True），verified=True 时返回 200 语义。 诚实边界：本文件测的是判定函数，不冒充 HTTP 端点 已接线——_merge_task / _release_task 本批未接 live_gate_decision（无生产 reader 可注入；见 gate_verification 模块头残留），真实 live 验证仍待 B6 真实 App 安装与 B1 等用户侧事项。 """ import ast import unittest from pathlib import Path try: from cloud.gate_verification import LiveChecksResult from cloud.gate_verification import LiveEvidenceReader from cloud.gate_verification import LiveProtectionResult from cloud.gate_verification import live_gate_decision from cloud.gate_verification import verify_live_evidence except ImportError: import sys sys.path.insert(0, str(Path(__file__).resolve().parents[2])) from cloud.gate_verification import LiveChecksResult from cloud.gate_verification import LiveEvidenceReader from cloud.gate_verification import LiveProtectionResult from cloud.gate_verification import live_gate_decision from cloud.gate_verification import verify_live_evidence def _claimed_green(): return { "ci_status": "success", "ci_checks": [{"name": "ci", "status": "success"}], "required_checks": ["ci"], "branch_protection": "protected", } class FakeReader(LiveEvidenceReader): """实现协议的假 reader：不发网络请求，只返回预设值。 同时记录收到的 repo/ref/branch，断言接线把求证目标 原样传给 reader（不猜目标）。 """ def __init__(self, checks, protection, checks_raw=None, protection_raw=None): self.checks = checks self.protection = protection self.checks_raw = checks_raw self.protection_raw = protection_raw self.seen = [] def get_checks(self, repo, ref): self.seen.append(("checks", repo, ref)) return LiveChecksResult(checks=dict(self.checks), raw=self.checks_raw) def get_branch_protection(self, repo, branch): self.seen.append(("protection", repo, branch)) return LiveProtectionResult(status=self.protection, raw=self.protection_raw) class FailingReader(LiveEvidenceReader): def __init__(self, fail_on): self.fail_on = fail_on def get_checks(self, repo, ref): if self.fail_on == "checks": raise RuntimeError("github down") return LiveChecksResult(checks={"ci": "success"}) def get_branch_protection(self, repo, branch): if self.fail_on == "protection": raise RuntimeError("github down") return LiveProtectionResult(status="protected") class LiveEvidenceTestCase(unittest.TestCase): def test_matching_fake_reader_verified(self): reader = FakeReader({"ci": "success"}, "protected", checks_raw={"total_count": 1}, protection_raw={"required": True}) result = verify_live_evidence( _claimed_green(), reader, "o/r", "sha1", branch="main") self.assertTrue(result["checked"]) self.assertTrue(result["verified"]) self.assertEqual(result["mismatches"], []) self.assertIsNone(result["reason"]) self.assertEqual(result["live"]["ci_checks"], {"ci": "success"}) self.assertEqual(result["live"]["branch_protection"], "protected") self.assertEqual(result["live"]["checks_raw"], {"total_count": 1}) self.assertEqual(result["live"]["protection_raw"], {"required": True}) self.assertIn(("checks", "o/r", "sha1"), reader.seen) self.assertIn(("protection", "o/r", "main"), reader.seen) def test_lied_all_green_vs_failing_live_mismatch_to_409(self): # 谎报全绿：声称 ci success + required ci，但实时数据 # 里 ci 是 failure、且 required 的 lint 查无此项。 claimed = _claimed_green() claimed["required_checks"] = ["ci", "lint"] reader = FakeReader({"ci": "failure"}, "protected") result = verify_live_evidence(claimed, reader, "o/r", "sha1") self.assertTrue(result["checked"]) self.assertFalse(result["verified"]) self.assertTrue(any("ci_not_green_live" in m for m in result["mismatches"])) self.assertTrue(any("check_not_green_live: ci" in m for m in result["mismatches"])) self.assertTrue(any("required_check_not_green_live: lint" in m for m in result["mismatches"])) decision = live_gate_decision(claimed, reader, "o/r", "sha1") self.assertFalse(decision["allowed"]) self.assertTrue(decision["blocked"]) self.assertEqual(decision["http_status"], 409) self.assertEqual(decision["status"], "blocked") self.assertEqual(decision["code"], "live_evidence_mismatch") self.assertFalse(decision["live"]["verified"]) def test_protection_mismatch_blocks(self): reader = FakeReader({"ci": "success"}, "unprotected") result = verify_live_evidence( _claimed_green(), reader, "o/r", "sha1") self.assertTrue(result["checked"]) self.assertFalse(result["verified"]) self.assertTrue(any("branch_protection" in m for m in result["mismatches"])) decision = live_gate_decision( _claimed_green(), reader, "o/r", "sha1") self.assertEqual(decision["http_status"], 409) self.assertEqual(decision["status"], "blocked") def test_no_reader_is_not_a_pass(self): result = verify_live_evidence( _claimed_green(), None, "o/r", "sha1") self.assertFalse(result["checked"]) self.assertFalse(result["verified"]) self.assertEqual(result["reason"], "no_trusted_reader") self.assertIsNone(result["live"]) decision = live_gate_decision( _claimed_green(), None, "o/r", "sha1") self.assertFalse(decision["allowed"]) self.assertEqual(decision["http_status"], 409) self.assertEqual(decision["status"], "blocked") self.assertEqual(decision["code"], "live_evidence_unverified") def test_no_repo_or_ref_is_not_a_pass(self): reader = FakeReader({"ci": "success"}, "protected") for repo, ref in ((None, "sha1"), ("o/r", None), ("o/r", ""), ("", "sha1")): result = verify_live_evidence( _claimed_green(), reader, repo, ref) self.assertFalse(result["checked"]) self.assertFalse(result["verified"]) self.assertEqual(result["reason"], "no_verification_target") decision = live_gate_decision( _claimed_green(), reader, repo, ref) self.assertEqual(decision["http_status"], 409) def test_reader_failure_is_not_a_pass(self): for fail_on in ("checks", "protection"): result = verify_live_evidence( _claimed_green(), FailingReader(fail_on), "o/r", "sha1") self.assertFalse(result["checked"]) self.assertFalse(result["verified"]) self.assertTrue( result["reason"].startswith("reader_failed")) decision = live_gate_decision( _claimed_green(), FailingReader(fail_on), "o/r", "sha1") self.assertEqual(decision["http_status"], 409) self.assertEqual(decision["code"], "live_evidence_unverified") def test_malformed_reader_result_is_not_a_pass(self): class BadChecks(LiveEvidenceReader): def get_checks(self, repo, ref): return 42 def get_branch_protection(self, repo, branch): return LiveProtectionResult(status="protected") class BadStatus(LiveEvidenceReader): def get_checks(self, repo, ref): return LiveChecksResult(checks={"ci": 1}) def get_branch_protection(self, repo, branch): return LiveProtectionResult(status="protected") for reader in (BadChecks(), BadStatus()): result = verify_live_evidence( _claimed_green(), reader, "o/r", "sha1") self.assertFalse(result["checked"]) self.assertFalse(result["verified"]) self.assertTrue( result["reason"].startswith("reader_failed")) def test_matching_fake_reader_decision_allows(self): reader = FakeReader({"ci": "success"}, "protected") decision = live_gate_decision( _claimed_green(), reader, "o/r", "sha1", branch="main") self.assertTrue(decision["allowed"]) self.assertFalse(decision["blocked"]) self.assertEqual(decision["http_status"], 200) self.assertEqual(decision["status"], "allowed") self.assertEqual(decision["code"], "live_evidence_verified") def test_branch_defaults_to_ref_and_explicit_branch_wins(self): reader = FakeReader({"ci": "success"}, "protected") verify_live_evidence(_claimed_green(), reader, "o/r", "sha1") self.assertIn(("protection", "o/r", "sha1"), reader.seen) reader2 = FakeReader({"ci": "success"}, "protected") verify_live_evidence(_claimed_green(), reader2, "o/r", "sha1", branch="main") self.assertIn(("protection", "o/r", "main"), reader2.seen) def test_callable_legacy_reader_still_works(self): def live_ok(repo, ref): self.assertEqual((repo, ref), ("o/r", "sha1")) return {"ci_checks": {"ci": "success"}, "branch_protection": "protected"} result = verify_live_evidence( _claimed_green(), live_ok, "o/r", "sha1") self.assertTrue(result["verified"]) def live_bad(repo, ref): return {"ci_checks": {"ci": "failure"}, "branch_protection": "protected"} result = verify_live_evidence( _claimed_green(), live_bad, "o/r", "sha1") self.assertFalse(result["verified"]) self.assertTrue(any("ci_not_green_live" in m for m in result["mismatches"])) decision = live_gate_decision( _claimed_green(), live_bad, "o/r", "sha1") self.assertEqual(decision["http_status"], 409) def test_no_real_github_call_in_production_module(self): # 注意：旧模块头 docstring 本来就以文字描述过“真实 # reader 用 github_app 的 _github_api_request”（未来 # 设想，非代码），所以断言须区分文字与代码：代码 # 位置（import / 调用 / 非 docstring 字符串）里不 # 得出现真实调用，docstring 文字不算。 path = Path(__file__).resolve().parents[1] / "gate_verification.py" tree = ast.parse(path.read_text()) imported = set() docstring_nodes = set() for node in ast.walk(tree): if isinstance(node, (ast.Module, ast.ClassDef, ast.FunctionDef, ast.AsyncFunctionDef)): body = node.body if body and isinstance(body[0], ast.Expr) \ and isinstance(body[0].value, ast.Constant) \ and isinstance(body[0].value.value, str): docstring_nodes.add(id(body[0].value)) if isinstance(node, ast.Import): imported.update(a.name.split(".")[0] for a in node.names) elif isinstance(node, ast.ImportFrom): if node.module: imported.add(node.module.split(".")[0]) for banned in ("urllib", "requests", "http", "socket"): self.assertNotIn(banned, imported, "production module must not import %s" % banned) code_idents = set() for node in ast.walk(tree): if isinstance(node, ast.Attribute): code_idents.add(node.attr) elif isinstance(node, ast.Name): code_idents.add(node.id) elif isinstance(node, ast.Constant) \ and isinstance(node.value, str) \ and id(node) not in docstring_nodes: self.assertNotIn("api.github.com", node.value) self.assertNotIn("_github_api_request", code_idents) if __name__ == "__main__": unittest.main()