"""gapfill2 H2：全仓低于 90% 文件的扫尾（生产代码 0 改动）。 基线（本批自己跑的 CI 同款覆盖率，基底 5d0651e）低于 90% 的 文件共 37 个。本文件补其中三个缺口最小、可纯函数 / 注入式 真实驱动的文件，缺口大的文件（resident、sandbox、ledger、 playwright_browser 等，缺口均在 50 行以上且多需 Docker / 真实 外部服务）不在本批硬凑，见总结残留清单： - cloud/approval_timeout.py：57 语句、缺失 8 行（41-42、46、 48、62-63、65、109）→ 本文件全覆盖。 - cloud/merge_gate_hard.py：89 语句、缺失 12 行（103、108、 142-143、147、157-158、201-202、212-215）→ 本文件全覆盖。 - cloud/worker_sweep.py：60 语句、缺失 8 行（54、82-83、 117-118、131-132、174）→ 本文件覆盖除 174（``__main__`` 守卫行，import 无法覆盖）外的 7 行。 真实模型 0 轮，不联网；approval_timeout 与 merge_gate_hard 均为纯函数直调，worker_sweep 用假 AuthStore / 假 reap 注入。 """ import sqlite3 import sys import unittest from pathlib import Path from unittest import mock REPO_ROOT = Path(__file__).resolve().parents[2] sys.path.insert(0, str(REPO_ROOT)) import cloud.worker_sweep as sweep_mod # noqa: E402 from cloud.approval_timeout import ( # noqa: E402 approval_timeout_stats, recommend_approval_timeout, timeout_rate, ) from cloud.merge_gate_hard import evaluate_hard_merge_gate # noqa: E402 def _gate(**overrides): body = { "files_changed": ["apps/web/a.py"], "owned_dir": "apps/web", "protection": "protected", "ci_status": "success", "ci_checks": [{"name": "ci", "status": "success"}], "required_checks": ["ci"], "contract_version": "v2", "contract_frozen": True, "coordinator_approved": True, "unresolved_review_comments": 0, } body.update(overrides) return evaluate_hard_merge_gate(**body) class ApprovalTimeoutBranchTestCase(unittest.TestCase): def test_timeout_rate_non_numeric_none(self): # 41-42：计数无法转 int 时回 None，不是 0%。 self.assertIsNone(timeout_rate("abc", 10)) self.assertIsNone(timeout_rate(1, "xyz")) def test_timeout_rate_clamps(self): # 46：expired 为负时按 0 计；48：expired 超过 decided # 时按 decided 计（比率封顶 1.0）。 self.assertEqual(timeout_rate(-3, 10), 0.0) self.assertEqual(timeout_rate(15, 10), 1.0) def test_recommend_current_fallbacks(self): # 62-63：current 非法时按默认值 600 计；65：current # <= 0 时同样按默认值计。 self.assertEqual( recommend_approval_timeout(0, 10, current_seconds="bad") ["current_seconds"], 600) self.assertEqual( recommend_approval_timeout(0, 10, current_seconds=0) ["current_seconds"], 600) self.assertEqual( recommend_approval_timeout(0, 10, current_seconds=-5) ["current_seconds"], 600) def test_stats_empty_project_list(self): # 109：project_ids 为空列表时直接回零统计，不拼 IN 查询。 conn = sqlite3.connect(":memory:") try: self.assertEqual( approval_timeout_stats(conn, []), {"decided_count": 0, "expired_count": 0, "pending_count": 0}) finally: conn.close() class MergeGateHardBranchTestCase(unittest.TestCase): def test_ci_check_entry_not_dict(self): # 103 + 147：ci_checks 条目不是 dict 时索引为 None、 # 记 ci_check_failed 块。 result = _gate(ci_checks=["not-a-dict"]) self.assertFalse(result["allowed"]) self.assertIn("ci_check_failed", result["blocks"]) def test_ci_check_entry_bad_name_status(self): # 108 + 147：条目 name 为空时索引为 None、记 # ci_check_failed 块。 result = _gate(ci_checks=[{"name": "", "status": "success"}]) self.assertFalse(result["allowed"]) self.assertIn("ci_check_failed", result["blocks"]) def test_ci_checks_not_list(self): # 142-143：ci_checks 非列表时记 ci_evidence_missing。 result = _gate(ci_checks="ci") self.assertFalse(result["allowed"]) self.assertIn("ci_evidence_missing", result["blocks"]) def test_required_checks_not_list(self): # 157-158：required_checks 非列表时记 ci_evidence_missing。 result = _gate(required_checks="ci") self.assertFalse(result["allowed"]) self.assertIn("ci_evidence_missing", result["blocks"]) def test_reviews_bool_is_missing_evidence(self): # 201-202：评审意见为 bool 时不是“0 条”，而是无证据。 result = _gate(unresolved_review_comments=True) self.assertFalse(result["allowed"]) self.assertIn("review_evidence_missing", result["blocks"]) def test_reviews_negative_and_other_type(self): # 212：负数评审意见记 review_evidence_missing； # 214-215：其它类型（字符串）同样记无证据。 result = _gate(unresolved_review_comments=-1) self.assertIn("review_evidence_missing", result["blocks"]) result = _gate(unresolved_review_comments="many") self.assertIn("review_evidence_missing", result["blocks"]) class WorkerSweepBranchTestCase(unittest.TestCase): def _fake_store_cls(self, user_id, audit_raises=False, close_raises=False): class _FakeConn: def close(self): if close_raises: raise RuntimeError("close failed") class _FakeStore: def __init__(self, db_path): self.conn = _FakeConn() def authenticate(self, api_key): return user_id if api_key else None def audit(self, *args, **kwargs): if audit_raises: raise RuntimeError("audit down") return _FakeStore def test_global_admin_ids_explicit(self): # 54：显式传入 admin_ids 时去空白、去空项。 self.assertEqual( sweep_mod.global_admin_ids([" a ", "", "b", " a "]), {"a", "b"}) def test_denied_audit_failure_swallowed(self): # 82-83：非管理员被拒时，拒绝审计本身抛错也被吞掉， # PermissionError 照常抛出，绝不调 reap。 fake = self._fake_store_cls("u1", audit_raises=True) with mock.patch.object(sweep_mod, "AuthStore", fake): with self.assertRaises(PermissionError): sweep_mod.sweep_stale_workers_admin( "/tmp/x.db", "k", admin_ids=["admin1"]) def test_sweep_audit_failure_marks_summary_false(self): # 117-118：汇总审计抛错时 audit_summary 置 False， # 调用仍正常返回（reap 结果不受影响）。 fake = self._fake_store_cls("admin1", audit_raises=True) with mock.patch.object(sweep_mod, "AuthStore", fake): with mock.patch.object( sweep_mod, "reap_stale_workers", return_value=[]): summary = sweep_mod.sweep_stale_workers_admin( "/tmp/x.db", "k", admin_ids=["admin1"]) self.assertFalse(summary["audit_summary"]) self.assertEqual(summary["killed"], []) def test_store_close_failure_swallowed(self): # 131-132：finally 里 store.conn.close() 抛错时被吞掉， # 汇总照常返回。 fake = self._fake_store_cls("admin1", close_raises=True) with mock.patch.object(sweep_mod, "AuthStore", fake): with mock.patch.object( sweep_mod, "reap_stale_workers", return_value=[]): summary = sweep_mod.sweep_stale_workers_admin( "/tmp/x.db", "k", admin_ids=["admin1"]) self.assertTrue(summary["audit_summary"]) self.assertEqual(summary["user_id"], "admin1") if __name__ == "__main__": unittest.main()