From 6ba822daa24abc6bb23678881a0dd750f12d5e18 Mon Sep 17 00:00:00 2001 From: host Date: Wed, 4 Nov 2026 16:20:10 +0000 Subject: [PATCH] gapfill2: worker/sub90 coverage tests, G7 rollback branch, G5 caller-contract docs 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. --- cloud/audit_scorecard.py | 16 +- cloud/scorecard_baseline_store.py | 4 +- cloud/server.py | 10 + cloud/tests/test_gapfill2_g7_rollback.py | 163 ++++++ cloud/tests/test_gapfill2_h4_docs.py | 61 +++ cloud/tests/test_gapfill2_sub90.py | 200 ++++++++ cloud/tests/test_gapfill2_worker_gaps.py | 610 +++++++++++++++++++++++ 7 files changed, 1062 insertions(+), 2 deletions(-) create mode 100644 cloud/tests/test_gapfill2_g7_rollback.py create mode 100644 cloud/tests/test_gapfill2_h4_docs.py create mode 100644 cloud/tests/test_gapfill2_sub90.py create mode 100644 cloud/tests/test_gapfill2_worker_gaps.py diff --git a/cloud/audit_scorecard.py b/cloud/audit_scorecard.py index 8f984ce..de8fafc 100644 --- a/cloud/audit_scorecard.py +++ b/cloud/audit_scorecard.py @@ -31,6 +31,9 @@ build_scorecard(..., baseline_head=None) 的 baseline_head 是机器字段， 只能由调用方显式传入当前 main 头；本模块绝不在生产代码里探测 版本控制状态、绝不读版本控制目录、绝不编造一个头。 + 接受任意字符串、原样记录，不做任何格式校验（不要求形如 + commit hash）；本模块不验证该值是否真实、可信，可信 main 头 + 的来源仍是调用方责任。 不传入（或传入空串）时 baseline_head 为 None，并在 missing_items 里记 baseline_head_missing，不把缺失读成“与当前一致”。 check_baseline_stale(scorecard, current_main_head) 是唯一的漂移判定： @@ -916,6 +919,16 @@ def check_baseline_stale(scorecard, current_main_head): 参数全部由调用方传入：scorecard 是 build_scorecard 的输出 dict， current_main_head 是调用方当前确认的 main 头。本函数不在生产 代码里探测版本控制状态，也不读版本控制目录。 + 调用方契约（gapfill2 H4，只写文档、不改行为）： + current_main_head 接受任意字符串、原样比对，不做任何格式 + 校验（不要求形如 commit hash，不校验长度、字符集、前缀）； + 本函数不验证该值是否真实、可信、是否确为 main 头——可信 + main 头的来源仍是调用方责任。空串先归一化为 None，与未 + 传入同等处理。注意端点层与函数层的区别：评分卡端点在缺失 + 或空串时根本不调用本函数（不做判定、不附加判定字段）； + 调用方若直接调本函数且 current_main_head 缺失，本函数按 + 失败关闭返回 stale=True（见下三种情形之二），那是函数层 + 面的判定结果，不是端点行为。 返回 dict（调用方凭 stale / usable_as_pass_basis 判断，绝不能 只看旧卡的 overall / overall_pass）： @@ -1068,7 +1081,8 @@ def build_scorecard(evidence, thresholds=None, rubric_version=None, （user_side_items 可选占位），product=build_product_scorecard （无真实产品轮证据时恒为 not_pass）。新增键不改变旧字段的值。 另新增 rubric_version（见 cloud/scorecard_rubric.py）、 - baseline_head（机器字段，见 check_baseline_stale）、 + baseline_head（机器字段，见 check_baseline_stale；接受任意 + 字符串原样记录、不做格式校验，空串按未传入处理）、 evidence_ledger_head 与 evidence_row_ids（见 compute_evidence_ledger）三个机制字段；无证据行时头为 None。 """ diff --git a/cloud/scorecard_baseline_store.py b/cloud/scorecard_baseline_store.py index f2fc55f..4f95673 100644 --- a/cloud/scorecard_baseline_store.py +++ b/cloud/scorecard_baseline_store.py @@ -13,7 +13,9 @@ 行字段： - rubric_version：评分口径版本（调用方传入的字符串，如 "v1"； 本模块不规定口径内容、不替用户拍板阈值）。 -- baseline_head：冻结时调用方传入的 main 头（字符串，非空）。 +- baseline_head：冻结时调用方传入的 main 头（字符串，非空； + 接受任意字符串原样存、不做格式校验，本模块不验证其真实性， + 可信来源仍是调用方责任）。 - frozen_at：冻结时间戳（秒，调用方不传时取当前时间）。 - scorecard_snapshot：build_scorecard 输出的完整 dict，以 JSON 文本存入，读回时还原为 dict。 diff --git a/cloud/server.py b/cloud/server.py index 365f468..3f0b8e5 100644 --- a/cloud/server.py +++ b/cloud/server.py @@ -9194,6 +9194,16 @@ pid 须为整数、address 须为字符串，否则 400（注册表里的 参数也永远返回 200 + 报告，只是多一个判定字段。 生产代码绝不读 git、不起 subprocess 查版本：main 头 只能由调用方传入，端点本身不探测版本控制状态。 + 调用方契约（gapfill2 H4，只写文档、不改行为）： + current_main_head 由调用方提供，接受任意字符串、原样 + 使用，不做任何格式校验（不要求形如 commit hash，不 + 校验长度、字符集、前缀）；端点不验证该值是否真实、 + 是否可信、是否确为 main 头——可信 main 头的来源仍是 + 调用方责任。缺失（未传该查询参数）与空串一律按未传入 + 处理：端点不调用判定、不附加 baseline_staleness 字段， + 响应与以前完全一致。传入非空值后，端点永远返回 200 + + 报告、只附加判定字段：判定为 stale 也不挡查询，绝不改 + 写报告的 overall / overall_pass。 权限：与 /metrics 同款，任意项目成员即可读（不要求 owner）， 但本端点不统计任何项目数据、不写库、不写账本、不改链头、 diff --git a/cloud/tests/test_gapfill2_g7_rollback.py b/cloud/tests/test_gapfill2_g7_rollback.py new file mode 100644 index 0000000..5919e75 --- /dev/null +++ b/cloud/tests/test_gapfill2_g7_rollback.py @@ -0,0 +1,163 @@ +"""gapfill2 H3：_merge_task 成功审计段里 rollback 自身的 except: pass。 + +上一批唯一未覆盖分支：G7 新段里 ``conn.rollback()`` 本身抛错时的 +防御分支（基底行号 server.py:6615-6616）。rollback 从不失败，故 +全量覆盖率里该分支一直缺失。本文件只用注入式测试覆盖它，绝不 +改动 rollback 语义、绝不把该分支改成可观测行为： + +- 注入一个 rollback 会抛错的连接代理（除 rollback 外全部透传 + 真实连接），同时让审计失败（沿用 G7 的 RuntimeError 注入）， + 使执行流必然走进 ``except Exception: pass``； +- 断言：rollback 的异常被吞掉（原审计错误原样上抛，不被掩盖）、 + 任务状态仍是 completed、审计 0 行、原样重试后成功 200。 + +真实模型 0 轮，不联网；不跑真实 HTTP（handler 直调 + +真实 AuthStore/DB，审计行直接查库验证）。 +""" + +import os +import sqlite3 +import sys +import tempfile +import time +import unittest +from pathlib import Path +from types import SimpleNamespace +from unittest import mock + +REPO_ROOT = Path(__file__).resolve().parents[2] +sys.path.insert(0, str(REPO_ROOT)) + +import cloud.server as server_mod # noqa: E402 +from cloud.auth import AuthStore # noqa: E402 +from cloud.server import CloudControlPlaneHandler, init_db # noqa: E402 + +NO_BLOCK = {"merge_blocked": False, "release_blocked": False, + "blocked_by": None, "round": None, "failed_checks": []} + +HARD_BODY = { + "files_changed": ["apps/web/a.py"], + "owned_dir": "apps/web", + "branch_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, +} + + +class _RollbackFailingConn: + """代理真实连接：除 rollback 外全部透传，rollback 计数后抛错。 + + execute / commit / close / row_factory 等一律委托给真实连接， + 只有 rollback 被替换为“计数 + 抛错”，精确模拟“rollback 自身 + 失败”这一防御分支的前提，不改动其它任何连接语义。 + """ + + def __init__(self, real, calls): + self._real = real + self._calls = calls + + def rollback(self): + self._calls["n"] += 1 + raise RuntimeError("rollback down") + + def __getattr__(self, name): + return getattr(self._real, name) + + +class RollbackFailureSwallowedTestCase(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.db_path = os.path.join(self.tmp.name, "control.db") + init_db(self.db_path) + self.auth = AuthStore(self.db_path) + self.owner_id, _key = self.auth.create_user("owner") + conn = sqlite3.connect(self.db_path) + try: + conn.execute( + "INSERT INTO projects(id, name, owner_id, created_at," + " repo_url, auto_pr) VALUES(?, ?, ?, ?, ?, 0)", + ("p1", "P", self.owner_id, time.time(), None)) + conn.execute( + "INSERT INTO tasks(id, project_id, prompt, status," + " created_by, created_at, updated_at)" + " VALUES(?, ?, ?, ?, ?, ?, ?)", + ("t1", "p1", "do work", "completed", self.owner_id, + time.time(), time.time())) + conn.commit() + finally: + conn.close() + self.auth.create_project("p1", self.owner_id) + + def tearDown(self): + try: + self.auth.conn.close() + except Exception: + pass + self.tmp.cleanup() + + def _handler(self): + h = CloudControlPlaneHandler.__new__(CloudControlPlaneHandler) + h.server = SimpleNamespace(db_path=self.db_path) + sent = [] + h._send_json = lambda status, obj, extra_headers=None: sent.append( + (status, obj)) + h._read_json = lambda: (dict(HARD_BODY), None) + h.sent = sent + return h + + def _merge(self, h): + with mock.patch.object(server_mod, "_pushback_block", + return_value=dict(NO_BLOCK)): + h._merge_task(self.owner_id, "t1") + + def _status(self): + conn = sqlite3.connect(self.db_path) + try: + return conn.execute( + "SELECT status FROM tasks WHERE id = 't1'").fetchone()[0] + finally: + conn.close() + + def _merge_audits(self): + return [a for a in self.auth.list_audit("p1") + if a["action"] == "merge_task"] + + def test_rollback_failure_is_swallowed(self): + # 审计失败 + rollback 自身失败：rollback 的异常必须被吞掉， + # 原审计错误原样上抛（不被 rollback 错误掩盖），状态回滚 + # 语义不变（completed、审计 0 行），原样重试后成功。 + h = self._handler() + real_db = CloudControlPlaneHandler._db + calls = {"n": 0} + + def failing_db(handler_self): + return _RollbackFailingConn(real_db(handler_self), calls) + + with mock.patch.object(CloudControlPlaneHandler, "_db", + failing_db): + with mock.patch.object( + server_mod, "_audit_on_conn", + side_effect=RuntimeError("audit down")): + with self.assertRaises(RuntimeError) as ctx: + self._merge(h) + self.assertEqual(str(ctx.exception), "audit down") + self.assertEqual(calls["n"], 1) + self.assertEqual(h.sent, []) + self.assertEqual(self._status(), "completed") + self.assertEqual(self._merge_audits(), []) + # 原样重试（审计恢复、连接恢复后）必须成功：状态仍是 + # completed，不是卡在 merged 导致重复合并 409。 + h2 = self._handler() + self._merge(h2) + self.assertEqual(h2.sent[-1][0], 200) + self.assertEqual(self._status(), "merged") + self.assertEqual(len(self._merge_audits()), 1) + + +if __name__ == "__main__": + unittest.main() diff --git a/cloud/tests/test_gapfill2_h4_docs.py b/cloud/tests/test_gapfill2_h4_docs.py new file mode 100644 index 0000000..911b8eb --- /dev/null +++ b/cloud/tests/test_gapfill2_h4_docs.py @@ -0,0 +1,61 @@ +"""gapfill2 H4：G5 调用方契约的文档钉死（只改文档、不改行为）。 + +本文件断言的全部是 docstring 文本，不是行为：上一批残留 +“可信 main 头的来源仍是调用方责任”在本批补进 scorecard +相关模块与端点的 docstring 后，必须写清四点—— +current_main_head 由调用方提供、接受什么形式的值、缺失与 +空串按未传入处理且不做 stale 判定、传入后永远 200 且不改写 +overall；且绝不新增任何校验或拒绝行为（行为断言见既有 +test_gapfill_g5_baseline_snapshot，本文件不重复测行为）。 +""" + +import sys +import unittest +from pathlib import Path + +REPO_ROOT = Path(__file__).resolve().parents[2] +sys.path.insert(0, str(REPO_ROOT)) + +import cloud.audit_scorecard as scorecard_mod # noqa: E402 +import cloud.scorecard_baseline_store as store_mod # noqa: E402 +from cloud.server import CloudControlPlaneHandler # noqa: E402 + + +def _flat(doc): + """docstring 空白归一化：去掉全部空白后按语义短语查， + 不依赖换行位置（中文短语常被换行切开）。""" + return "".join((doc or "").split()) + + +class CallerContractDocsTestCase(unittest.TestCase): + def test_endpoint_docstring_states_caller_contract(self): + doc = _flat(CloudControlPlaneHandler._audit_scorecard.__doc__) + self.assertIn("由调用方提供", doc) + self.assertIn("接受任意字符串", doc) + self.assertIn("不做任何格式校验", doc) + self.assertIn("不要求形如commithash", doc) + self.assertIn("缺失", doc) + self.assertIn("空串一律按未传入处理", doc) + self.assertIn("不附加baseline_staleness字段", doc) + self.assertIn("永远返回200", doc) + self.assertIn("绝不改写报告的overall/overall_pass", doc) + self.assertIn("可信main头的来源仍是调用方责任", doc) + + def test_check_baseline_stale_docstring_states_contract(self): + doc = _flat(scorecard_mod.check_baseline_stale.__doc__) + self.assertIn("接受任意字符串", doc) + self.assertIn("不做任何格式校验", doc) + self.assertIn("空串先归一化为None", doc) + self.assertIn("根本不调用本函数", doc) + self.assertIn("可信", doc) + + def test_module_and_store_docstrings_state_contract(self): + self.assertIn("接受任意字符串", + _flat(scorecard_mod.__doc__)) + self.assertIn("不做格式校验", _flat(store_mod.__doc__)) + self.assertIn("可信来源仍是调用方责任", + _flat(store_mod.__doc__)) + + +if __name__ == "__main__": + unittest.main() diff --git a/cloud/tests/test_gapfill2_sub90.py b/cloud/tests/test_gapfill2_sub90.py new file mode 100644 index 0000000..a0ee692 --- /dev/null +++ b/cloud/tests/test_gapfill2_sub90.py @@ -0,0 +1,200 @@ +"""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() diff --git a/cloud/tests/test_gapfill2_worker_gaps.py b/cloud/tests/test_gapfill2_worker_gaps.py new file mode 100644 index 0000000..1418167 --- /dev/null +++ b/cloud/tests/test_gapfill2_worker_gaps.py @@ -0,0 +1,610 @@ +"""gapfill2 H1：worker.py 覆盖缺口补测试（生产代码 0 改动）。 + +基线（本批自己跑的 CI 同款覆盖率，基底 5d0651e）： +cloud/worker.py 3493 语句、277 缺失、92%。本文件只加真实断言 +测试，瞄准基线缺失行里可真实驱动的分支；每组测试头上注明 +它覆盖的基底行号。需要 Docker / 协调员 / run_one 主循环等重 +型 harness 的缺失行（run_one、_run_attempt、_docker_task_context、 +run_loop、run_all 等）不在本文件硬凑，见总结残留清单。 + +真实模型 0 轮，不联网；不启动真实 Docker / 子进程，凡需注入 +处均用假连接、假 runner、假 proc 或方法补丁，且断言均为行为 +断言（返回值、落盘行、调用记录），不是“跑过即过”。 +""" + +import json +import logging +import os +import sqlite3 +import sys +import tempfile +import threading +import time +import unittest +from pathlib import Path +from types import SimpleNamespace +from unittest import mock + +REPO_ROOT = Path(__file__).resolve().parents[2] +sys.path.insert(0, str(REPO_ROOT)) + +import cloud.worker as worker_mod # noqa: E402 +from cloud.server import init_db # noqa: E402 +from cloud.worker import DockerExecutor, Worker # noqa: E402 + + +def _worker(db_path, **attrs): + w = Worker.__new__(Worker) + w.db_path = db_path + w.tenant_isolation = False + w.executor = SimpleNamespace() + w.logger = logging.getLogger("test_gapfill2_worker") + w._cancel_requested = set() + w._running = {} + w._running_lock = threading.Lock() + w._loop_stop = False + w.quota_poll_interval_seconds = 30 + w.process_factory = None + w.repos_runner = None + for key, value in attrs.items(): + setattr(w, key, value) + return w + + +def _insert_task(db_path, task_id="T1", status="running", goal="old goal", + goal_version=0, applied=0): + now = time.time() + conn = sqlite3.connect(db_path) + try: + conn.execute( + "INSERT INTO tasks(id, project_id, prompt, status," + " idempotency_key, depends_on, priority, created_by," + " created_at, updated_at, last_heartbeat, goal, goal_mode," + " verify_command, goal_max_rounds, goal_version," + " applied_goal_version) VALUES(?, 'p1', ?, ?, NULL, '[]'," + " 0, 'tester', ?, ?, NULL, ?, 1, NULL, 0, ?, ?)", + (task_id, goal, status, now, now, goal, goal_version, + applied)) + conn.commit() + finally: + conn.close() + + +def _task_row(db_path, task_id="T1"): + conn = sqlite3.connect(db_path) + try: + conn.row_factory = sqlite3.Row + return dict(conn.execute( + "SELECT * FROM tasks WHERE id = ?", (task_id,)).fetchone()) + finally: + conn.close() + + +def _write_state(state_path, tasks): + with open(state_path, "w", encoding="utf-8") as fh: + json.dump({"tasks": tasks}, fh) + + +class GoalSignalBranchTestCase(unittest.TestCase): + """goal 信号写侧 / 镜像分支（基底缺失 1294 / 1300 / 1306 / + 1313-1314 / 1339-1340 / 1381-1382 / 1397-1398）。""" + + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.db_path = os.path.join(self.tmp.name, "test.db") + init_db(self.db_path) + self.state_path = os.path.join(self.tmp.name, "host-state.json") + + def tearDown(self): + self.tmp.cleanup() + + def _signal_worker(self): + w = _worker(self.db_path) + w._host_state_path_for_task = lambda task: self.state_path + return w + + def test_bad_goal_rejected(self): + # 1294：db_goal 为空或非字符串时返回 False，不写信号。 + w = self._signal_worker() + task = {"id": "T1", "project_id": "p1"} + self.assertFalse(w._write_goal_signal_if_newer(task, None, 2)) + self.assertFalse(w._write_goal_signal_if_newer(task, "", 2)) + self.assertFalse(w._write_goal_signal_if_newer(task, 123, 2)) + self.assertFalse(os.path.exists( + os.path.join(self.tmp.name, "goal-updates"))) + + def test_no_state_path_returns_false(self): + # 1300：解析不出 state 路径时返回 False，不猜路径。 + w = _worker(self.db_path) + w._host_state_path_for_task = lambda task: None + self.assertFalse(w._write_goal_signal_if_newer( + {"id": "T1", "project_id": "p1"}, "new goal", 2)) + + def test_entry_not_dict_returns_false(self): + # 1306：state 里本任务条目不是 dict 时返回 False。 + _write_state(self.state_path, {"T1": "not-a-dict"}) + w = self._signal_worker() + self.assertFalse(w._write_goal_signal_if_newer( + {"id": "T1", "project_id": "p1"}, "new goal", 2)) + + def test_corrupt_state_swallowed(self): + # 1313-1314：state 文件损坏（json 解析失败）时异常被吞掉， + # 返回 False，绝不炸掉调用方。 + with open(self.state_path, "w", encoding="utf-8") as fh: + fh.write("{not json") + w = self._signal_worker() + self.assertFalse(w._write_goal_signal_if_newer( + {"id": "T1", "project_id": "p1"}, "new goal", 2)) + + def test_write_signal_success_control(self): + # 对照项（非缺失行）：版本大于已应用版本时写信号成功， + # 证明上面几组失败测试的 harness 本身能走通成功路径。 + _write_state(self.state_path, { + "T1": {"applied_goal_version": 0, "goal_rounds_done": 1}}) + w = self._signal_worker() + self.assertTrue(w._write_goal_signal_if_newer( + {"id": "T1", "project_id": "p1"}, "new goal", 2)) + signal_dir = os.path.join(self.tmp.name, "goal-updates") + self.assertTrue(os.path.isdir(signal_dir)) + self.assertEqual(len(os.listdir(signal_dir)), 1) + + def test_sync_resume_missing_id_swallowed(self): + # 1339-1340：续跑认领钩子读库时任务缺 id（KeyError）时 + # 异常被吞掉，返回 False，续跑本身不受影响。 + w = _worker(self.db_path) + self.assertFalse(w._sync_resume_goal_signal({})) + + def test_mirror_bad_applied_ignored(self): + # 1381-1382：state 里 applied_goal_version 不是数字时， + # 转换失败置 None、跳过回写，不改任务行、不炸镜像。 + _insert_task(self.db_path, applied=0) + _write_state(self.state_path, {"T1": { + "goal_rounds_done": 3, "status": "goal-running", + "applied_goal_version": "not-a-number", + "thread_id": "th-1"}}) + w = self._signal_worker() + self.assertEqual( + w._mirror_checkpoint_once({"id": "T1", "project_id": "p1"}), 3) + self.assertEqual(_task_row(self.db_path)["applied_goal_version"], + 0) + + def test_mirror_db_read_failure_skips_signal_only(self): + # 1397-1398：镜像内读库（goal/goal_version）失败时只跳过 + # 反向信号，检查点镜像本身继续、返回轮次，不炸、不回 -1。 + _insert_task(self.db_path, applied=0) + _write_state(self.state_path, {"T1": { + "goal_rounds_done": 2, "status": "goal-running", + "applied_goal_version": 0, "thread_id": "th-1"}}) + w = self._signal_worker() + real_connect = worker_mod._db_connect + + class _FailingGoalReadConn: + def __init__(self, real): + self._real = real + + def execute(self, sql, params=()): + if "SELECT goal, goal_version" in sql: + raise sqlite3.OperationalError( + "injected goal read failure") + return self._real.execute(sql, params) + + def __getattr__(self, name): + return getattr(self._real, name) + + def failing_connect(db_path): + return _FailingGoalReadConn(real_connect(db_path)) + + with mock.patch.object(worker_mod, "_db_connect", + failing_connect): + self.assertEqual( + w._mirror_checkpoint_once( + {"id": "T1", "project_id": "p1"}), 2) + conn = sqlite3.connect(self.db_path) + try: + row = conn.execute( + "SELECT round_no FROM task_checkpoints" + " WHERE task_id = 'T1'").fetchone() + finally: + conn.close() + self.assertEqual(row, (2,)) + + +class TenantEnvBranchTestCase(unittest.TestCase): + """租户目录 / 环境分支（基底缺失 3062 / 3091 / 3121-3122 / + 3148-3149 / 3156-3157 / 3181-3182 / 3196）。""" + + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + + def tearDown(self): + self.tmp.cleanup() + + def _tenant_worker(self, **extra): + w = _worker(os.path.join(self.tmp.name, "test.db")) + w.tenant_isolation = True + w.tenants_root = os.path.join(self.tmp.name, "tenants") + w.tenant_template_home = os.path.join(self.tmp.name, "tmpl-home") + w.tenant_template_repo = os.path.join(self.tmp.name, "tmpl-repo") + w.seed_skills = False + w.websearch_mcp = False + w.imagegen_mcp = False + w.playwright_mcp = False + for key, value in extra.items(): + setattr(w, key, value) + return w + + def test_tenant_dirs_missing_test_repo_template(self): + # 3062：test-repo 为空目录、模板不存在时直接抛错， + # 不静默播种一个空 test-repo。 + os.makedirs(self._tenant_worker().tenant_template_home) + with open(os.path.join(self.tmp.name, "tmpl-home", + "config.toml"), "w") as fh: + fh.write("[x]\n") + w = self._tenant_worker() + with self.assertRaises(RuntimeError) as ctx: + w._tenant_dirs("p1") + self.assertIn("tmpl-repo", str(ctx.exception)) + + def test_tenant_mounts_includes_repo_checkout(self): + # 3091：repo_checkout 目录已存在时，挂载列表里追加 + # repos/<安全项目名> 挂载；不存在时不硬挂载。 + tmpl_home = os.path.join(self.tmp.name, "tmpl-home") + tmpl_repo = os.path.join(self.tmp.name, "tmpl-repo") + os.makedirs(tmpl_home) + os.makedirs(tmpl_repo) + with open(os.path.join(tmpl_home, "config.toml"), "w") as fh: + fh.write("[x]\n") + with open(os.path.join(tmpl_repo, "sample.py"), "w") as fh: + fh.write("# sample\n") + w = self._tenant_worker() + dirs = w._tenant_dirs("p1") + os.makedirs(dirs["repo_checkout"], exist_ok=True) + mounts = w._tenant_mounts("p1") + self.assertIn( + "%s:/var/lib/codex-app-server/repos/p1:rw" + % dirs["repo_checkout"], mounts) + + def test_env_host_dir_tenant_docker(self): + # 3121-3122：租户 + Docker 时环境宿主目录走 + # tenants_root/<项目>/envs/。 + w = self._tenant_worker() + w.executor = DockerExecutor.__new__(DockerExecutor) + result = w._env_host_dir({"project_id": "p1", "id": "env1"}) + self.assertTrue(result.startswith(w.tenants_root)) + self.assertTrue(result.endswith( + os.path.join("envs", "env1"))) + + def test_fetch_environment_db_failure_none(self): + # 3148-3149：库连接失败时返回 None，不炸 worker。 + w = _worker(os.path.join(self.tmp.name, "no-such-dir", + "test.db")) + self.assertIsNone(w._fetch_environment("e1")) + + def test_fetch_environment_narrow_schema_none_fields(self): + # 3156-3157：老库 environments 表缺可选列时，_get 兜底 + # 回 None，不炸整行；必备列仍正常读出。 + db_path = os.path.join(self.tmp.name, "test.db") + conn = sqlite3.connect(db_path) + try: + conn.execute( + "CREATE TABLE environments(id TEXT, project_id TEXT," + " name TEXT, status TEXT)") + conn.execute( + "INSERT INTO environments VALUES('e1', 'p1', 'E'," + " 'ready')") + conn.commit() + finally: + conn.close() + w = _worker(db_path) + env = w._fetch_environment("e1") + self.assertEqual(env["id"], "e1") + self.assertEqual(env["project_id"], "p1") + self.assertEqual(env["status"], "ready") + self.assertIsNone(env["branch"]) + self.assertIsNone(env["repo_url"]) + + def test_fetch_environment_missing_required_column_none(self): + # 3170-3171：environments 行缺必备列（project_id）时， + # 外层兜底回 None，不炸 worker。 + db_path = os.path.join(self.tmp.name, "test.db") + conn = sqlite3.connect(db_path) + try: + conn.execute("CREATE TABLE environments(id TEXT)") + conn.execute("INSERT INTO environments VALUES('e1')") + conn.commit() + finally: + conn.close() + w = _worker(db_path) + self.assertIsNone(w._fetch_environment("e1")) + + def test_run_git_typeerror_fallback(self): + # 3181-3182：repos_runner 单参调用抛 TypeError 时， + # 回落到 (cmd, timeout=None) 双参调用。 + calls = [] + _unset = object() + + def runner(*args, timeout=_unset): + calls.append((args, timeout)) + if timeout is _unset: + raise TypeError("single-arg not supported") + return SimpleNamespace(returncode=0, stdout="ok", + stderr="") + + w = _worker(os.path.join(self.tmp.name, "test.db")) + w.repos_runner = runner + result = w._run_git(["git", "status"]) + self.assertEqual(result.returncode, 0) + self.assertEqual(result.stdout, "ok") + self.assertEqual(len(calls), 2) + self.assertIs(calls[1][1], None) + + def test_prepare_environment_no_repo_url(self): + # 3196：环境无 repo_url 时直接返回错误文本，不建目录。 + w = _worker(os.path.join(self.tmp.name, "test.db")) + self.assertEqual(w._prepare_environment({}), + "environment has no repo_url") + + +class DbFailureBranchTestCase(unittest.TestCase): + """查库失败即回落的分支（基底缺失 3333-3334 / 3339-3340 / + 3583-3584 / 3604-3605 / 3620-3621 / 3650-3651 / 3665-3666 / + 3694-3695 / 5343-5344 / 5798-5799）。""" + + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.db_path = os.path.join(self.tmp.name, "test.db") + init_db(self.db_path) + + def tearDown(self): + self.tmp.cleanup() + + def _bad_db_worker(self): + return _worker(os.path.join(self.tmp.name, "no-such-dir", + "test.db")) + + def test_project_repo_url_db_failure_none(self): + # 3333-3334：projects 表查库失败时回 None，不炸派发。 + self.assertIsNone(self._bad_db_worker()._project_repo_url("p1")) + + def test_project_repo_url_row_missing_key_none(self): + # 3339-3340：行里缺 repo_url 键（非 Row 行）时回 None。 + class _Cursor: + def fetchone(self): + return {} + + class _Conn: + def execute(self, sql, params=()): + return _Cursor() + + def close(self): + pass + + w = _worker(self.db_path) + with mock.patch.object(worker_mod, "_db_connect", + lambda path: _Conn()): + self.assertIsNone(w._project_repo_url("p1")) + + def test_claim_followup_db_connect_failure_none(self): + # 3583-3584：followup 认领时库连接失败回 None。 + self.assertIsNone( + self._bad_db_worker()._claim_next_followup("t1")) + + def test_claim_followup_missing_table_none(self): + # 3604-3605：task_followups 表不存在时回 None，不炸 run_one。 + fresh = os.path.join(self.tmp.name, "fresh.db") + sqlite3.connect(fresh).close() + w = _worker(fresh) + self.assertIsNone(w._claim_next_followup("t1")) + + def test_finish_followup_db_failure_silent(self): + # 3620-3621：结束 followup 时库失败静默，不抛。 + self._bad_db_worker()._finish_followup("f1", "done") + + def test_convert_to_goal_db_failure_false(self): + # 3650-3651：转 goal 时库失败回 False，不悄悄改状态。 + self.assertFalse( + self._bad_db_worker()._convert_running_task_to_goal( + "t1", "new goal")) + + def test_restore_followup_db_failure_silent(self): + # 3665-3666：放回 followup 时库失败静默，不抛。 + self._bad_db_worker()._restore_followup("f1") + + def test_queue_self_followup_db_failure_false(self): + # 3694-3695：排自有 followup 时库失败回 False。 + w = self._bad_db_worker() + self.assertFalse(w._queue_self_followup( + {"id": "t1", "project_id": "p1"}, "body")) + + def test_coord_child_rows_missing_key_skipped(self): + # 5343-5344：子任务行缺列（非 Row 行）时跳过该行， + # 不炸等齐判断。 + class _Cursor: + def fetchone(self): + return {"id": "c1"} + + class _Conn: + def execute(self, sql, params=()): + return _Cursor() + + def close(self): + pass + + w = _worker(self.db_path) + with mock.patch.object(worker_mod, "_db_connect", + lambda path: _Conn()): + self.assertEqual(w._coord_child_rows(["c1"]), {}) + + def test_record_ledger_attempt_outbox_failure_swallowed(self): + # 5798-5799：台账写失败后 outbox 追加本身也失败时， + # 仍只吞掉、返回 None，不影响任务状态与返回值。 + w = self._bad_db_worker() + w.attempt_ledger_hook = True + import cloud.ledger_outbox as outbox_mod # noqa: E402 + with mock.patch.object( + outbox_mod, "append_failure", + side_effect=RuntimeError("outbox down")): + self.assertIsNone( + w._record_ledger_attempt("t1", "fail", True)) + + +class ProcessWatchdogBranchTestCase(unittest.TestCase): + """进程 / 看门狗分支（基底缺失 2505-2506 / 2892-2893 / + 3853-3854 / 3874-3875 / 3881-3882）。""" + + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.db_path = os.path.join(self.tmp.name, "test.db") + init_db(self.db_path) + + def tearDown(self): + self.tmp.cleanup() + + def test_docker_cancel_watchdog_db_interrupted(self): + # 2505-2506：看门狗查库发现任务已 interrupted 时杀容器。 + killed = [] + w = _worker(self.db_path) + w.executor = SimpleNamespace( + kill_container=lambda name: killed.append(name)) + w.get_task = lambda task_id: {"status": "interrupted"} + stop_event = threading.Event() + w._docker_cancel_watchdog("t1", "c1", stop_event) + self.assertEqual(killed, ["c1"]) + + def test_execute_via_process_kill_failure_swallowed(self): + # 2892-2893：communicate 超时后 proc.kill() 本身抛错时 + # 被吞掉，第二次 communicate 的结果仍正常返回 completed。 + calls = {"kill": 0} + + class _FakeProc: + pid = 999 + returncode = 0 + + def poll(self): + return 0 + + def communicate(self, timeout=None): + if timeout is not None: + raise RuntimeError("communicate timeout") + return ("out", "err") + + def kill(self): + calls["kill"] += 1 + raise RuntimeError("kill failed") + + w = _worker(self.db_path) + w.process_factory = lambda task: _FakeProc() + with mock.patch.object(Worker, "_ensure_gateway", + return_value=(None, None)): + result = w._execute_via_process({"id": "t1"}) + self.assertEqual(calls["kill"], 1) + self.assertEqual(result["status"], "completed") + self.assertEqual(result["stdout"], "out") + self.assertEqual(result["stderr"], "err") + self.assertEqual(result["returncode"], 0) + + def test_marker_auth_close_failure_swallowed(self): + # 3853-3854：权限查完后 store.conn.close() 抛错时被吞掉， + # 权限结论（True）不受影响。 + class _FakeConn: + def close(self): + raise RuntimeError("close failed") + + class _FakeStore: + def __init__(self, db_path): + self.conn = _FakeConn() + + def check_permission(self, creator, project_id, action): + return True + + w = _worker(self.db_path) + with mock.patch.object(worker_mod, "AuthStore", _FakeStore): + self.assertEqual( + w._marker_auth({"created_by": "u1", + "project_id": "p1"}, "read"), + (True, "")) + + def test_task_files_worktree_remote_host(self): + # 3874-3875：worktree 在别机时直接拒绝，不指错目录。 + w = _worker(self.db_path) + path, reason = w._task_files_worktree( + {"executor": "local", "worktree_host": "other-host-xyz", + "id": "t1"}) + self.assertIsNone(path) + self.assertIn("other-host-xyz", reason) + + def test_task_files_worktree_bad_project_falls_through(self): + # 3881-3882：project_id 缺失致安全名解析失败时， + # safe_project 置 None、走本地 worktree 分支，不炸。 + w = _worker(self.db_path) + w.host_py_path = os.path.join(self.tmp.name, "host.py") + path, reason = w._task_files_worktree( + {"executor": "local", "project_id": None, "id": "t1"}) + self.assertIsNone(path) + self.assertIn("没有本地 worktree", reason) + + +class PollWaitBranchTestCase(unittest.TestCase): + """轮询 / 等待分支（基底缺失 3711-3712 / 3720-3721 / + 3728-3729 / 3747 / 3768-3769 / 3774-3775 / 3804-3805 / 3825）。""" + + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.db_path = os.path.join(self.tmp.name, "test.db") + init_db(self.db_path) + + def tearDown(self): + self.tmp.cleanup() + + def _poll(self, timeout=10): + return {"task_id": "T2", "interval_seconds": 1, + "timeout_seconds": timeout, "chars": 100} + + def test_poll_cancelled_before_snapshot(self): + # 3711-3712 + 3747：父任务已取消时不等快照、直接回 + # “读取失败”，不挂起。 + w = _worker(self.db_path) + w._cancel_requested = {"P1"} + result = w._poll_task_snapshot( + {"id": "P1", "project_id": "p1"}, self._poll()) + self.assertEqual(result, "轮询任务 T2：读取失败。") + + def test_poll_db_failure_returns_missing(self): + # 3720-3721 + 3728-3729：心跳与快照查库都失败时， + # 异常被吞掉、回“不存在”，不炸等待循环。 + w = _worker(os.path.join(self.tmp.name, "no-such-dir", + "test.db")) + w.get_task = lambda task_id: None + result = w._poll_task_snapshot( + {"id": "P1", "project_id": "p1"}, self._poll()) + self.assertEqual(result, "轮询任务 T2：不存在（同项目内也查不到）。") + + def test_wait_cancelled(self): + # 3768-3769 + 3825 + 3831：父任务已取消时等待中止， + # 未读到的目标记“读取失败”，末尾附取消说明。 + w = _worker(self.db_path) + w._cancel_requested = {"P1"} + lines = w._wait_task_targets( + {"id": "P1", "project_id": "p1"}, + {"task_ids": ["T2"], "interval_seconds": 1, + "timeout_seconds": 10, "chars": 100}) + self.assertEqual(lines[0], "[等待 1 个任务]") + self.assertIn("等待任务 T2：读取失败。", lines) + self.assertIn("说明：等待期间父任务被取消，等待已中止。", lines) + + def test_wait_db_failure_silent(self): + # 3774-3775 + 3804-3805：心跳与批量快照查库都失败时， + # 异常被吞掉、超时 0 立即收尾，不炸、不挂起。 + w = _worker(os.path.join(self.tmp.name, "no-such-dir", + "test.db")) + w.get_task = lambda task_id: None + lines = w._wait_task_targets( + {"id": "P1", "project_id": "p1"}, + {"task_ids": ["T2"], "interval_seconds": 1, + "timeout_seconds": 0, "chars": 100}) + self.assertEqual(lines[0], "[等待 1 个任务]") + self.assertIn("等待任务 T2：读取失败。", lines) + + +if __name__ == "__main__": + unittest.main()