host
aba06d9a24
H1 worker.py 92%->94% (277->210 miss): 35 real-assertion tests covering goal-signal, tenant/env, followup, poll/wait, process/watchdog and injected defensive branches; prod code 0 changes. H2 sub-90 sweep: approval_timeout 100%, merge_gate_hard 100%, worker_sweep 98% (only __main__ line 174 left); 14 tests. H3 G7 rollback except:pass (server.py:6615-6616) covered by an injected rollback-raising connection; original error still propagates, status stays completed, retry succeeds; no semantics change. H4 caller contract documented only (docstrings in _audit_scorecard endpoint, check_baseline_stale, build_scorecard, baseline store): current_main_head caller-supplied, any string as-is, no format checks, missing/empty treated as not passed with no stale determination at the endpoint, always 200, overall never rewritten; no new validation or rejection, no response-structure change. 3 doc tests pin the wording. H5 design notes (no decisions, no code changes) live in verification/code-to10-gapfill2-design-notes.md, referenced by the batch summary. No bug found by the new tests; no existing assertion touched.
201 lines
8.0 KiB
Python
201 lines
8.0 KiB
Python
"""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()
|