From 12e72dab9ba73d59413cba3671c8752aee2f3eb7 Mon Sep 17 00:00:00 2001 From: host Date: Sun, 1 Nov 2026 15:31:34 +0000 Subject: [PATCH] code-to10 v4 coverage thickening: 219 new assertion tests (batches A/B1/C/D1-D6) Baseline merge head e5ff26a: 28144 stmts, 2807 missing, exact 90.0263%, margin +7. After thickening (full run): 2350 missing, exact 91.6501%, margin +464. Full suite 3516 items: 3509 pass, 1 failure (pre-existing November date failure, untouched), 6 skipped. No production code changed. --- cloud/tests/test_v2cov_gaps_a.py | 564 ++++++++++++++++++++++++ cloud/tests/test_v2cov_gaps_b1.py | 540 +++++++++++++++++++++++ cloud/tests/test_v2cov_gaps_d1.py | 301 +++++++++++++ cloud/tests/test_v2cov_gaps_d2.py | 576 +++++++++++++++++++++++++ cloud/tests/test_v2cov_gaps_d4.py | 85 ++++ tests/test_v2cov_gaps_c.py | 502 ++++++++++++++++++++++ tests/test_v2cov_gaps_d3.py | 692 ++++++++++++++++++++++++++++++ tests/test_v2cov_gaps_d5.py | 233 ++++++++++ tests/test_v2cov_gaps_d6.py | 104 +++++ 9 files changed, 3597 insertions(+) create mode 100644 cloud/tests/test_v2cov_gaps_a.py create mode 100644 cloud/tests/test_v2cov_gaps_b1.py create mode 100644 cloud/tests/test_v2cov_gaps_d1.py create mode 100644 cloud/tests/test_v2cov_gaps_d2.py create mode 100644 cloud/tests/test_v2cov_gaps_d4.py create mode 100644 tests/test_v2cov_gaps_c.py create mode 100644 tests/test_v2cov_gaps_d3.py create mode 100644 tests/test_v2cov_gaps_d5.py create mode 100644 tests/test_v2cov_gaps_d6.py diff --git a/cloud/tests/test_v2cov_gaps_a.py b/cloud/tests/test_v2cov_gaps_a.py new file mode 100644 index 0000000..8b794f3 --- /dev/null +++ b/cloud/tests/test_v2cov_gaps_a.py @@ -0,0 +1,564 @@ +"""Coverage batch v4-A: evidence modules, handler close/502 branches, +server defensive branches, transplant v05 newline branch. + +Targets are the baseline-missing lines of the v4 merge head, verified +against the subset-coverage intersection (see the v4 summary). Stubbing +discipline: the method under test is never stubbed. Collaborators that +are stubbed are: CostStore / _validate_sub2api_key / +_list_sub2api_models (handler_identity namespace -- the v2 batch patched +cloud.server by mistake after the split, which is why these lines stayed +uncovered), _check_project / _project_exists predicates for the two +project 404 branches (unreachable through the real check, which 404s +earlier when the project row is absent), parse_paths_field for the +upload count-mismatch branch (the real parser raises first, so the +handler's own duplicate check is defensive), os.environ.get for the +_default_executor None branch (a real environ value can never be +None), and the _connect/_init/_ensure helpers for the init_db +Postgres branch (no live PG in this environment). +""" + +import json +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 +from cloud import evidence_binding +from cloud import gate_verification +from cloud import ledger_onboarding +from cloud import ledger_outbox +from cloud.evidence_binding import ( + EvidenceNonceStore, + bind_evidence, + binding_presence, + validate_bound_evidence, +) +from cloud.gate_verification import verify_gate_evidence +from cloud.ledger_onboarding import _table_columns, verify_ledger_ready +from cloud.ledger_outbox import ( + append_failure, + outbox_path_for, + read_failures, + replay_failures, +) +from cloud.server import CloudControlPlaneHandler, init_db + + +def _handler(db_path): + h = CloudControlPlaneHandler.__new__(CloudControlPlaneHandler) + h.server = SimpleNamespace(db_path=db_path) + sent = [] + h._send_json = lambda status, obj, extra_headers=None: sent.append( + (status, obj)) + h.sent = sent + return h + + +def _conn(db_path): + c = sqlite3.connect(db_path) + c.row_factory = sqlite3.Row + return c + + +class _Base(unittest.TestCase): + 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 _h(self, body=None, err=None): + h = _handler(self.db_path) + h._read_json = lambda: (body, err) + return h + + def _project(self, pid="p1", owner="u1"): + from cloud.server import _auth_store + c = _conn(self.db_path) + c.execute( + "INSERT INTO projects(id, name, owner_id, created_at," + " repo_url, auto_pr) VALUES(?, ?, ?, ?, ?, 0)", + (pid, "P", owner, time.time(), None)) + c.commit() + c.close() + with _auth_store(self.db_path) as store: + store.create_project(pid, owner) + + def _task(self, tid="t1", pid="p1", status="completed"): + c = _conn(self.db_path) + c.execute( + "INSERT INTO tasks(id, project_id, prompt, status," + " created_by, created_at, updated_at)" + " VALUES(?, ?, ?, ?, ?, ?, ?)", + (tid, pid, "do work", status, "u1", time.time(), + time.time())) + c.commit() + c.close() + + def _last(self, h): + self.assertTrue(h.sent, "handler sent no response") + return h.sent[-1] + + +class EvidenceBindingGapTestCase(unittest.TestCase): + def test_bind_non_dict_raises(self): + with self.assertRaises(ValueError): + bind_evidence(["not", "a", "dict"]) + + def test_validate_non_dict(self): + self.assertEqual(validate_bound_evidence("x"), + (False, "evidence_binding_missing")) + + def test_validate_both_missing(self): + self.assertEqual(validate_bound_evidence({}), + (False, "evidence_binding_missing")) + + def test_validate_bool_ts(self): + self.assertEqual( + validate_bound_evidence( + {"evidence_ts": True, "evidence_nonce": "n1"}), + (False, "evidence_ts_not_a_number")) + + def test_validate_blank_nonce(self): + self.assertEqual( + validate_bound_evidence( + {"evidence_ts": 1000.0, "evidence_nonce": " "}), + (False, "evidence_nonce_empty")) + + def test_binding_presence_non_dict(self): + self.assertEqual(binding_presence(None), "absent") + + def test_has_exception_returns_false(self): + class _RaisingConn: + def execute(self, *args): + raise RuntimeError("db gone") + store = EvidenceNonceStore(_RaisingConn()) + self.assertFalse(store.has("n1")) + + +class GateVerificationGapTestCase(unittest.TestCase): + def _live(self, ci_checks): + return {"ci_checks": ci_checks, "branch_protection": "protected"} + + def test_reader_non_dict_result(self): + out = verify_gate_evidence( + {}, reader=lambda repo, ref: ["not", "a", "dict"], + repo="r", ref="s") + self.assertFalse(out["checked"]) + self.assertFalse(out["verified"]) + self.assertIn("non-dict", out["reason"]) + + def test_live_checks_non_dict_treated_empty(self): + out = verify_gate_evidence( + {}, reader=lambda repo, ref: self._live("nope"), + repo="r", ref="s") + self.assertTrue(out["checked"]) + self.assertTrue(out["verified"]) + + def test_ci_checks_non_dict_item_skipped(self): + out = verify_gate_evidence( + {"ci_checks": ["oops", {"name": "ci", "status": "success"}]}, + reader=lambda repo, ref: self._live({"ci": "success"}), + repo="r", ref="s") + self.assertTrue(out["checked"]) + self.assertTrue(out["verified"]) + self.assertEqual(out["mismatches"], []) + + def test_ci_checks_nameless_item_skipped(self): + out = verify_gate_evidence( + {"ci_checks": [{"status": "success"}]}, + reader=lambda repo, ref: self._live({}), + repo="r", ref="s") + self.assertTrue(out["checked"]) + self.assertTrue(out["verified"]) + + +class _FakeConn: + def __init__(self, rows=None, exc=None): + self._rows = rows + self._exc = exc + + def execute(self, sql, params=()): + if self._exc is not None: + raise self._exc + result = mock.Mock() + result.fetchall.return_value = self._rows + return result + + +class LedgerOnboardingGapTestCase(unittest.TestCase): + def test_table_columns_exception_returns_none(self): + self.assertIsNone(_table_columns( + _FakeConn(exc=RuntimeError("pg dialect")), "task_attempts")) + + def test_table_columns_malformed_rows_return_none(self): + # Row tuples too short for row[1]: IndexError -> None. + self.assertIsNone(_table_columns(_FakeConn(rows=[(1,)]), + "task_attempts")) + + def _sqlite(self, ledger_cols, marker_cols): + conn = sqlite3.connect(":memory:") + if ledger_cols is not None: + conn.execute("CREATE TABLE task_attempts(%s)" + % ", ".join("%s TEXT" % c for c in ledger_cols)) + if marker_cols is not None: + conn.execute("CREATE TABLE task_attempt_origin_markers(%s)" + % ", ".join("%s TEXT" % c for c in marker_cols)) + return conn + + def test_verify_ledger_missing_column(self): + conn = self._sqlite( + ["task_id", "attempt_no", "outcome", "is_defect", + "defect_category", "created_at"], + ["task_id", "origin", "queued_at"]) + try: + out = verify_ledger_ready(conn) + finally: + conn.close() + self.assertFalse(out["tables_ready"]) + self.assertEqual( + out["missing"], + ["task_attempts missing columns: corrects_attempt_no"]) + + def test_verify_marker_missing_column(self): + conn = self._sqlite( + ["task_id", "attempt_no", "outcome", "is_defect", + "defect_category", "created_at", "corrects_attempt_no"], + ["task_id", "origin"]) + try: + out = verify_ledger_ready(conn) + finally: + conn.close() + self.assertFalse(out["tables_ready"]) + self.assertEqual( + out["missing"], + ["task_attempt_origin_markers missing columns: queued_at"]) + + +class LedgerOutboxGapTestCase(unittest.TestCase): + def test_outbox_path_explicit_env_wins(self): + self.assertEqual( + outbox_path_for("/x/db.sqlite", + env={"CLOUD_LEDGER_FAILURE_OUTBOX": "/o/f.jsonl"}), + "/o/f.jsonl") + + def test_outbox_path_memory_and_empty_none(self): + self.assertIsNone(outbox_path_for(":memory:", env={})) + self.assertIsNone(outbox_path_for("", env={})) + + def test_append_failure_falsy_path(self): + self.assertFalse(append_failure("", "t1", "failure", "boom")) + self.assertFalse(append_failure(None, "t1", "failure", "boom")) + + def test_read_failures_missing_file(self): + self.assertEqual(read_failures("/definitely/not/here.jsonl"), + ([], 0)) + + def test_read_failures_bad_lines_counted(self): + with tempfile.TemporaryDirectory() as d: + path = os.path.join(d, "out.jsonl") + with open(path, "w", encoding="utf-8") as fh: + fh.write('{"not json\n') + fh.write('\n') + fh.write(json.dumps({"task_id": "t1", + "outcome": "weird"}) + "\n") + fh.write(json.dumps([1, 2]) + "\n") + fh.write(json.dumps({"task_id": "t2", + "outcome": "success"}) + "\n") + records, bad = read_failures(path) + self.assertEqual([r["task_id"] for r in records], ["t2"]) + self.assertEqual(bad, 3) + + def test_replay_failed_record_kept_and_rewritten(self): + with tempfile.TemporaryDirectory() as d: + path = os.path.join(d, "out.jsonl") + self.assertTrue(append_failure(path, "t9", "failure", "db")) + def _boom(conn, task_id, outcome, is_defect=False): + raise RuntimeError("still down") + stats = replay_failures(path, None, record_fn=_boom) + self.assertEqual(stats["replayed"], 0) + self.assertEqual(stats["failed"], 1) + self.assertEqual(stats["remaining"], 1) + self.assertEqual(stats["bad_lines"], 0) + records, bad = read_failures(path) + self.assertEqual([r["task_id"] for r in records], ["t9"]) + self.assertEqual(bad, 0) + + def test_replay_directory_path_write_failure_swallowed(self): + with tempfile.TemporaryDirectory() as d: + stats = replay_failures(d, None, + record_fn=lambda *a, **k: None) + self.assertEqual(stats, {"replayed": 0, "failed": 0, + "remaining": 0, "bad_lines": 0}) + + def test_replay_bad_lines_not_rewritten(self): + with tempfile.TemporaryDirectory() as d: + path = os.path.join(d, "out.jsonl") + with open(path, "w", encoding="utf-8") as fh: + fh.write('{"broken\n') + fh.write(json.dumps({"task_id": "t3", + "outcome": "success"}) + "\n") + seen = [] + def _ok(conn, task_id, outcome, is_defect=False): + seen.append(task_id) + stats = replay_failures(path, None, record_fn=_ok) + self.assertEqual(stats["replayed"], 1) + self.assertEqual(stats["bad_lines"], 1) + self.assertEqual(stats["remaining"], 1) + self.assertEqual(seen, ["t3"]) + with open(path, encoding="utf-8") as fh: + self.assertIn('{"broken', fh.read()) + + +class HandlerGapTestCase(_Base): + def test_project_cost_close_failure_swallowed(self): + class _Store: + def __init__(self, db_path): + self._conn = mock.Mock() + self._conn.close.side_effect = RuntimeError("down") + + def dual_column_report_by_project(self, project_id): + return {} + h = self._h() + with mock.patch("cloud.handler_identity.CostStore", _Store): + h._project_cost("p1", {}) + self.assertEqual(self._last(h)[0], 200) + + def test_task_cost_close_failure_swallowed(self): + self._project() + self._task() + class _Store: + def __init__(self, db_path): + self._conn = mock.Mock() + self._conn.close.side_effect = RuntimeError("down") + + def dual_column_report_by_task(self, task_id): + return {} + h = self._h() + with mock.patch("cloud.handler_identity.CostStore", _Store): + h._task_cost("u1", "t1") + self.assertEqual(self._last(h)[0], 200) + + def test_relay_models_sub2api_generic_error_502(self): + h = self._h() + with mock.patch.dict(os.environ, {"CLOUD_GATEWAY": "sub2api"}), \ + mock.patch("cloud.handler_identity._validate_sub2api_key", + side_effect=RuntimeError("net down")): + h._relay_models("u1") + self.assertEqual(self._last(h)[0], 502) + + def test_relay_models_sub2api_list_error_502(self): + h = self._h() + with mock.patch.dict(os.environ, {"CLOUD_GATEWAY": "sub2api"}), \ + mock.patch("cloud.handler_identity._validate_sub2api_key", + return_value=("ok", "key", "sub", "kid")), \ + mock.patch("cloud.handler_identity._list_sub2api_models", + side_effect=RuntimeError("gw down")): + h._relay_models("u1") + self.assertEqual(self._last(h)[0], 502) + + def test_models_overview_close_failure_swallowed(self): + self._project() + conn = mock.Mock() + conn.execute.return_value.fetchall.return_value = [] + conn.close.side_effect = RuntimeError("down") + + class _Store: + def __init__(self, db_path): + self._conn = conn + h = self._h() + with mock.patch("cloud.handler_identity.CostStore", _Store): + h._models_overview("u1") + self.assertEqual(self._last(h)[0], 200) + + def test_create_environment_project_row_gone_404(self): + h = self._h(body={"project_id": "pX", "name": "env1"}) + h._check_project = lambda *args: None + h._create_environment("u1") + code, payload = self._last(h) + self.assertEqual(code, 404) + self.assertEqual(payload["error"], "project not found") + + def test_update_project_row_gone_404(self): + h = self._h(body={"repo_url": ""}) + h._check_project = lambda *args: None + h._update_project("u1", "pX") + code, payload = self._last(h) + self.assertEqual(code, 404) + self.assertEqual(payload["error"], "project not found") + + +class ServerGapTestCase(_Base): + def test_read_json_empty_body(self): + h = _handler(self.db_path) + h._read_body = lambda: b"" + self.assertEqual(h._read_json(), ({}, None)) + + def test_default_executor_none_env_value(self): + real_get = os.environ.get + + def _fake_get(key, default=None): + if key == "CLOUD_DEFAULT_EXECUTOR": + return None + return real_get(key, default) + with mock.patch.object(os.environ, "get", side_effect=_fake_get): + self.assertEqual(server_mod._default_executor(), "docker") + + def _backfill_conn(self, mode, rollback_raises=False): + class _Fake: + def __init__(self): + self.rollback_calls = 0 + self.rollback_raises = rollback_raises + self.mode = mode + + def execute(self, sql, params=()): + text = sql.strip() + if text.startswith("ROLLBACK TO"): + raise RuntimeError("rollback-to down") + if text.startswith("SAVEPOINT"): + if self.mode == "savepoint": + raise RuntimeError("savepoint down") + return mock.Mock() + if "INSERT OR IGNORE INTO user_quotas" in text: + raise RuntimeError("insert down") + return mock.Mock() + + def rollback(self): + self.rollback_calls += 1 + if self.rollback_raises: + raise RuntimeError("rollback down") + return _Fake() + + def test_backfill_savepoint_failure_rollback_ok(self): + conn = self._backfill_conn("savepoint") + self.assertEqual(server_mod._backfill_default_quotas(conn), 0) + self.assertEqual(conn.rollback_calls, 1) + + def test_backfill_savepoint_failure_rollback_raises(self): + conn = self._backfill_conn("savepoint", rollback_raises=True) + self.assertEqual(server_mod._backfill_default_quotas(conn), 0) + self.assertEqual(conn.rollback_calls, 1) + + def test_backfill_insert_failure_rollback_to_failure(self): + conn = self._backfill_conn("insert") + self.assertEqual(server_mod._backfill_default_quotas(conn), 0) + self.assertEqual(conn.rollback_calls, 1) + + def test_backfill_insert_failure_rollback_raises(self): + conn = self._backfill_conn("insert", rollback_raises=True) + self.assertEqual(server_mod._backfill_default_quotas(conn), 0) + self.assertEqual(conn.rollback_calls, 1) + + def test_init_db_postgres_branch(self): + conn = mock.Mock() + with mock.patch.dict(os.environ, + {"CLOUD_DB_BACKEND": "postgres"}), \ + mock.patch("cloud.server._connect", + return_value=conn), \ + mock.patch("cloud.server._init_postgres_schema"), \ + mock.patch("cloud.server.migrate_tasks_table"), \ + mock.patch("cloud.server.migrate_projects_table"), \ + mock.patch("cloud.server._ensure_followups_table"), \ + mock.patch("cloud.server._ensure_continue_messages_table"), \ + mock.patch("cloud.server._ensure_environments_table"), \ + mock.patch("cloud.server._ensure_quotas_table"), \ + mock.patch("cloud.server._backfill_default_quotas"), \ + mock.patch("cloud.server._ensure_gateway_keys"), \ + mock.patch("cloud.server._ensure_uploads_tables"), \ + mock.patch("cloud.server._ensure_task_checkpoints_table"), \ + mock.patch("cloud.server._ensure_task_steps_table"): + init_db("/nonexistent/pg.db") + conn.commit.assert_called_once_with() + + def _multipart(self, boundary, fields, files): + out = b"" + for name, value in fields.items(): + out += (("--%s\r\nContent-Disposition: form-data; " + "name=\"%s\"\r\n\r\n%s\r\n") + % (boundary, name, value)).encode("utf-8") + for part_name, filename, data in files: + out += (("--%s\r\nContent-Disposition: form-data; " + "name=\"%s\"; filename=\"%s\"\r\n\r\n") + % (boundary, part_name, filename)).encode("utf-8") + out += data + b"\r\n" + out += ("--%s--\r\n" % boundary).encode("utf-8") + return out + + def test_create_upload_paths_files_mismatch(self): + boundary = "TESTBOUNDARY" + raw = self._multipart( + boundary, {"paths": '["a.txt", "b.txt"]'}, + [("file0", "a.txt", b"hello"), ("file1", "b.txt", b"world")]) + h = _handler(self.db_path) + h.headers = {"Content-Type": + "multipart/form-data; boundary=" + boundary, + "Content-Length": str(len(raw))} + h._read_body = lambda: raw + with mock.patch("cloud.server.parse_paths_field", + return_value=["only-one.txt"]): + h._create_upload("u1") + code, payload = h.sent[-1] + self.assertEqual(code, 400) + self.assertIn("不一致", payload["error"]) + + def test_release_task_bound_bad_ts_400(self): + self._task() + 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, + "evidence_ts": "not-a-number", + "evidence_nonce": "n1", + } + h = self._h(body=body) + with mock.patch.dict(os.environ, + {"CODEX_GLOBAL_ADMINS": "admin1"}): + h._release_task("admin1", "t1") + code, payload = self._last(h) + self.assertEqual(code, 400) + self.assertEqual(payload["error"], "evidence_ts_not_a_number") + + +class TransplantGapTestCase(unittest.TestCase): + def test_transform_v05_notify_with_trailing_newline(self): + from importlib import util as _util + script = (REPO_ROOT / "cloud" / "skills" / "software-engineering" + / "tools" / "check_transplant.py") + spec = _util.spec_from_file_location("check_transplant_v4a", + str(script)) + module = _util.module_from_spec(spec) + spec.loader.exec_module(module) + lines = [ + "前言。" + module.AUDIT_WARNING, + module.TIGHTEN3_NOTIFY, + "filler line after notify", + module.QA_HEADER, + "qa body", + ] + out_lines, stopped = module.transform_main(lines, "v05") + text = "\n".join(out_lines) + self.assertNotIn(module.TIGHTEN3_NOTIFY, text) + self.assertTrue(any("QA 全链路" in s for s in stopped)) + self.assertTrue(any("收紧三" in s for s in stopped)) + + +if __name__ == "__main__": + unittest.main() diff --git a/cloud/tests/test_v2cov_gaps_b1.py b/cloud/tests/test_v2cov_gaps_b1.py new file mode 100644 index 0000000..3a8419e --- /dev/null +++ b/cloud/tests/test_v2cov_gaps_b1.py @@ -0,0 +1,540 @@ +"""Coverage batch v4-B1: cloud/worker.py small pure/error branches. + +Stubbing discipline: the method under test is never stubbed. Patched +collaborators are: register_worker / heartbeat_worker (registry sinks), +AuthStore (audit sink), AutoMemoryStore / _auto_prefix / _todo_list_db / +_db_connect (data collaborators), log_event (only to drive the +log-failure swallow branches), os.killpg (to drive the kill-error +branches without depending on a real process group dying mid-test), +and instance-level _host_state_path_for_task / _rebuild... / +_task_never_started / _fetch_environment predicates where the real +path cannot produce the branch without a race or a missing row. +Every assertion reads a real return value or a real DB row. +""" + +import os +import signal +import sqlite3 +import subprocess +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)) + +from cloud.server import init_db +from cloud.worker import Worker, _code_signature + +FAKE_ENV = {"XCM_API_KEY": "test-placeholder"} + + +def _worker(db_path, **kwargs): + return Worker(db_path, "/fake/host.py", env=FAKE_ENV, **kwargs) + + +class _Base(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) + + def tearDown(self): + self.tmp.cleanup() + + def _insert_task(self, task_id="t1", status="queued", **cols): + conn = sqlite3.connect(self.db_path) + try: + conn.execute( + "INSERT INTO tasks(id, project_id, prompt, status," + " created_by, created_at, updated_at)" + " VALUES(?, ?, ?, ?, ?, ?, ?)", + (task_id, "p1", "do work", status, "u1", time.time(), + time.time())) + for key, value in cols.items(): + conn.execute( + "UPDATE tasks SET %s = ? WHERE id = ?" % key, + (value, task_id)) + conn.commit() + finally: + conn.close() + + def _row(self, task_id="t1"): + conn = sqlite3.connect(self.db_path) + try: + conn.row_factory = sqlite3.Row + return conn.execute( + "SELECT * FROM tasks WHERE id = ?", (task_id,)).fetchone() + finally: + conn.close() + + +class InitAndRegistryTestCase(_Base): + def test_code_signature_stat_failure_none(self): + with mock.patch.object(os, "stat", side_effect=OSError("gone")): + self.assertIsNone(_code_signature("/nonexistent/host.py")) + + def test_init_router_type_error(self): + with self.assertRaises(ValueError): + _worker(self.db_path, router="not-a-router") + + def test_init_fallback_models_normalized(self): + w = _worker(self.db_path, model="m0", + fallback_models=[123, " m1 ", "m1", "m0", "m2"]) + self.assertEqual(w.fallback_models, ["m1", "m2"]) + + def test_init_executor_budget_flag_failure_swallowed(self): + class _BadExecutor: + @property + def ignore_host_budget(self): + return False + + @ignore_host_budget.setter + def ignore_host_budget(self, value): + raise RuntimeError("read-only") + w = _worker(self.db_path, executor=_BadExecutor()) + self.assertIsInstance(w.executor, _BadExecutor) + + def test_register_failure_swallowed(self): + w = _worker(self.db_path) + with mock.patch("cloud.worker.register_worker", + side_effect=RuntimeError("db down")): + w.register() + + def test_registry_heartbeat_failure_swallowed(self): + w = _worker(self.db_path) + with mock.patch("cloud.worker.heartbeat_worker", + side_effect=RuntimeError("db down")): + w._registry_heartbeat() + + def test_close_cost_store_failure_swallowed(self): + w = _worker(self.db_path) + store = mock.Mock() + store._conn.close.side_effect = RuntimeError("down") + w._cost_store = store + w.close() + + +class StepAndCheckpointTestCase(_Base): + def test_mark_interrupted_reason_db_failure_swallowed(self): + w = _worker(self.tmp.name) + w._mark_interrupted_reason("t1", "quota") + + def test_record_step_conn_failure_swallowed(self): + w = _worker(self.db_path) + conn = mock.Mock() + conn.execute.side_effect = RuntimeError("db down") + w._record_step(conn, "t1", "round", "hello") + + def test_checkpoint_round_real_and_failure(self): + w = _worker(self.db_path) + conn = sqlite3.connect(self.db_path) + try: + conn.execute( + "CREATE TABLE IF NOT EXISTS task_checkpoints(" + "task_id TEXT, round_no INTEGER, thread_id TEXT," + " goal_status TEXT, state_sha TEXT, verify_out TEXT," + " created_at DOUBLE PRECISION," + " PRIMARY KEY(task_id, round_no))") + conn.execute("INSERT INTO task_checkpoints VALUES" + "('t1', 2, 'th', 'ok', 'sha', '', 0)") + conn.execute("INSERT INTO task_checkpoints VALUES" + "('t1', 5, 'th', 'ok', 'sha', '', 0)") + conn.commit() + self.assertEqual(w._checkpoint_round(conn, "t1"), 5) + self.assertEqual(w._checkpoint_round(conn, "nobody"), -1) + finally: + conn.close() + bad = mock.Mock() + bad.execute.side_effect = RuntimeError("db down") + self.assertEqual(w._checkpoint_round(bad, "t1"), -1) + + def test_checkpoint_mirror_loop_exception_swallowed(self): + w = _worker(self.db_path) + calls = [] + w._mirror_checkpoint_once = lambda task: (_ for _ in ()).throw( + RuntimeError("mirror down")) + class _Stop: + def __init__(self): + self.n = 0 + + def wait(self, timeout): + self.n += 1 + return self.n >= 2 + w._checkpoint_mirror_loop({"id": "t1"}, _Stop()) + + def test_host_state_path_mounts_tuple(self): + w = _worker(self.db_path) + w.executor = SimpleNamespace(extra_mounts=[("/srv/work", + "/app/work")], + config_path=None) + self.assertEqual( + w._host_state_path_for_task({"project_id": "p1", "id": "t1"}), + "/srv/work/host-state.json") + + def test_host_state_path_non_tuple_mount_and_none_task(self): + w = _worker(self.db_path) + w.executor = SimpleNamespace(extra_mounts=[123], config_path=None) + self.assertEqual( + w._host_state_path_for_task({"project_id": "p1", "id": "t1"}), + "/opt/codex-app-server/work/host-state.json") + self.assertIsNone(w._host_state_path_for_task(None)) + + def test_mirror_checkpoint_once_bad_json(self): + w = _worker(self.db_path) + work = os.path.join(self.tmp.name, "work") + os.makedirs(work, exist_ok=True) + with open(os.path.join(work, "host-state.json"), "w") as fh: + fh.write("{not json") + w.executor = SimpleNamespace(extra_mounts=[(work, "/app/work")], + config_path=None) + self.assertEqual( + w._mirror_checkpoint_once({"project_id": "p1", "id": "t1"}), -1) + + def test_import_host_steps_task_none_swallowed(self): + w = _worker(self.db_path) + conn = sqlite3.connect(self.db_path) + try: + w._import_host_steps(conn, None, "/tmp/x/host-state.json") + finally: + conn.close() + + def test_rebuild_trow_missing(self): + w = _worker(self.db_path) + conn = sqlite3.connect(self.db_path) + try: + conn.execute( + "CREATE TABLE IF NOT EXISTS task_checkpoints(" + "task_id TEXT, round_no INTEGER, thread_id TEXT," + " goal_status TEXT, state_sha TEXT, verify_out TEXT," + " created_at DOUBLE PRECISION," + " PRIMARY KEY(task_id, round_no))") + conn.execute("INSERT INTO task_checkpoints VALUES" + "('ghost', 3, 'th', 'ok', 'sha', '', 0)") + conn.commit() + finally: + conn.close() + self.assertFalse(w._rebuild_host_state_from_checkpoint( + {"id": "ghost", "project_id": "p1"})) + + def test_rebuild_state_path_none(self): + self._insert_task() + w = _worker(self.db_path) + conn = sqlite3.connect(self.db_path) + try: + conn.execute( + "CREATE TABLE IF NOT EXISTS task_checkpoints(" + "task_id TEXT, round_no INTEGER, thread_id TEXT," + " goal_status TEXT, state_sha TEXT, verify_out TEXT," + " created_at DOUBLE PRECISION," + " PRIMARY KEY(task_id, round_no))") + conn.execute("INSERT INTO task_checkpoints VALUES" + "('t1', 3, 'th', 'ok', 'sha', '', 0)") + conn.commit() + finally: + conn.close() + w._host_state_path_for_task = lambda task: None + self.assertFalse(w._rebuild_host_state_from_checkpoint( + {"id": "t1", "project_id": "p1"})) + + def test_rebuild_already_present_true(self): + self._insert_task() + w = _worker(self.db_path) + conn = sqlite3.connect(self.db_path) + try: + conn.execute( + "CREATE TABLE IF NOT EXISTS task_checkpoints(" + "task_id TEXT, round_no INTEGER, thread_id TEXT," + " goal_status TEXT, state_sha TEXT, verify_out TEXT," + " created_at DOUBLE PRECISION," + " PRIMARY KEY(task_id, round_no))") + conn.execute("INSERT INTO task_checkpoints VALUES" + "('t1', 3, 'th', 'ok', 'sha', '', 0)") + conn.commit() + finally: + conn.close() + work = os.path.join(self.tmp.name, "work") + os.makedirs(work, exist_ok=True) + with open(os.path.join(work, "host-state.json"), "w") as fh: + import json as _json + _json.dump({"version": 1, "tasks": {"t1": {"goal": "g"}}}, fh) + w.executor = SimpleNamespace(extra_mounts=[(work, "/app/work")], + config_path=None) + self.assertTrue(w._rebuild_host_state_from_checkpoint( + {"id": "t1", "project_id": "p1"})) + + def test_rebuild_db_failure_false(self): + w = _worker(self.tmp.name) + self.assertFalse(w._rebuild_host_state_from_checkpoint( + {"id": "t1", "project_id": "p1"})) + + def test_never_started_state_path_none(self): + w = _worker(self.db_path) + w._host_state_path_for_task = lambda task: None + self.assertFalse(w._task_never_started( + {"id": "t1", "project_id": "p1"})) + + def test_resolve_breakpoint_log_failure_swallowed(self): + w = _worker(self.db_path) + w._rebuild_host_state_from_checkpoint = lambda task: (_ for _ in ( + )).throw(RuntimeError("rebuild down")) + w._task_never_started = lambda task: True + with mock.patch("cloud.worker.log_event", + side_effect=RuntimeError("log down")): + out = w._resolve_breakpoint_resume( + {"id": "t1", "project_id": "p1"}) + self.assertFalse(out["_goal_resume_breakpoint"]) + + +class EnsureColumnsTestCase(_Base): + def _fake(self, mode): + all_cols = ["result_text", "result_text_full", "error_text", + "actual_model", "started_at", "finished_at", + "child_pid", "executor", "pr_url", "pr_number", "goal", + "verify_command", "goal_max_rounds", "goal_mode", + "worktree_host", "claimed_by_worker", "model", "effort", + "child_model", "child_effort", "approval", "plan", + "forked_from", "environment_id", "resume_message", + "turn_id", "root_turn_id", "coordinator_parts", + "parent_task_id", "coordinator_part", + "goal_resume_pending", "paused_by_user", + "interrupted_reason", "auto_resume_count", + "no_time_limit", "resume_after", + "auto_resume_stall_count", + "checkpoint_round_at_last_resume", + "last_error_signature"] + + class _Fake: + def execute(self, sql, params=()): + if sql.strip().startswith("PRAGMA"): + rows = [(i, name, "TEXT", 0, None, 0) + for i, name in enumerate( + all_cols if mode == "create" else ["id"])] + result = mock.Mock() + result.fetchall.return_value = rows + return result + if sql.strip().startswith("ALTER"): + raise RuntimeError("alter down") + if sql.strip().startswith("CREATE"): + if mode == "create": + raise RuntimeError("create down") + return mock.Mock() + return mock.Mock() + return _Fake() + + def test_ensure_columns_pragma_failure(self): + w = _worker(self.db_path) + bad = mock.Mock() + bad.execute.side_effect = RuntimeError("pragma down") + w._ensure_result_columns(bad) + + def test_ensure_columns_alter_failure(self): + w = _worker(self.db_path) + w._ensure_result_columns(self._fake("alter")) + + def test_ensure_columns_create_failure(self): + w = _worker(self.db_path) + w._ensure_result_columns(self._fake("create")) + + +class WriteResultTestCase(_Base): + def test_write_result_child_pid_branch(self): + self._insert_task() + w = _worker(self.db_path) + self.assertTrue(w._write_execution_result( + "t1", result_text="done", child_pid=99)) + self.assertEqual(self._row()["child_pid"], 99) + + def test_write_result_old_column_fallback(self): + self._insert_task() + w = _worker(self.db_path) + w._ensure_result_columns = lambda conn: None + real = sqlite3.connect(self.db_path) + + class _Flaky: + def __init__(self): + self.flipped = False + + def execute(self, sql, params=()): + if "result_text_full" in sql and not self.flipped: + self.flipped = True + raise sqlite3.OperationalError( + "no such column: result_text_full") + return real.execute(sql, params) + + def rollback(self): + raise RuntimeError("rollback down") + + def commit(self): + return real.commit() + + def close(self): + return real.close() + with mock.patch("cloud.worker._db_connect", + return_value=_Flaky()): + self.assertTrue(w._write_execution_result( + "t1", result_text="fallback-ok")) + self.assertEqual(self._row()["result_text"], "fallback-ok") + + +class KillAuditNotifyTestCase(_Base): + def test_kill_proc_group_term_error(self): + w = _worker(self.db_path) + proc = subprocess.Popen(["sleep", "30"], start_new_session=True) + try: + with mock.patch("os.killpg", + side_effect=ProcessLookupError("gone")): + w._kill_proc_group(proc) + self.assertIsNone(proc.poll()) + finally: + proc.kill() + proc.wait() + + def test_kill_proc_group_sigkill_error(self): + w = _worker(self.db_path) + proc = subprocess.Popen(["sleep", "30"], start_new_session=True) + try: + def _killpg(pgid, sig): + if sig == signal.SIGKILL: + raise ProcessLookupError("gone") + return None + with mock.patch("os.killpg", side_effect=_killpg): + w._kill_proc_group(proc) + self.assertIsNone(proc.poll()) + finally: + proc.kill() + proc.wait() + + def test_audit_close_failure_swallowed(self): + w = _worker(self.db_path) + + class _Store: + def __init__(self, db_path): + self.conn = mock.Mock() + self.conn.close.side_effect = RuntimeError("down") + + def audit(self, *args): + return None + with mock.patch("cloud.worker.AuthStore", _Store): + w._audit("p1", "act") + + def test_notify_errors_and_exception_swallowed(self): + w = _worker(self.db_path) + with mock.patch("cloud.push.dispatch_event", + return_value={"errors": ["e1"]}): + w._notify_task_finished("t1", "p1", "completed") + with mock.patch("cloud.push.dispatch_event", + side_effect=RuntimeError("push down")): + w._notify_task_finished("t1", "p1", "failed") + + +class InjectionTestCase(_Base): + def test_auto_memory_no_owner(self): + w = _worker(self.db_path) + task = {"id": "t1", "project_id": "p1", "prompt": "p"} + self.assertIs(w._apply_auto_memory_injection(task), task) + + def test_auto_memory_store_and_log_failure(self): + w = _worker(self.db_path) + w.logger = mock.Mock() + w.logger.warning.side_effect = RuntimeError("log down") + task = {"id": "t1", "project_id": "p1", "prompt": "p", + "created_by": "u1"} + with mock.patch("cloud.worker.AutoMemoryStore", + side_effect=RuntimeError("store down")): + self.assertIs(w._apply_auto_memory_injection(task), task) + + def test_auto_memory_goal_branch(self): + w = _worker(self.db_path) + + class _Store: + def __init__(self, db_path): + pass + + def get_task_flags(self, task_id): + return {} + + def close(self): + pass + task = {"id": "t1", "project_id": "p1", "prompt": "P", + "goal": "G", "goal_mode": 1, "created_by": "u1"} + with mock.patch("cloud.worker.AutoMemoryStore", _Store), \ + mock.patch("cloud.worker._auto_prefix", + return_value="MEM"): + out = w._apply_auto_memory_injection(task) + self.assertEqual(out["goal"], "MEM\n\nG") + self.assertEqual(out["prompt"], "MEM\n\nP") + self.assertTrue(out["_memory_injected"]) + self.assertEqual(task["goal"], "G") + + def test_todo_injection_db_and_log_failure(self): + w = _worker(self.tmp.name) + w.logger = mock.Mock() + w.logger.warning.side_effect = RuntimeError("log down") + task = {"id": "t1", "project_id": "p1", "prompt": "p"} + self.assertIs(w._apply_todo_injection(task), task) + + def test_todo_injection_over_twenty(self): + w = _worker(self.db_path) + todos = [{"id": "td%d" % i, "title": "T%d" % i, + "status": "pending"} for i in range(21)] + task = {"id": "t1", "project_id": "p1", "prompt": "p"} + with mock.patch("cloud.worker._todo_list_db", + return_value=todos): + out = w._apply_todo_injection(task) + self.assertTrue(out["_todo_injected"]) + self.assertIn("还有 1 条未列出", out["prompt"]) + + def test_project_injection_db_and_log_failure(self): + w = _worker(self.tmp.name) + w.logger = mock.Mock() + w.logger.warning.side_effect = RuntimeError("log down") + task = {"id": "t1", "project_id": "p1", "prompt": "p"} + self.assertIs(w._apply_project_injection(task), task) + + def test_project_injection_row_key_errors(self): + w = _worker(self.db_path) + + class _BadRow: + def __getitem__(self, key): + raise KeyError(key) + + class _FakeConn: + def execute(self, sql, params=()): + result = mock.Mock() + result.fetchone.return_value = _BadRow() + return result + + def close(self): + pass + task = {"id": "t1", "project_id": "p1", "prompt": "p"} + with mock.patch("cloud.worker._db_connect", + return_value=_FakeConn()): + out = w._apply_project_injection(task) + self.assertTrue(out["_project_injected"]) + self.assertIn("项目：p1（id=p1）", out["prompt"]) + self.assertIn("仓库：未绑定", out["prompt"]) + + def test_project_injection_env_missing(self): + conn = sqlite3.connect(self.db_path) + try: + conn.execute( + "INSERT INTO projects(id, name, owner_id, created_at," + " repo_url, auto_pr) VALUES('p1', 'Proj', 'u1', 0, NULL, 0)") + conn.commit() + finally: + conn.close() + w = _worker(self.db_path) + w._fetch_environment = lambda env_id: None + task = {"id": "t1", "project_id": "p1", "prompt": "p", + "environment_id": "envX"} + out = w._apply_project_injection(task) + self.assertIn("envX（查不到该环境）", out["prompt"]) + + +if __name__ == "__main__": + unittest.main() diff --git a/cloud/tests/test_v2cov_gaps_d1.py b/cloud/tests/test_v2cov_gaps_d1.py new file mode 100644 index 0000000..9bd6e24 --- /dev/null +++ b/cloud/tests/test_v2cov_gaps_d1.py @@ -0,0 +1,301 @@ +"""Coverage batch v4-D1: cloud/worker.py resume/checkpoint/notify branches. + +Stubbing discipline: the method under test is never stubbed. Patched +collaborators are: _quota_exceeded / _attempt_mark / _db_connect / +dispatch_event / AutoMemoryStore / the instance logger (data and sink +collaborators), plus instance-level _mirror_checkpoint_once where the +real path cannot raise without a race. Every assertion reads a real +return value or a real DB row. +""" + +import json +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)) + +from cloud.server import init_db +from cloud.worker import Worker + +FAKE_ENV = {"XCM_API_KEY": "test-placeholder"} + + +def _worker(db_path, **kwargs): + return Worker(db_path, "/fake/host.py", env=FAKE_ENV, **kwargs) + + +class _Base(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) + + def tearDown(self): + self.tmp.cleanup() + + def _insert_task(self, task_id="t1", status="queued", **cols): + conn = sqlite3.connect(self.db_path) + try: + conn.execute( + "INSERT INTO tasks(id, project_id, prompt, status," + " created_by, created_at, updated_at)" + " VALUES(?, ?, ?, ?, ?, ?, ?)", + (task_id, "p1", "do work", status, "u1", time.time(), + time.time())) + for key, value in cols.items(): + conn.execute( + "UPDATE tasks SET %s = ? WHERE id = ?" % key, + (value, task_id)) + conn.commit() + finally: + conn.close() + + def _row(self, task_id="t1"): + conn = sqlite3.connect(self.db_path) + try: + conn.row_factory = sqlite3.Row + return conn.execute( + "SELECT * FROM tasks WHERE id = ?", (task_id,)).fetchone() + finally: + conn.close() + + def _fake_executor(self, state_dir): + cfg = os.path.join(state_dir, "config.json") + with open(cfg, "w", encoding="utf-8") as fh: + json.dump({"state": "host-state.json"}, fh) + return SimpleNamespace(extra_mounts=[], config_path=cfg) + + +class AutoResumeExceptionTestCase(_Base): + def test_quota_check_raises_stays_interrupted(self): + # 1093-1094: _quota_exceeded raising inside auto_resume_interrupted + # must be swallowed into "quota_check_failed" and the task skipped. + self._insert_task(status="interrupted", goal_mode=1, + interrupted_reason="quota") + w = _worker(self.db_path) + with mock.patch.object( + Worker, "_quota_exceeded", + side_effect=RuntimeError("quota db down")): + resumed = w.auto_resume_interrupted() + self.assertNotIn("t1", resumed) + self.assertEqual(self._row()["status"], "interrupted") + + def test_attempt_mark_failure_swallowed_on_resume(self): + # 1163-1164: _attempt_mark raising after a resume must not block + # the resume itself. + self._insert_task(status="interrupted", goal_mode=1, + interrupted_reason="worker_shutdown") + w = _worker(self.db_path, attempt_ledger_hook=True) + with mock.patch("cloud.worker._attempt_mark", + side_effect=RuntimeError("ledger down")): + resumed = w.auto_resume_interrupted() + self.assertIn("t1", resumed) + self.assertEqual(self._row()["status"], "queued") + + def test_db_connect_failure_returns_empty(self): + # 1182-1183: outer except of auto_resume_interrupted. + w = _worker(self.db_path) + with mock.patch("cloud.worker._db_connect", + side_effect=RuntimeError("db down")): + self.assertEqual(w.auto_resume_interrupted(), []) + + def test_checkpoint_select_failure_returns_minus_one(self): + # 1102-1103: checkpoint SELECT raising inside auto_resume must + # fall back to cp_round=-1 and still resume a platform task. + # NOTE: dropping the table cannot reach this branch, because + # _ensure_result_columns recreates task_checkpoints before the + # SELECT; the branch is reached only when the SELECT itself + # raises, so the connection is faked for that statement only. + self._insert_task(status="interrupted", goal_mode=1, + interrupted_reason="worker_shutdown") + w = _worker(self.db_path) + w._ensure_result_columns = lambda conn: None + row = ("t1", "p1", "u1", None, "m0", "worker_shutdown", None, + 0, None, 1, None, None) + + class _Cursor: + def __init__(self, rows=None, rowcount=1): + self._rows = rows or [] + self.rowcount = rowcount + + def fetchall(self): + return self._rows + + def fetchone(self): + return self._rows[0] if self._rows else None + + class _FakeConn: + def execute(self, sql, params=()): + if "task_checkpoints" in sql: + raise RuntimeError("checkpoint read down") + if "FROM tasks" in sql and "SELECT" in sql: + return _Cursor(rows=[row]) + return _Cursor() + + def commit(self): + pass + + def close(self): + pass + + with mock.patch("cloud.worker._db_connect", + return_value=_FakeConn()): + resumed = w.auto_resume_interrupted() + self.assertIn("t1", resumed) + + def test_checkpoint_round_and_mirror_loop_exceptions(self): + w = _worker(self.db_path) + conn = sqlite3.connect(self.db_path) + try: + self.assertEqual(w._checkpoint_round(conn, "nope"), -1) + finally: + conn.close() + bad = mock.Mock() + bad.execute.side_effect = RuntimeError("boom") + self.assertEqual(w._checkpoint_round(bad, "nope"), -1) + # 1230-1231: first mirror call raising, stop already set. + w._mirror_checkpoint_once = mock.Mock( + side_effect=RuntimeError("mirror down")) + stop = threading.Event() + stop.set() + w._checkpoint_mirror_loop({"id": "t1"}, stop) + # 1233-1236: loop-body mirror call raising, then stop. + calls = {"n": 0} + + def _wait(timeout=None): + calls["n"] += 1 + return calls["n"] > 1 + + class _Stop: + def wait(self, timeout=None): + return _wait(timeout) + + w._checkpoint_mirror_loop({"id": "t1"}, _Stop()) + + +class HostStatePathTestCase(_Base): + def test_non_tuple_mount_item_falls_through(self): + # 1256: a non-tuple mounts entry yields (None, None) and the + # loop falls through to the /opt fallback. + w = _worker(self.db_path, executor=SimpleNamespace( + extra_mounts=["not-a-tuple"], config_path=None)) + path = w._host_state_path_for_task({"project_id": "p1"}) + self.assertEqual(path, "/opt/codex-app-server/work/host-state.json") + + def test_mirror_missing_state_returns_minus_one(self): + # 1286: no state file at all. + state_dir = os.path.join(self.tmp.name, "cfg") + os.makedirs(state_dir) + w = _worker(self.db_path, + executor=self._fake_executor(state_dir)) + self.assertEqual( + w._mirror_checkpoint_once({"id": "t1", "project_id": "p1"}), -1) + + def test_rebuild_from_checkpoint_missing_state_file(self): + # 1411: state file absent -> fresh {"version": 1, "tasks": {}} + # is built and the rebuilt entry is written back. + self._insert_task(status="interrupted", verify_command="true", + goal_max_rounds=5) + conn = sqlite3.connect(self.db_path) + try: + conn.execute( + "INSERT INTO task_checkpoints(task_id, round_no," + " thread_id, goal_status) VALUES(?, ?, ?, ?)", + ("t1", 2, "th1", "interrupted")) + conn.commit() + finally: + conn.close() + state_dir = os.path.join(self.tmp.name, "cfg2") + os.makedirs(state_dir) + w = _worker(self.db_path, + executor=self._fake_executor(state_dir)) + self.assertTrue(w._rebuild_host_state_from_checkpoint( + {"id": "t1", "project_id": "p1"})) + with open(os.path.join(state_dir, "host-state.json"), + encoding="utf-8") as fh: + state = json.load(fh) + self.assertEqual(state["tasks"]["t1"]["goal_rounds_done"], 2) + self.assertEqual(state["tasks"]["t1"]["thread_id"], "th1") + self.assertTrue(state["tasks"]["t1"]["rebuilt_from_checkpoint"]) + + def test_record_step_failure_swallowed(self): + # 1208-1209: step INSERT raising is logged, never raised. + w = _worker(self.db_path) + bad = mock.Mock() + bad.execute.side_effect = RuntimeError("db down") + w._record_step(bad, "t1", "round", "summary") + bad.execute.assert_called() + + +class WriteResultTestCase(_Base): + def test_child_pid_branch_writes_full(self): + # 1664-1670, 1672-1673, 1675: child_pid is not None branch. + self._insert_task(status="running") + w = _worker(self.db_path) + self.assertTrue(w._write_execution_result( + "t1", result_text="out", result_full="full-out", + child_pid=1234, turn_id="turn-1")) + row = self._row() + self.assertEqual(row["child_pid"], 1234) + self.assertEqual(row["result_text_full"], "full-out") + self.assertEqual(row["turn_id"], "turn-1") + + def test_write_result_non_column_error_reraised(self): + # 1686: an execute error unrelated to result_text_full is raised. + self._insert_task(status="running") + w = _worker(self.db_path) + w._ensure_result_columns = lambda conn: None + + class _BadConn: + def execute(self, *args, **kwargs): + raise ValueError("boom") + + def rollback(self): + pass + + def commit(self): + pass + + def close(self): + pass + + with mock.patch("cloud.worker._db_connect", + return_value=_BadConn()): + with self.assertRaises(ValueError): + w._write_execution_result("t1", result_text="out") + + +class NotifyAndMemoryTestCase(_Base): + def test_notify_errors_branch_logged(self): + # 1784-1786: dispatch_event returning errors is logged. + w = _worker(self.db_path) + with mock.patch("cloud.push.dispatch_event", + return_value={"errors": ["sub down"]}): + w._notify_task_finished("t1", "p1", "completed") + # No exception, no state change to assert beyond survival. + + def test_auto_memory_logger_failure_swallowed(self): + # 1823: logger.warning itself raising inside the injection + # failure path is swallowed and the task is returned unchanged. + w = _worker(self.db_path) + task = {"id": "t1", "project_id": "p1", "created_by": "u1", + "prompt": "do work"} + w.logger = mock.Mock() + w.logger.warning.side_effect = RuntimeError("logger down") + with mock.patch("cloud.worker.AutoMemoryStore", + side_effect=RuntimeError("store down")): + out = w._apply_auto_memory_injection(task) + self.assertIs(out, task) + + +if __name__ == "__main__": + unittest.main() diff --git a/cloud/tests/test_v2cov_gaps_d2.py b/cloud/tests/test_v2cov_gaps_d2.py new file mode 100644 index 0000000..bf4f1bb --- /dev/null +++ b/cloud/tests/test_v2cov_gaps_d2.py @@ -0,0 +1,576 @@ +"""Coverage batch v4-D2: cloud/worker.py tree/quota/cancel/watchdog/ +cleanup/gateway/process branches. + +Stubbing discipline: the method under test is never stubbed. Patched +collaborators are: _tenant_safe_name / _tenant_dirs / _project_repo_url +/ _quota_limit_and_used / _list_gateway_models / _gateway_name / +_gateway_origin / ready_tasks / orphaned_containers / CostStore / +_db_connect / dispatch sinks / instance-level _kill_proc_group / +_set_child_pid / get_task / heartbeat (data, sink and predicate +collaborators). Every assertion reads a real return value or a real +DB row, except watchdog/cleanup tests where the assertion is the +returned list or the absence of a raise plus a real side-effect set. +""" + +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)) + +from cloud.server import init_db +from cloud.worker import Worker + +FAKE_ENV = {"XCM_API_KEY": "test-placeholder"} + + +def _worker(db_path, **kwargs): + return Worker(db_path, "/fake/host.py", env=FAKE_ENV, **kwargs) + + +class _Base(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) + + def tearDown(self): + self.tmp.cleanup() + + def _insert_task(self, task_id="t1", status="queued", **cols): + conn = sqlite3.connect(self.db_path) + try: + conn.execute( + "INSERT INTO tasks(id, project_id, prompt, status," + " created_by, created_at, updated_at)" + " VALUES(?, ?, ?, ?, ?, ?, ?)", + (task_id, "p1", "do work", status, "u1", time.time(), + time.time())) + for key, value in cols.items(): + conn.execute( + "UPDATE tasks SET %s = ? WHERE id = ?" % key, + (value, task_id)) + conn.commit() + finally: + conn.close() + + def _row(self, task_id="t1"): + conn = sqlite3.connect(self.db_path) + try: + conn.row_factory = sqlite3.Row + return conn.execute( + "SELECT * FROM tasks WHERE id = ?", (task_id,)).fetchone() + finally: + conn.close() + + +class _StopOnce: + """wait() returns False once, then True.""" + + def __init__(self): + self.n = 0 + + def wait(self, timeout=None): + self.n += 1 + return self.n > 1 + + +class TreeSnapshotTestCase(_Base): + def _docker_worker(self): + # Subclass of the real DockerExecutor (no-op __init__) so the + # isinstance check in _tree_snapshot_source passes without + # patching the module-level class. + from cloud.worker import DockerExecutor as _RealDocker + + class _FakeDocker(_RealDocker): + def __init__(self): + pass + + w = _worker(self.db_path, executor=_FakeDocker()) + w.tenants_root = self.tmp.name + w.tenant_isolation = True + return w + + def test_safe_name_failure_returns_none(self): + # 2003-2004 + w = self._docker_worker() + w._tenant_safe_name = mock.Mock(side_effect=RuntimeError("bad")) + self.assertIsNone(w._tree_snapshot_source({"project_id": "p1"})) + + def test_checkout_exists_returns_checkout(self): + # 2007 + w = self._docker_worker() + w._tenant_safe_name = mock.Mock(return_value="safe") + os.makedirs(os.path.join(self.tmp.name, "safe", "safe")) + self.assertEqual( + w._tree_snapshot_source({"project_id": "p1"}), + os.path.join(self.tmp.name, "safe", "safe")) + + def test_tenant_dirs_failure_returns_none(self): + # 2017-2018 + w = self._docker_worker() + w._tenant_safe_name = mock.Mock(return_value="safe") + w._project_repo_url = mock.Mock(return_value=None) + w._tenant_dirs = mock.Mock(side_effect=RuntimeError("dirs down")) + self.assertIsNone(w._tree_snapshot_source({"project_id": "p1"})) + + def test_test_repo_missing_returns_none(self): + # 2023 + w = self._docker_worker() + w._tenant_safe_name = mock.Mock(return_value="safe") + w._project_repo_url = mock.Mock(return_value=None) + w._tenant_dirs = mock.Mock( + return_value={"test_repo": "/nonexistent/repo-xyz"}) + self.assertIsNone(w._tree_snapshot_source({"project_id": "p1"})) + + def test_injection_source_failure_swallowed(self): + # 2051-2053, 2056 + w = _worker(self.db_path) + w._tree_snapshot_source = mock.Mock( + side_effect=RuntimeError("snap down")) + task = {"id": "t1", "project_id": "p1", "prompt": "do work"} + self.assertIs(w._apply_tree_injection(task), task) + + def test_injection_logger_failure_swallowed(self): + # 2054-2055 + w = _worker(self.db_path) + w._tree_snapshot_source = mock.Mock( + side_effect=RuntimeError("snap down")) + w.logger = mock.Mock() + w.logger.warning.side_effect = RuntimeError("logger down") + task = {"id": "t1", "project_id": "p1", "prompt": "do work"} + self.assertIs(w._apply_tree_injection(task), task) + + +class QuotaTestCase(_Base): + def test_limit_commit_failure_swallowed_unlimited(self): + # 2131-2132: commit raising, explicit-NULL quota row -> None. + conn = sqlite3.connect(self.db_path) + try: + conn.execute( + "INSERT INTO user_quotas(user_id, quota_usd_micros," + " updated_at) VALUES(?, NULL, ?)", ("u1", time.time())) + conn.commit() + finally: + conn.close() + w = _worker(self.db_path) + real = sqlite3.connect(self.db_path) + + class _Proxy: + def __init__(self, conn): + self._conn = conn + + def __getattr__(self, name): + return getattr(self._conn, name) + + def commit(self): + raise RuntimeError("commit down") + + def close(self): + self._conn.close() + + with mock.patch("cloud.worker._db_connect", + return_value=_Proxy(real)): + quota, used, month, unpriced = w._quota_limit_and_used("u1") + self.assertIsNone(quota) + self.assertEqual((used, unpriced), (0, 0)) + + def test_limit_coststore_close_failure_swallowed(self): + # 2142-2143 + from cloud.cost import CostStore + CostStore(self.db_path).close() if hasattr( + CostStore(self.db_path), "close") else None + store = CostStore(self.db_path) + try: + store._conn.close() + except Exception: + pass + w = _worker(self.db_path) + + class _BadStore: + def __init__(self, db_path): + self._conn = mock.Mock() + self._conn.close.side_effect = RuntimeError("close down") + + with mock.patch("cloud.worker.CostStore", _BadStore): + quota, used, month, unpriced = w._quota_limit_and_used("u9") + self.assertIsNotNone(quota) + self.assertEqual((used, unpriced), (0, 0)) + + def test_quota_parent_row_type_error(self): + # 2196-2197 + w = _worker(self.db_path) + + class _Cursor: + def fetchone(self): + return 5 + + class _Conn: + def execute(self, *a, **k): + return _Cursor() + + def close(self): + pass + + with mock.patch("cloud.worker._db_connect", return_value=_Conn()): + msg = w._quota_exceeded({"parent_task_id": "p0"}, None) + self.assertIn("缺少属主", msg) + + def test_quota_parent_lookup_failure(self): + # 2198-2199 + w = _worker(self.db_path) + with mock.patch("cloud.worker._db_connect", + side_effect=RuntimeError("db down")): + msg = w._quota_exceeded({"parent_task_id": "p0"}, None) + self.assertIn("缺少属主", msg) + + def test_quota_limit_failure_logger_failure(self): + # 2213-2214, both on_error flavours + w = _worker(self.db_path) + w._quota_limit_and_used = mock.Mock( + side_effect=RuntimeError("quota db down")) + w.logger = mock.Mock() + w.logger.warning.side_effect = RuntimeError("logger down") + self.assertIn("quota_check_unavailable", + w._quota_exceeded({"created_by": "u1"}, None, + on_error="closed")) + self.assertIn("quota_check_failed", + w._quota_exceeded({"created_by": "u1"}, None)) + + def test_ready_budget_exceeded_empty(self): + # 2252 + w = _worker(self.db_path) + w.usage_turns = 5 + w.max_turns = 1 + self.assertEqual(w._ready_tasks(), []) + + def test_ready_value_error_empty(self): + # 2258-2259 + self._insert_task() + w = _worker(self.db_path) + with mock.patch("cloud.worker.ready_tasks", + side_effect=ValueError("cycle")): + self.assertEqual(w._ready_tasks(), []) + + +class HeartbeatCancelTestCase(_Base): + def test_heartbeat_task_failure_swallowed(self): + # 2330-2331 + w = _worker(self.db_path) + w._running_set("t1") + w.heartbeat = mock.Mock(side_effect=RuntimeError("beat down")) + w._heartbeat_running_tasks() + + def test_cancel_proc_wait_failure_swallowed(self): + # 2349-2350 + self._insert_task(status="running") + w = _worker(self.db_path) + proc = mock.Mock() + proc.wait.side_effect = RuntimeError("wait down") + w._running_set("t1", proc=proc) + w._kill_proc_group = lambda p: None + self.assertTrue(w.cancel("t1")) + self.assertEqual(self._row()["status"], "interrupted") + + def test_cancel_container_kill_failure_swallowed(self): + # 2359-2360 + self._insert_task(status="running") + w = _worker(self.db_path, executor=SimpleNamespace( + kill_container=mock.Mock( + side_effect=RuntimeError("kill down")))) + w._running_set("t1", container="c1") + self.assertTrue(w.cancel("t1")) + self.assertEqual(self._row()["status"], "interrupted") + + +class WatchdogTestCase(_Base): + def test_cancel_watchdog_lookup_failure_continues(self): + # 2407-2408 + w = _worker(self.db_path) + w.get_task = mock.Mock(side_effect=RuntimeError("db down")) + w._docker_cancel_watchdog("t1", "c1", _StopOnce()) + + def test_cancel_watchdog_kill_failure_swallowed(self): + # 2420-2421 + w = _worker(self.db_path, executor=SimpleNamespace( + kill_container=mock.Mock( + side_effect=RuntimeError("kill down")))) + w._cancel_requested.add("t1") + w._docker_cancel_watchdog("t1", "c1", _StopOnce()) + + def test_quota_watchdog_check_failure_continues(self): + # 2446-2447 + w = _worker(self.db_path) + w._docker_quota_watchdog( + "t1", "c1", _StopOnce(), + lambda: (_ for _ in ()).throw(RuntimeError("check down"))) + + def test_quota_watchdog_kill_failure_swallowed(self): + # 2459-2460 + w = _worker(self.db_path, executor=SimpleNamespace( + kill_container=mock.Mock( + side_effect=RuntimeError("kill down")))) + w._docker_quota_watchdog("t1", "c1", _StopOnce(), lambda: True) + self.assertIn("t1", w._quota_hit_tasks) + + +class CleanupTestCase(_Base): + def test_orphan_no_executor_caps_returns_empty(self): + # 2485 + w = _worker(self.db_path) + self.assertEqual(w.cleanup_orphaned_containers(), []) + + def test_orphan_list_failure_returns_empty(self): + # 2488-2491 + w = _worker(self.db_path, executor=SimpleNamespace( + list_containers=mock.Mock( + side_effect=RuntimeError("list down")), + kill_container=mock.Mock())) + self.assertEqual(w.cleanup_orphaned_containers(), []) + + def test_orphan_identify_failure_returns_empty(self): + # 2496-2499 + w = _worker(self.db_path, executor=SimpleNamespace( + list_containers=mock.Mock(return_value=["c1"]), + kill_container=mock.Mock())) + with mock.patch("cloud.worker.orphaned_containers", + side_effect=RuntimeError("identify down")): + self.assertEqual(w.cleanup_orphaned_containers(), []) + + def test_orphan_kill_failure_skipped(self): + # 2509-2510 + kill = mock.Mock(side_effect=RuntimeError("kill down")) + w = _worker(self.db_path, executor=SimpleNamespace( + list_containers=mock.Mock(return_value=["c1"]), + kill_container=kill)) + with mock.patch("cloud.worker.orphaned_containers", + return_value=[{"container": "c1", + "pid_verified": True}]): + self.assertEqual(w.cleanup_orphaned_containers(), []) + + def test_stale_cleanup_failure_returns_empty(self): + # 2534-2537 + w = _worker(self.db_path, executor=SimpleNamespace( + cleanup_stale_containers=mock.Mock( + side_effect=RuntimeError("cleanup down")))) + self.assertEqual(w.cleanup_stale_containers(), []) + + def test_stale_cleanup_nonempty_returned(self): + # 2539 + w = _worker(self.db_path, executor=SimpleNamespace( + cleanup_stale_containers=mock.Mock(return_value=["c1"]))) + self.assertEqual(w.cleanup_stale_containers(), ["c1"]) + + +class GatewayTestCase(_Base): + def test_key_available_env_failure_false(self): + # 2564-2565 + class _BadEnv: + def get(self, *a, **k): + raise RuntimeError("env down") + + w = Worker(self.db_path, "/fake/host.py", env=_BadEnv()) + self.assertFalse(w.key_available()) + + def _sub2api_env(self): + return (mock.patch("cloud.worker._gateway_name", + return_value="sub2api"), + mock.patch("cloud.worker._gateway_origin", + return_value="http://gw.example")) + + def test_gateway_parent_row_type_error(self): + # 2607-2608 -> owner missing -> RuntimeError + w = _worker(self.db_path) + name_p, origin_p = self._sub2api_env() + + class _Cursor: + def fetchone(self): + return 5 + + class _Conn: + def execute(self, *a, **k): + return _Cursor() + + def close(self): + pass + + with name_p, origin_p, mock.patch( + "cloud.worker._db_connect", return_value=_Conn()): + with self.assertRaises(RuntimeError): + w._gateway_context({"parent_task_id": "p0"}) + + def test_gateway_parent_lookup_failure(self): + # 2609-2610 -> owner missing -> RuntimeError + w = _worker(self.db_path) + name_p, origin_p = self._sub2api_env() + with name_p, origin_p, mock.patch( + "cloud.worker._db_connect", + side_effect=RuntimeError("db down")): + with self.assertRaises(RuntimeError): + w._gateway_context({"parent_task_id": "p0"}) + + def test_sub2api_default_model_none_available(self): + # 2660-2662 + w = _worker(self.db_path) + with mock.patch.dict(os.environ, + {"CLOUD_SUB2API_DEFAULT_MODEL": ""}): + with mock.patch("cloud.worker._gateway_origin", + return_value="http://gw.example"): + with mock.patch("cloud.worker._list_gateway_models", + return_value=[]): + with self.assertRaises(RuntimeError): + w._sub2api_default_model( + {"SUB2API_API_KEY": "k"}) + + def test_sub2api_default_model_first(self): + # 2663 + w = _worker(self.db_path) + with mock.patch.dict(os.environ, + {"CLOUD_SUB2API_DEFAULT_MODEL": ""}): + with mock.patch("cloud.worker._gateway_origin", + return_value="http://gw.example"): + with mock.patch("cloud.worker._list_gateway_models", + return_value=["m1", "m2"]): + self.assertEqual( + w._sub2api_default_model( + {"SUB2API_API_KEY": "k"}), "m1") + + def test_task_seed_dir_db_failure_none(self): + # 2680-2681 + w = _worker(self.db_path) + with mock.patch("cloud.worker._db_connect", + side_effect=RuntimeError("db down")): + self.assertIsNone(w._task_seed_dir({"id": "t1"})) + + def test_cleanup_seed_failure_swallowed(self): + # 2701-2702 + w = _worker(self.db_path) + w._task_seed_dir = mock.Mock( + side_effect=RuntimeError("seed down")) + w._cleanup_task_seed({"id": "t1"}) + + +class ProcessTestCase(_Base): + def _finished_proc(self, returncode=0): + proc = mock.Mock() + proc.poll.return_value = returncode + proc.communicate.return_value = ("out", "err") + proc.returncode = returncode + proc.pid = 4242 + return proc + + def _proc_worker(self, proc): + return _worker(self.db_path, + executor=SimpleNamespace( + popen_task=mock.Mock(return_value=proc))) + + def test_executor_popen_path(self): + # 2730 + 2820 (no _result_from_runner on the executor) + self._insert_task(status="running") + proc = self._finished_proc() + w = self._proc_worker(proc) + out = w._execute_via_process({"id": "t1", "project_id": "p1"}) + self.assertEqual(out["status"], "completed") + self.assertEqual(out["stdout"], "out") + + def test_set_child_pid_failure_swallowed(self): + # 2736-2737 + self._insert_task(status="running") + w = _worker(self.db_path, + process_factory=lambda task: self._finished_proc()) + w._set_child_pid = mock.Mock( + side_effect=RuntimeError("pid down")) + out = w._execute_via_process({"id": "t1", "project_id": "p1"}) + self.assertEqual(out["status"], "completed") + + def test_quota_interval_none_defaults(self): + # 2747 + self._insert_task(status="running") + w = _worker(self.db_path, + process_factory=lambda task: self._finished_proc()) + w.quota_poll_interval_seconds = None + out = w._execute_via_process({"id": "t1", "project_id": "p1"}) + self.assertEqual(out["status"], "completed") + + def test_loop_db_lookup_failure_continues(self): + # 2763-2764 + self._insert_task(status="running") + proc = mock.Mock() + proc.poll.side_effect = [None, 0] + proc.communicate.return_value = ("out", "err") + proc.returncode = 0 + proc.pid = 4242 + w = _worker(self.db_path, + process_factory=lambda task: proc) + w.get_task = mock.Mock(side_effect=RuntimeError("db down")) + out = w._execute_via_process({"id": "t1", "project_id": "p1"}) + self.assertEqual(out["status"], "completed") + + def test_loop_db_interrupted_cancels(self): + # 2767-2769 + self._insert_task(status="running") + proc = mock.Mock() + proc.poll.side_effect = [None, 0] + proc.communicate.return_value = ("out", "err") + proc.returncode = 0 + proc.pid = 4242 + w = _worker(self.db_path, + process_factory=lambda task: proc) + w.get_task = mock.Mock(return_value={"status": "interrupted"}) + w._kill_proc_group = lambda p: None + out = w._execute_via_process({"id": "t1", "project_id": "p1"}) + self.assertEqual(out["status"], "cancelled") + + def test_loop_quota_check_failure_continues(self): + # 2780-2781 + self._insert_task(status="running") + proc = mock.Mock() + proc.poll.side_effect = [None, 0] + proc.communicate.return_value = ("out", "err") + proc.returncode = 0 + proc.pid = 4242 + w = _worker(self.db_path, + process_factory=lambda task: proc) + w.quota_poll_interval_seconds = 0 + out = w._execute_via_process( + {"id": "t1", "project_id": "p1"}, + quota_check=mock.Mock(side_effect=RuntimeError("quota down"))) + self.assertEqual(out["status"], "completed") + + def test_communicate_failure_falls_back(self): + # 2789-2794 + self._insert_task(status="running") + proc = mock.Mock() + proc.poll.return_value = 0 + proc.communicate.side_effect = [RuntimeError("comm down"), + ("out2", "err2")] + proc.returncode = 0 + proc.pid = 4242 + w = _worker(self.db_path, + process_factory=lambda task: proc) + out = w._execute_via_process({"id": "t1", "project_id": "p1"}) + self.assertEqual(out["stdout"], "out2") + proc.kill.assert_called_once() + + def test_goal_result_from_runner(self): + # 2817 + self._insert_task(status="running") + proc = self._finished_proc() + marker = {"status": "goal-met", "task_id": "t1"} + w = _worker(self.db_path, executor=SimpleNamespace( + popen_task=mock.Mock(return_value=proc), + _goal_result_from_runner=mock.Mock(return_value=marker))) + out = w._execute_via_process({"id": "t1", "project_id": "p1"}, + kind="goal") + self.assertEqual(out, marker) + + +if __name__ == "__main__": + unittest.main() diff --git a/cloud/tests/test_v2cov_gaps_d4.py b/cloud/tests/test_v2cov_gaps_d4.py new file mode 100644 index 0000000..6da3672 --- /dev/null +++ b/cloud/tests/test_v2cov_gaps_d4.py @@ -0,0 +1,85 @@ +"""Coverage batch v4-D4: cloud/server.py dispatch-safe/upload branches +and cloud/ledger_outbox.py replay rewrite. + +Stubbing discipline: the method under test is never stubbed. Patched +collaborators are: _dispatch / _send_json / _read_body / parse_multipart +/ parse_paths_field (sinks and data providers). Every assertion reads a +real return value, a real captured response, or a real file. +""" + +import json +import os +import sys +import tempfile +import unittest +from pathlib import Path +from unittest import mock + +REPO_ROOT = Path(__file__).resolve().parents[2] +sys.path.insert(0, str(REPO_ROOT)) + +from cloud import server as server_mod +from cloud.ledger_outbox import replay_failures + + +class DispatchSafeTestCase(unittest.TestCase): + def test_dispatch_and_send_both_raise_swallowed(self): + # 2310-2311: _dispatch raising and the 500 fallback also raising. + handler = object.__new__(server_mod.CloudControlPlaneHandler) + handler._responded = False + handler.path = "/x" + handler._dispatch = mock.Mock( + side_effect=RuntimeError("dispatch down")) + handler._send_json = mock.Mock( + side_effect=RuntimeError("send down")) + handler._dispatch_safe("GET") + handler._send_json.assert_called_once_with( + 500, {"error": "internal_error"}) + + +class UploadMismatchTestCase(unittest.TestCase): + def test_paths_files_count_mismatch(self): + # 3149 + handler = object.__new__(server_mod.CloudControlPlaneHandler) + handler.headers = {"Content-Type": "multipart/form-data", + "Content-Length": "10"} + handler._read_body = mock.Mock(return_value=b"body") + sent = [] + handler._send_json = lambda code, payload: sent.append( + (code, payload)) + with mock.patch.object( + server_mod, "parse_multipart", + return_value=({}, [("a", b"x"), ("b", b"y")])): + with mock.patch.object( + server_mod, "parse_paths_field", + return_value=["only-one"]): + handler._create_upload("u1") + self.assertEqual(sent, [(400, {"error": "附件路径与文件数量不一致"})]) + + +class OutboxReplayTestCase(unittest.TestCase): + def test_failed_record_rewritten_back(self): + # 170: a failed replay row is rewritten into the outbox file. + with tempfile.TemporaryDirectory() as tmp: + path = os.path.join(tmp, "outbox.jsonl") + good = {"task_id": "t1", "outcome": "success"} + bad = {"task_id": "t2", "outcome": "failure"} + with open(path, "w", encoding="utf-8") as f: + f.write(json.dumps(good) + "\n") + f.write(json.dumps(bad) + "\n") + + def _record(conn, task_id, outcome, is_defect=False): + if task_id == "t2": + raise RuntimeError("ledger down") + + stats = replay_failures(path, None, record_fn=_record) + self.assertEqual(stats["replayed"], 1) + self.assertEqual(stats["failed"], 1) + self.assertEqual(stats["remaining"], 1) + with open(path, encoding="utf-8") as f: + rows = [json.loads(line) for line in f if line.strip()] + self.assertEqual(rows, [bad]) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_v2cov_gaps_c.py b/tests/test_v2cov_gaps_c.py new file mode 100644 index 0000000..c14c8f4 --- /dev/null +++ b/tests/test_v2cov_gaps_c.py @@ -0,0 +1,502 @@ +"""Coverage batch v4-C: host.py state/lock/worktree/verify/notify/ +pr/dispatch/goal-loop branches. + +Stubbing discipline: the function under test is never stubbed. +Patched collaborators are: socket.gethostname / os.getmtime / os.rmdir / +os.replace / log_event / save_state / build_verify_audit_report / +subprocess.run / gh runner / FakeServer (all sinks or data providers), +plus host.HERE (a path constant) and os.environ git identity vars for +the real-git worktree tests. Every assertion reads a real return value, +a real file, or real stdout. +""" + +import contextlib +import io +import json +import os +import socket +import subprocess +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[1] +sys.path.insert(0, str(REPO_ROOT)) + +import host +from host import ( + _audit_verify, + _cmd_verify_audit, + _seed_worktree, + _state_lock, + _state_tmp_path, + build_verify_audit_report, + copy_tree_excluding_git, + dispatch, + log_event, + make_worktree, + merge_pr, + notify_failure, + parse_pr_checks, + recover_interrupted, + restore_state, + run_gated_verify, + run_gh, + run_goal_loop, + run_health_check, + run_task_turn, + run_verify_argv, + save_state, + task_diff, +) + +GIT_ENV = { + "GIT_AUTHOR_NAME": "t", "GIT_AUTHOR_EMAIL": "t@t", + "GIT_COMMITTER_NAME": "t", "GIT_COMMITTER_EMAIL": "t@t", + "GIT_CONFIG_NOSYSTEM": "1", +} + + +def _git(args, cwd, env_extra=None): + env = dict(os.environ) + env.update(GIT_ENV) + if env_extra: + env.update(env_extra) + return subprocess.run(["git"] + args, cwd=cwd, capture_output=True, + text=True, env=env) + + +def _init_repo(path): + os.makedirs(path, exist_ok=True) + _git(["init", "-q"], path) + with open(os.path.join(path, "f.txt"), "w") as fh: + fh.write("base\n") + _git(["add", "-A"], path) + _git(["commit", "-qm", "init"], path) + + +class StateLockTestCase(unittest.TestCase): + def test_state_tmp_path_socket_failure(self): + with mock.patch.object(socket, "gethostname", + side_effect=OSError("no hostname")): + out = _state_tmp_path("/tmp/state.json") + self.assertIn("unknown", out) + + def test_state_lock_getmtime_failure_timeout(self): + with tempfile.TemporaryDirectory() as d: + path = os.path.join(d, "state.json") + os.mkdir(path + ".lockdir") + with mock.patch.object(host, "_STATE_LOCK_TIMEOUT", 0), \ + mock.patch("os.path.getmtime", + side_effect=OSError("gone")): + with self.assertRaises(RuntimeError): + with _state_lock(path): + pass + + def test_state_lock_stale_rmdir_once_failure(self): + with tempfile.TemporaryDirectory() as d: + path = os.path.join(d, "state.json") + lockdir = path + ".lockdir" + os.mkdir(lockdir) + old = time.time() - 100000 + os.utime(lockdir, (old, old)) + real_rmdir = os.rmdir + calls = [] + + def _flaky(p): + calls.append(p) + if len(calls) == 1: + raise OSError("someone else") + return real_rmdir(p) + with mock.patch("os.rmdir", side_effect=_flaky): + with _state_lock(path): + pass + self.assertFalse(os.path.exists(lockdir)) + + def test_state_lock_final_rmdir_failure_swallowed(self): + with tempfile.TemporaryDirectory() as d: + path = os.path.join(d, "state.json") + with _state_lock(path): + os.rmdir(path + ".lockdir") + self.assertFalse(os.path.exists(path + ".lockdir")) + + +class RestoreRecoverSaveLogTestCase(unittest.TestCase): + def test_restore_state_replace_failure(self): + with tempfile.TemporaryDirectory() as d: + path = os.path.join(d, "state.json") + with open(path, "w") as fh: + fh.write("{broken") + with mock.patch("os.replace", + side_effect=RuntimeError("rename down")): + ok, msg = restore_state(path) + self.assertFalse(ok) + self.assertIn("改名留底失败", msg) + + def test_restore_state_no_file_no_backups(self): + # 1047 行（corrupt=None）只在“严格读失败、但复查时文件已 + # 不在”这一竞态下可达：单线程里文件缺失时严格读直接返回 + # 空 state、无需恢复；这里用 exists 的调用序列注入该竞态。 + with tempfile.TemporaryDirectory() as d: + path = os.path.join(d, "state.json") + with open(path, "w") as fh: + fh.write("{broken") + real_exists = os.path.exists + calls = [] + + def _exists(p): + calls.append(p) + if len(calls) == 1: + return True + return False + with mock.patch("os.path.exists", side_effect=_exists): + ok, msg = restore_state(path) + self.assertFalse(ok) + self.assertIn("没有可用的备份", msg) + + def test_recover_interrupted_log_failure(self): + state = {"tasks": {"t": {"status": "running", + "last_turn_at": 0, "created_at": 0}}} + with mock.patch.object(host, "log_event", + side_effect=RuntimeError("log down")): + out = recover_interrupted(state, "/tmp/state.json") + self.assertEqual(out, ["t"]) + self.assertEqual(state["tasks"]["t"]["status"], "interrupted") + + def test_save_state_conflict_log_failure(self): + with tempfile.TemporaryDirectory() as d: + path = os.path.join(d, "state.json") + with open(path, "w") as fh: + json.dump({"version": 5, "tasks": {}}, fh) + os.mkdir(os.path.join(d, "host-run.log")) + state = {"tasks": {}, "_loaded_version": 1} + save_state(path, state) + with open(path) as fh: + saved = json.load(fh) + self.assertEqual(saved["version"], 6) + + def test_save_state_last_writer_socket_failure(self): + with tempfile.TemporaryDirectory() as d: + path = os.path.join(d, "state.json") + with mock.patch.object(socket, "gethostname", + side_effect=OSError("no hostname")): + save_state(path, {"tasks": {}}) + with open(path) as fh: + saved = json.load(fh) + self.assertEqual(saved["last_writer"], str(os.getpid())) + + def test_log_event_failure_swallowed(self): + with tempfile.TemporaryDirectory() as d: + blocker = os.path.join(d, "blocker") + with open(blocker, "w") as fh: + fh.write("x") + log_event(os.path.join(blocker, "state.json"), "hi") + + def test_health_check_pid_alive_and_close_failure(self): + with tempfile.TemporaryDirectory() as d: + path = os.path.join(d, "state.json") + with open(path, "w") as fh: + json.dump({"version": 1, "tasks": {}}, fh) + with open(os.path.join(d, "host-daemon.pid"), "w") as fh: + fh.write(str(os.getpid())) + server = mock.Mock() + server.list_threads.return_value = {"data": []} + server.close.side_effect = RuntimeError("close down") + out = run_health_check(path, {}, sys.executable, d, "a", + server_factory=lambda: server) + self.assertTrue(out["checks"]["daemon_alive"]) + self.assertEqual(out["checks"]["daemon_pid"], os.getpid()) + self.assertTrue(out["checks"]["server_ok"]) + + +class WorktreeDiffTestCase(unittest.TestCase): + def test_make_worktree_existing(self): + with tempfile.TemporaryDirectory() as d: + with mock.patch.object(host, "HERE", d): + base = os.path.join(d, "work", "host-worktrees") + os.makedirs(os.path.join(base, "t1")) + path, branch = make_worktree("/tmp/repo", "t1") + self.assertEqual(path, os.path.join(base, "t1")) + self.assertIsNone(branch) + + def test_make_worktree_branch_fallback(self): + with tempfile.TemporaryDirectory() as d: + repo = os.path.join(d, "repo") + _init_repo(repo) + _git(["branch", "host/t1"], repo) + with mock.patch.object(host, "HERE", d): + path, branch = make_worktree(repo, "t1") + self.assertEqual(branch, "host/t1") + self.assertTrue(os.path.isdir(path)) + + def test_make_worktree_both_fail(self): + with tempfile.TemporaryDirectory() as d: + repo = os.path.join(d, "repo") + os.makedirs(repo) + _git(["init", "-q"], repo) + with mock.patch.object(host, "HERE", d): + with self.assertRaises(RuntimeError): + make_worktree(repo, "t1") + + def test_make_worktree_copy_non_git(self): + with tempfile.TemporaryDirectory() as d: + repo = os.path.join(d, "repo") + os.makedirs(repo) + with open(os.path.join(repo, "a.txt"), "w") as fh: + fh.write("hi") + with mock.patch.object(host, "HERE", d): + path, branch = make_worktree(repo, "t1") + self.assertIsNone(branch) + with open(os.path.join(path, "a.txt")) as fh: + self.assertEqual(fh.read(), "hi") + + def test_task_diff_unreadable(self): + with tempfile.TemporaryDirectory() as d: + repo = os.path.join(d, "repo") + _init_repo(repo) + with open(os.path.join(repo, "bin.dat"), "wb") as fh: + fh.write(b"\xff\xfe\x00abc") + out = task_diff(repo) + self.assertEqual(out["untracked_contents"]["bin.dat"], + "") + + +class VerifyAuditTestCase(unittest.TestCase): + def test_audit_verify_log_failure(self): + with mock.patch.object(host, "log_event", + side_effect=RuntimeError("log down")): + _audit_verify("/tmp/state.json", "true", "argv", 0) + + def test_run_verify_argv_empty(self): + self.assertEqual(run_verify_argv([], "/tmp"), (0, "")) + + def test_run_verify_argv_non_str(self): + with self.assertRaises(TypeError): + run_verify_argv(["echo", 123], "/tmp") + + def test_run_gated_verify_blank(self): + self.assertEqual(run_gated_verify(" ", "/tmp"), (0, "")) + + def test_build_report_non_dict_task(self): + report = build_verify_audit_report( + {"tasks": {"bad": "str", + "ok": {"verify_command": None}}}) + self.assertEqual([r["task"] for r in report], []) + + def test_cmd_verify_audit_notes_and_simple(self): + rows = [ + {"task": "n", "kind": "shell", "features": ["f"], + "reason": "", "argv": None, "notes": ["note1"], + "applied": False}, + {"task": "s", "kind": "simple", "features": [], + "reason": "", "argv": ["true"], "notes": [], + "applied": False}, + ] + args = SimpleNamespace(apply=False, json=False, state="st.json") + buf = io.StringIO() + with mock.patch.object(host, "build_verify_audit_report", + return_value=rows), \ + contextlib.redirect_stdout(buf): + _cmd_verify_audit(args, {"tasks": {}}) + out = buf.getvalue() + self.assertIn("notes=note1", out) + self.assertIn("[可转 argv=['true']]", out) + + def test_cmd_verify_audit_json(self): + args = SimpleNamespace(apply=False, json=True, state="st.json") + buf = io.StringIO() + with mock.patch.object(host, "build_verify_audit_report", + return_value=[]), \ + contextlib.redirect_stdout(buf): + _cmd_verify_audit(args, {"tasks": {}}) + parsed = json.loads(buf.getvalue()) + self.assertEqual(parsed["applied"], 0) + self.assertEqual(parsed["tasks"], []) + + +class NotifyTestCase(unittest.TestCase): + def _state_path(self, d): + return os.path.join(d, "state.json") + + def test_notify_no_command_log_failure(self): + with tempfile.TemporaryDirectory() as d: + with mock.patch.object(host, "log_event", + side_effect=RuntimeError("log down")): + msg = notify_failure(self._state_path(d), {}, "t1", + {"status": "failed", "text": "boom"}) + self.assertTrue(msg.startswith("FAILED t1")) + + def test_notify_template_error_log_failure(self): + with tempfile.TemporaryDirectory() as d: + with mock.patch.object(host, "log_event", + side_effect=RuntimeError("log down")): + msg = notify_failure( + self._state_path(d), {"notify_command": "echo {unknown}"}, + "t1", {"status": "failed", "text": "boom"}) + self.assertTrue(msg.startswith("FAILED t1")) + + def test_notify_rc_log_failure(self): + with tempfile.TemporaryDirectory() as d: + with mock.patch.object(host, "log_event", + side_effect=RuntimeError("log down")): + msg = notify_failure( + self._state_path(d), {"notify_command": "true"}, + "t1", {"status": "failed", "text": "boom"}) + self.assertTrue(msg.startswith("FAILED t1")) + + def test_notify_timeout_log_failure(self): + with tempfile.TemporaryDirectory() as d: + with mock.patch.object(host, "log_event", + side_effect=RuntimeError("log down")), \ + mock.patch("subprocess.run", + side_effect=subprocess.TimeoutExpired( + "cmd", 30)): + msg = notify_failure( + self._state_path(d), {"notify_command": "sleep 60"}, + "t1", {"status": "failed", "text": "boom"}) + self.assertTrue(msg.startswith("FAILED t1")) + + def test_notify_generic_error_log_failure(self): + with tempfile.TemporaryDirectory() as d: + with mock.patch.object(host, "log_event", + side_effect=RuntimeError("log down")), \ + mock.patch("subprocess.run", + side_effect=RuntimeError("spawn down")): + msg = notify_failure( + self._state_path(d), {"notify_command": "true"}, + "t1", {"status": "failed", "text": "boom"}) + self.assertTrue(msg.startswith("FAILED t1")) + + +class PrDispatchTurnTestCase(unittest.TestCase): + def test_parse_pr_checks_branches(self): + checks, all_pass = parse_pr_checks("foo\t\nrandom words\npass\n") + self.assertEqual(checks, [{"name": "pass", "state": "pass"}]) + self.assertTrue(all_pass) + + def test_merge_pr_gh_failure(self): + def _runner(args): + if args[1] == "checks": + return SimpleNamespace(returncode=0, stdout="ci\tpass\n", + stderr="") + return SimpleNamespace(returncode=1, stdout="", + stderr="merge failed") + task = {"name": "t", "pr_number": 7, "pr_url": "u"} + ok, msg = merge_pr(task, gh_runner=_runner, repo="/tmp") + self.assertFalse(ok) + self.assertIn("gh pr merge 失败", msg) + + def test_run_gh_real_path_mocked(self): + sentinel = SimpleNamespace(returncode=0, stdout="", stderr="") + with mock.patch("subprocess.run", return_value=sentinel) as run: + out = run_gh(["pr", "checks", "1"], cwd="/tmp") + self.assertIs(out, sentinel) + run.assert_called_once() + + def test_copy_tree_skips_pyc(self): + with tempfile.TemporaryDirectory() as d: + src = os.path.join(d, "src") + dst = os.path.join(d, "dst") + os.makedirs(src) + with open(os.path.join(src, "a.pyc"), "wb") as fh: + fh.write(b"\x00") + with open(os.path.join(src, "b.txt"), "w") as fh: + fh.write("b") + copy_tree_excluding_git(src, dst) + self.assertFalse(os.path.exists(os.path.join(dst, "a.pyc"))) + self.assertTrue(os.path.exists(os.path.join(dst, "b.txt"))) + + def test_dispatch_relative_repo(self): + with tempfile.TemporaryDirectory() as d: + rel = os.path.join(d, "relrepo") + os.makedirs(rel) + server = SimpleNamespace( + create_thread=lambda cwd, **kw: "th1") + state = {"tasks": {}} + with mock.patch.object(host, "HERE", d): + task = dispatch(server, state, "t1", "prompt", "relrepo", + "m", "low", None, None, use_worktree=False) + self.assertEqual(task["repo"], rel) + self.assertEqual(task["cwd"], rel) + + def test_run_task_turn_type_error_reraise(self): + class _Server: + def run_turn(self, *args, **kwargs): + raise TypeError("boom") + task = {"thread_id": "th", "sandbox": None, "turns": []} + with self.assertRaises(TypeError): + run_task_turn(_Server(), {"tasks": {}}, task, "p", "m", "low", + interrupt_file="/tmp/if") + + def test_seed_worktree_git_skipped(self): + with tempfile.TemporaryDirectory() as d: + seed = os.path.join(d, "seed") + work = os.path.join(d, "work") + os.makedirs(os.path.join(seed, ".git")) + os.makedirs(work) + with open(os.path.join(seed, "f.txt"), "w") as fh: + fh.write("f") + _seed_worktree(work, seed) + self.assertTrue(os.path.exists(os.path.join(work, "f.txt"))) + self.assertFalse(os.path.exists(os.path.join(work, ".git"))) + + +class FakeServer: + def run_turn(self, *args, **kwargs): + return {"status": "completed", "text": ""} + + +class GoalLoopTestCase(unittest.TestCase): + def _setup(self, d): + state_path = os.path.join(d, "state.json") + task = {"name": "g", "thread_id": "x", "cwd": d, + "sandbox": None, "turns": []} + state = {"tasks": {"g": task}, "automations": []} + return state_path, task, state + + def test_goal_verify_oserror_interrupt(self): + with tempfile.TemporaryDirectory() as d: + state_path, task, state = self._setup(d) + out = run_goal_loop( + FakeServer(), state, state_path, task, "goal", None, 1, + "m", "low", "workspace-write", cfg={}, + verify_argv=["/nonexistent/binary-xyz"]) + self.assertTrue(out["interrupted"]) + self.assertEqual(task["status"], "interrupted") + + def test_goal_save_failure_interrupt(self): + with tempfile.TemporaryDirectory() as d: + state_path, task, state = self._setup(d) + orig = host.save_state + calls = [] + + def _flaky(path, st): + calls.append(1) + if len(calls) > 1: + raise RuntimeError("save down") + return orig(path, st) + with mock.patch.object(host, "save_state", + side_effect=_flaky): + out = run_goal_loop( + FakeServer(), state, state_path, task, "goal", "true", + 1, "m", "low", "workspace-write", cfg={}) + self.assertTrue(out["interrupted"]) + self.assertEqual(task["status"], "interrupted") + + def test_goal_steps_write_failure_swallowed(self): + with tempfile.TemporaryDirectory() as d: + state_path, task, state = self._setup(d) + with open(os.path.join(d, "host-steps"), "w") as fh: + fh.write("blocker") + out = run_goal_loop( + FakeServer(), state, state_path, task, "goal", "true", + 1, "m", "low", "workspace-write", cfg={}) + self.assertTrue(out["ok"]) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_v2cov_gaps_d3.py b/tests/test_v2cov_gaps_d3.py new file mode 100644 index 0000000..3e4ea59 --- /dev/null +++ b/tests/test_v2cov_gaps_d3.py @@ -0,0 +1,692 @@ +"""Coverage batch v4-D3: host.py wait/bin/client/run-turn/restore/ +merge/notify/pr/dispatch/main branches. + +Stubbing discipline: the function under test is never stubbed. Patched +collaborators are: _summarize_trace_item / _append_trace / request / +_wait_turn / interrupt_turn / resume_thread / ensure_alive / _start / +git / run_gh / _github_remote_url / task_diff / open_review_comments / +merge_task / _cmd_verify_audit / daemonize / Server / subprocess.run / +os.remove (sinks, data providers and predicates). Every assertion reads +a real return value, a real file, or real stdout. +""" + +import contextlib +import io +import json +import os +import queue +import sqlite3 +import subprocess +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[1] +sys.path.insert(0, str(REPO_ROOT)) + +import host +from host import ( + Server, + _wait_turn, + create_pr, + default_codex_bin, + dispatch_from_branch, + merge_task, + notify_failure, + parse_pr_checks, + restore_state, +) + +GIT_ENV = { + "GIT_AUTHOR_NAME": "t", "GIT_AUTHOR_EMAIL": "t@t", + "GIT_COMMITTER_NAME": "t", "GIT_COMMITTER_EMAIL": "t@t", + "GIT_CONFIG_NOSYSTEM": "1", +} + + +def _server(): + srv = object.__new__(Server) + srv.proc = None + srv.codex_bin = "codex" + srv.codex_home = None + srv.approval = "deny" + srv.model_provider = None + srv.progress_interval = 30 + srv.send_lock = threading.Lock() + srv.next_id = 1 + srv.pending = {} + srv.pending_lock = threading.Lock() + srv.turn_queues = {} + srv.events = [] + srv.events_lock = threading.Lock() + srv.restart_count = 0 + srv.trace_path = None + srv._trace_lines = 0 + return srv + + +class _FakeServer: + def __init__(self, *args, **kwargs): + self.closed = False + + def initialize(self): + pass + + def close(self): + self.closed = True + + def resume_thread(self, thread_id): + pass + + def interrupt_turn(self, thread_id, turn_id): + return {"ok": True} + + +class WaitAndBinTestCase(unittest.TestCase): + def test_wait_turn_tiny_timeout(self): + # 154: remaining <= 0 on the first slice. + q = queue.Queue() + self.assertEqual(_wait_turn(q, 1e-5), ("timeout", None)) + + def test_default_codex_bin_paired(self): + # 204 + with tempfile.TemporaryDirectory() as tmp: + os.makedirs(os.path.join(tmp, "bin")) + paired = os.path.join(tmp, "bin", "codex") + with open(paired, "w", encoding="utf-8") as f: + f.write("x") + with mock.patch.object(host, "HERE", tmp): + with mock.patch.dict(os.environ, {}, clear=False): + os.environ.pop("CODEX_BIN", None) + self.assertEqual(default_codex_bin(), paired) + + +class ServerClientTestCase(unittest.TestCase): + def test_ensure_alive_true_and_false(self): + # 456, 462-463 + srv = _server() + srv.proc = mock.Mock() + srv.proc.poll.return_value = None + self.assertTrue(srv.ensure_alive()) + srv2 = _server() + srv2.proc = None + srv2._start = mock.Mock(side_effect=RuntimeError("start down")) + self.assertFalse(srv2.ensure_alive()) + + def test_read_loop_non_dict_stderr_failure(self): + # 501-502 + srv = _server() + srv.proc = SimpleNamespace(stdout=[b"123\n"]) + bad_err = mock.Mock() + bad_err.write.side_effect = RuntimeError("stderr down") + with mock.patch.object(sys, "stderr", bad_err): + srv._read_loop() + bad_err.write.assert_called() + + def test_read_loop_append_trace_failure(self): + # 562-565 + srv = _server() + srv.trace_path = os.path.join( + tempfile.mkdtemp(), "trace.jsonl") + srv.proc = SimpleNamespace(stdout=[ + b'{"jsonrpc":"2.0","method":"agent/message","params":{}}\n']) + srv._append_trace = mock.Mock( + side_effect=RuntimeError("trace down")) + srv._read_loop() + self.assertEqual(len(srv.events), 1) + + def test_read_loop_events_append_failure(self): + # 570-571: outer except of the read loop. + srv = _server() + srv.proc = SimpleNamespace(stdout=[ + b'{"jsonrpc":"2.0","method":"agent/message","params":{}}\n']) + srv.events = mock.Mock() + srv.events.append.side_effect = RuntimeError("events down") + srv._read_loop() + + def test_request_broken_pipe_retry_then_raise(self): + # 596-600 + srv = _server() + srv.proc = SimpleNamespace( + stdin=mock.Mock()) + srv.proc.stdin.write.side_effect = BrokenPipeError("pipe down") + srv.ensure_alive = mock.Mock(return_value=True) + with self.assertRaises(BrokenPipeError): + srv.request("initialize", {}, timeout=60) + self.assertEqual(srv.ensure_alive.call_count, 1) + + def test_thread_method_error_branches(self): + # 627, 639, 651, 662, 673 + srv = _server() + srv.request = mock.Mock(return_value={"error": "boom"}) + with self.assertRaises(RuntimeError): + srv.create_thread("/tmp") + with self.assertRaises(RuntimeError): + srv.list_threads() + with self.assertRaises(RuntimeError): + srv.resume_thread("th1") + with self.assertRaises(RuntimeError): + srv.start_review("th1", "target") + with self.assertRaises(RuntimeError): + srv.fork_thread("th1") + with self.assertRaises(RuntimeError): + srv.archive_thread("th1") + + def test_fork_thread_non_dict_result_none(self): + # 668 + srv = _server() + srv.request = mock.Mock(return_value={"result": "not-a-dict"}) + self.assertIsNone(srv.fork_thread("th1")) + + def test_append_trace_break_at_cap(self): + # 430 + srv = _server() + tmp = tempfile.mkdtemp() + srv.trace_path = os.path.join(tmp, "trace.jsonl") + srv._trace_lines = host.TRACE_MAX_LINES - 1 + with mock.patch.object( + host, "_summarize_trace_item", + return_value=[{"a": 1}, {"b": 2}]): + srv._append_trace({"method": "x", "params": {}}) + with open(srv.trace_path, encoding="utf-8") as f: + self.assertEqual(len(f.readlines()), 1) + + def test_turn_error_empty_message_none(self): + # 891 + srv = _server() + home = tempfile.mkdtemp() + srv.codex_home = home + db_path = os.path.join(home, "thread_history_1.sqlite") + conn = sqlite3.connect(db_path) + try: + conn.execute( + "CREATE TABLE thread_turns(thread_id TEXT, turn_id TEXT," + " error_json TEXT)") + conn.execute( + "INSERT INTO thread_turns VALUES(?, ?, ?)", + ("th1", "turn1", '{"message": ""}')) + conn.commit() + finally: + conn.close() + self.assertIsNone(srv._turn_error_from_history("th1", "turn1")) + + def test_close_stdin_stdout_failures(self): + # 899-900, 907-908 + srv = _server() + proc = mock.Mock() + proc.stdin.close.side_effect = RuntimeError("stdin down") + proc.wait.return_value = 0 + proc.stdout.close.side_effect = RuntimeError("stdout down") + srv.proc = proc + srv.close() + + +class RunTurnTestCase(unittest.TestCase): + def _client(self): + srv = _server() + srv.request = mock.Mock(return_value={ + "result": {"turn": {"id": "turn1"}}}) + return srv + + def test_resume_failure_swallowed(self): + # 683-687 + srv = self._client() + srv.proc = None + srv.ensure_alive = mock.Mock(return_value=True) + srv.resume_thread = mock.Mock( + side_effect=RuntimeError("resume down")) + with mock.patch.object(host, "_wait_turn", + return_value=("completed", { + "params": {"turn": {"items": []}}})): + out = srv.run_turn("th1", "do work") + self.assertEqual(out["status"], "completed") + + def test_progress_success_and_failure(self): + # 735-753 + srv = self._client() + tmp = tempfile.mkdtemp() + srv.trace_path = os.path.join(tmp, "trace.jsonl") + srv._trace_lines = 0 + + def _fake_wait(q, timeout, progress_interval=None, + on_progress=None, interrupt_check=None): + on_progress(5) + srv.trace_path = "/nonexistent-dir-xyz/trace.jsonl" + on_progress(6) + return ("completed", {"params": {"turn": {"items": []}}}) + + with mock.patch.object(host, "_wait_turn", + side_effect=_fake_wait): + out = srv.run_turn("th1", "do work") + self.assertEqual(out["status"], "completed") + with open(os.path.join(tmp, "trace.jsonl"), + encoding="utf-8") as f: + self.assertEqual(len(f.readlines()), 1) + + def test_interrupted_error_and_remove_failure(self): + # 773-780, 784-785 + srv = self._client() + srv.interrupt_turn = mock.Mock( + return_value={"error": "nope"}) + with tempfile.TemporaryDirectory() as tmp: + interrupt_file = os.path.join(tmp, "req.json") + with open(interrupt_file, "w", encoding="utf-8") as f: + f.write("{}") + with mock.patch.object(host, "_wait_turn", + return_value=("interrupted-requested", + None)): + with mock.patch("os.remove", + side_effect=OSError("remove down")): + out = srv.run_turn("th1", "do work", + interrupt_file=interrupt_file) + self.assertEqual(out["status"], "interrupted") + + def test_wait_raises_empty_timeout(self): + # 799-802 + srv = self._client() + with mock.patch.object(host, "_wait_turn", + side_effect=queue.Empty()): + out = srv.run_turn("th1", "do work") + self.assertEqual(out["status"], "timeout") + + def test_fallback_final_answer_from_events(self): + # 822 + srv = self._client() + + def _fake_wait(q, timeout, progress_interval=None, + on_progress=None, interrupt_check=None): + srv.events.append({"params": {"item": { + "type": "agentMessage", "phase": "final_answer", + "text": "hello"}}}) + return ("completed", {"params": {"turn": {"items": []}}}) + + with mock.patch.object(host, "_wait_turn", + side_effect=_fake_wait): + out = srv.run_turn("th1", "do work") + self.assertEqual(out["text"], "hello") + + +class RestoreMergeNotifyTestCase(unittest.TestCase): + def test_restore_corrupt_state_from_bak(self): + # Corrupt main file -> renamed to .corrupt.* and restored from + # .bak. (The missing-file else branch at 1046-1047 is + # unreachable by construction: load_state_strict returns + # (empty state, None) for a missing file, which takes the + # early "state 完好" return; recorded in the gap ledger.) + with tempfile.TemporaryDirectory() as tmp: + state_path = os.path.join(tmp, "host-state.json") + with open(state_path, "w", encoding="utf-8") as f: + f.write("{not json") + with open(state_path + ".bak", "w", encoding="utf-8") as f: + json.dump({"version": 3, "tasks": {}}, f) + ok, msg = restore_state(state_path) + self.assertTrue(ok) + self.assertIn(".bak", msg) + with open(state_path, encoding="utf-8") as f: + self.assertEqual(json.load(f)["version"], 3) + + def _git_result(self, returncode=0, stdout="", stderr=""): + return SimpleNamespace(returncode=returncode, stdout=stdout, + stderr=stderr) + + def test_merge_commit_failure(self): + # 2019 + def _git(args, cwd): + if args[0] == "commit": + return self._git_result(1, "", "boom") + if args[0] == "rev-parse": + return self._git_result(0, "abc\n", "") + return self._git_result(0, "", "") + + with mock.patch.object(host, "git", side_effect=_git): + ok, msg = merge_task("/tmp", "/tmp", "b", "t", + verify_argv=[]) + self.assertFalse(ok) + self.assertIn("提交失败", msg) + + def test_merge_retry_abort(self): + # 2032-2033 + calls = [] + + def _git(args, cwd): + calls.append(args) + if args[0] == "rev-parse": + return self._git_result(0, "abc\n", "") + if args[0] == "merge" and "-X" not in args: + return self._git_result(1, "", "conflict") + if args[0] == "merge" and "-X" in args: + return self._git_result(1, "", "still conflict") + return self._git_result(0, "", "") + + with mock.patch.object(host, "git", side_effect=_git): + ok, msg = merge_task("/tmp", "/tmp", "b", "t", + verify_argv=[]) + self.assertFalse(ok) + self.assertIn("已 abort", msg) + + def test_merge_missing_lines_pyc_and_broken_link(self): + # 2048, 2055-2056 + with tempfile.TemporaryDirectory() as tmp: + repo = os.path.join(tmp, "repo") + os.makedirs(repo) + with open(os.path.join(repo, "skip.pyc"), "wb") as f: + f.write(b"\x00") + os.symlink("/nonexistent-target-xyz", + os.path.join(repo, "broken.txt")) + with open(os.path.join(repo, "real.txt"), "w", + encoding="utf-8") as f: + f.write("hello\n") + + def _git(args, cwd): + if args[0] == "rev-parse": + return self._git_result(0, "abc\n", "") + if args[0] == "merge" and "-X" not in args: + return self._git_result(1, "", "conflict") + if args[0] == "merge" and "-X" in args: + return self._git_result(0, "", "") + if args[0] == "merge-base": + return self._git_result(0, "base\n", "") + if args[0] == "diff": + return self._git_result(0, "+zzz-not-present-xyz\n", + "") + return self._git_result(0, "", "") + + with mock.patch.object(host, "git", side_effect=_git): + ok, msg = merge_task(repo, os.path.join(tmp, "wt"), + "b", "t", verify_argv=[]) + self.assertFalse(ok) + self.assertIn("改动被丢弃", msg) + + def test_notify_timeout_and_generic_failure(self): + # 2191, 2197-2198 + with tempfile.TemporaryDirectory() as tmp: + state_path = os.path.join(tmp, "host-state.json") + cfg = {"notify_command": "echo {task} {status} {text}"} + with mock.patch("subprocess.run", + side_effect=subprocess.TimeoutExpired( + "echo", 30)): + msg = notify_failure(state_path, cfg, "t1", + {"status": "failed", "text": "x"}) + self.assertTrue(msg.startswith("FAILED t1")) + with mock.patch("subprocess.run", + side_effect=RuntimeError("run down")): + msg = notify_failure(state_path, cfg, "t1", + {"status": "failed", "text": "x"}) + self.assertTrue(msg.startswith("FAILED t1")) + + def test_create_pr_push_and_gh_failures(self): + # 2322, 2331 + task = {"name": "t1", "branch": "host/t1", + "status": "completed"} + with mock.patch.object(host, "_github_remote_url", + return_value="https://example.com/r.git"): + with mock.patch.object( + host, "git", + return_value=self._git_result(1, "", "push boom")): + with self.assertRaises(RuntimeError): + create_pr("/tmp", dict(task)) + with mock.patch.object( + host, "git", + return_value=self._git_result(0, "", "")): + with mock.patch.object( + host, "run_gh", + return_value=self._git_result(1, "", "gh boom")): + with self.assertRaises(RuntimeError): + create_pr("/tmp", dict(task)) + + def test_parse_pr_checks_short_tab_line(self): + # 2356 + checks, all_pass = parse_pr_checks("build\t\n") + self.assertEqual(checks, []) + self.assertFalse(all_pass) + + def test_dispatch_from_branch_worktree_failure(self): + # 2453 + with tempfile.TemporaryDirectory() as tmp: + with mock.patch.object(host, "HERE", tmp): + def _git(args, cwd): + if args[0] == "worktree": + return self._git_result(1, "", "wt fail") + return self._git_result(0, "", "") + + with mock.patch.object(host, "git", side_effect=_git): + out, err = dispatch_from_branch( + SimpleNamespace(), {"tasks": {}}, "t1", + "do work", tmp, "main", tmp, None, None, + "workspace-write", "never") + self.assertIsNone(out) + self.assertIn("失败", err) + + +class MainBranchTestCase(unittest.TestCase): + def _state_cfg(self, tmp, tasks): + state_path = os.path.join(tmp, "host-state.json") + cfg_path = os.path.join(tmp, "host-config.json") + with open(state_path, "w", encoding="utf-8") as f: + json.dump({"tasks": tasks, "automations": []}, f) + with open(cfg_path, "w", encoding="utf-8") as f: + json.dump({}, f) + return state_path, cfg_path + + def _run_main(self, argv, patches=()): + with contextlib.ExitStack() as stack: + for p in patches: + stack.enter_context(p) + buf = io.StringIO() + with contextlib.redirect_stdout(buf): + with mock.patch.object(sys, "argv", argv): + host.main() + return buf.getvalue() + + def test_pr_missing_repo_exits(self): + # 3292-3293 + with tempfile.TemporaryDirectory() as tmp: + state_path, cfg_path = self._state_cfg( + tmp, {"t1": {"name": "t1"}}) + argv = ["host.py", "--state", state_path, + "--config", cfg_path, "pr", "--task", "t1"] + with self.assertRaises(SystemExit) as cm: + self._run_main(argv) + self.assertEqual(cm.exception.code, 1) + + def test_daemon_status_bad_pid(self): + # 3376-3377 + with tempfile.TemporaryDirectory() as tmp: + state_path, cfg_path = self._state_cfg(tmp, {}) + with open(os.path.join(tmp, "host-daemon.pid"), "w", + encoding="utf-8") as f: + f.write("abc") + argv = ["host.py", "--state", state_path, + "--config", cfg_path, "daemon-status"] + out = self._run_main(argv) + self.assertFalse(json.loads(out)["alive"]) + + def test_daemon_status_log_is_dir(self): + # 3383-3384 + with tempfile.TemporaryDirectory() as tmp: + state_path, cfg_path = self._state_cfg(tmp, {}) + with open(os.path.join(tmp, "host-daemon.pid"), "w", + encoding="utf-8") as f: + f.write(str(os.getpid())) + os.makedirs(os.path.join(tmp, "host-daemon.log")) + argv = ["host.py", "--state", state_path, + "--config", cfg_path, "daemon-status"] + out = self._run_main(argv) + data = json.loads(out) + self.assertTrue(data["alive"]) + self.assertEqual(data["log_tail"], "") + + def test_daemon_stop_inner_break(self): + # 3407-3408 + proc = subprocess.Popen(["sleep", "30"]) + try: + with tempfile.TemporaryDirectory() as tmp: + state_path, cfg_path = self._state_cfg(tmp, {}) + with open(os.path.join(tmp, "host-daemon.pid"), "w", + encoding="utf-8") as f: + f.write(str(proc.pid)) + argv = ["host.py", "--state", state_path, + "--config", cfg_path, "daemon-stop"] + out = self._run_main(argv) + self.assertIn("daemon 已停", out) + self.assertFalse(os.path.exists( + os.path.join(tmp, "host-daemon.pid"))) + finally: + try: + proc.kill() + except Exception: + pass + proc.wait() + + def test_daemon_stop_outer_except(self): + # 3411-3412 + with tempfile.TemporaryDirectory() as tmp: + state_path, cfg_path = self._state_cfg(tmp, {}) + pidfile = os.path.join(tmp, "host-daemon.pid") + with open(pidfile, "w", encoding="utf-8") as f: + f.write("999999") + argv = ["host.py", "--state", state_path, + "--config", cfg_path, "daemon-stop"] + out = self._run_main(argv) + self.assertIn("daemon 已停", out) + self.assertFalse(os.path.exists(pidfile)) + + def test_daemon_stop_remove_failure(self): + # 3415-3416 + with tempfile.TemporaryDirectory() as tmp: + state_path, cfg_path = self._state_cfg(tmp, {}) + with open(os.path.join(tmp, "host-daemon.pid"), "w", + encoding="utf-8") as f: + f.write("999999") + argv = ["host.py", "--state", state_path, + "--config", cfg_path, "daemon-stop"] + out = self._run_main( + argv, patches=[mock.patch( + "os.remove", side_effect=OSError("remove down"))]) + self.assertIn("daemon 已停", out) + + def test_daemon_stale_pid_daemonize_false(self): + # 3431-3435 + with tempfile.TemporaryDirectory() as tmp: + state_path, cfg_path = self._state_cfg(tmp, {}) + with open(os.path.join(tmp, "host-daemon.pid"), "w", + encoding="utf-8") as f: + f.write("999999") + argv = ["host.py", "--state", state_path, + "--config", cfg_path, "daemon"] + out = self._run_main( + argv, patches=[mock.patch.object( + host, "daemonize", return_value=False)]) + self.assertIn("daemon 已在后台启动", out) + + def test_verify_argv_empty_errors(self): + # 3524 + with tempfile.TemporaryDirectory() as tmp: + state_path, cfg_path = self._state_cfg( + tmp, {"t1": {"name": "t1", "repo": "/tmp", + "branch": "b", "cwd": tmp}}) + argv = ["host.py", "--state", state_path, + "--config", cfg_path, "merge", "--task", "t1", + "--verify-argv"] + with self.assertRaises(SystemExit) as cm: + self._run_main(argv) + self.assertEqual(cm.exception.code, 2) + + def test_verify_audit_dispatch(self): + # 3526-3527 + with tempfile.TemporaryDirectory() as tmp: + state_path, cfg_path = self._state_cfg(tmp, {}) + argv = ["host.py", "--state", state_path, + "--config", cfg_path, "verify-audit"] + audit_mock = mock.Mock() + self._run_main( + argv, patches=[mock.patch.object( + host, "_cmd_verify_audit", audit_mock)]) + audit_mock.assert_called_once() + + def test_status_untracked_contents(self): + # 3601 + with tempfile.TemporaryDirectory() as tmp: + state_path, cfg_path = self._state_cfg( + tmp, {"t1": {"name": "t1", "cwd": tmp}}) + argv = ["host.py", "--state", state_path, + "--config", cfg_path, "review", "--task", "t1"] + out = self._run_main(argv, patches=[ + mock.patch.object(host, "Server", + lambda **kw: _FakeServer()), + mock.patch.object(host, "task_diff", return_value={ + "status": "S", "untracked": "U", "diff": "D", + "untracked_contents": {"f.txt": "hello"}}), + mock.patch.object(host, "open_review_comments", + return_value=[]), + ]) + self.assertIn("--- untracked file: f.txt ---", out) + self.assertIn("hello", out) + + def test_merge_verify_argv_warning(self): + # 3620 + with tempfile.TemporaryDirectory() as tmp: + state_path, cfg_path = self._state_cfg( + tmp, {"t1": {"name": "t1", "repo": "/tmp", + "branch": "b", "cwd": tmp}}) + argv = ["host.py", "--state", state_path, + "--config", cfg_path, "merge", "--task", "t1", + "--verify", "echo hi", "--verify-argv", "echo", "hi"] + out = self._run_main(argv, patches=[ + mock.patch.object(host, "Server", + lambda **kw: _FakeServer()), + mock.patch.object(host, "merge_task", + return_value=(True, "ok")), + ]) + self.assertIn("verify_argv 已指定", out) + with open(state_path, encoding="utf-8") as f: + self.assertEqual(json.load(f)["tasks"]["t1"]["status"], + "merged") + + def test_fix_review_no_comments(self): + # 3637-3638 + with tempfile.TemporaryDirectory() as tmp: + state_path, cfg_path = self._state_cfg( + tmp, {"t1": {"name": "t1", "thread_id": "th1"}}) + argv = ["host.py", "--state", state_path, + "--config", cfg_path, "fix-review", "--task", "t1"] + out = self._run_main(argv, patches=[ + mock.patch.object(host, "Server", + lambda **kw: _FakeServer()), + mock.patch.object(host, "open_review_comments", + return_value=[]), + ]) + self.assertIn("没有未处理的评审评论", out) + + def test_interrupt_remove_failure(self): + # 3929-3931 + with tempfile.TemporaryDirectory() as tmp: + state_path, cfg_path = self._state_cfg(tmp, { + "t1": {"name": "t1", "thread_id": "th1", + "turns": [{"result": {"turn_id": "turn9"}}]}}) + req_dir = os.path.join(tmp, "interrupt-requests") + os.makedirs(req_dir) + with open(os.path.join(req_dir, "t1.json"), "w", + encoding="utf-8") as f: + f.write("{}") + argv = ["host.py", "--state", state_path, + "--config", cfg_path, "interrupt", "--task", "t1"] + out = self._run_main(argv, patches=[ + mock.patch.object(host, "Server", + lambda **kw: _FakeServer()), + mock.patch("os.remove", + side_effect=OSError("remove down")), + ]) + self.assertIn("turn=turn9", out) + with open(state_path, encoding="utf-8") as f: + self.assertEqual(json.load(f)["tasks"]["t1"]["status"], + "interrupted") + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_v2cov_gaps_d5.py b/tests/test_v2cov_gaps_d5.py new file mode 100644 index 0000000..50990ec --- /dev/null +++ b/tests/test_v2cov_gaps_d5.py @@ -0,0 +1,233 @@ +"""Coverage batch v4-D5: host.py leftover branches (wait-turn first +slice, run-turn interrupt raise, daemon-stop inner break, daemon +subcommand failure, interrupt poll paths, goal verify warning). + +Stubbing discipline: the function under test is never stubbed. Patched +collaborators are: time.monotonic / time.sleep / os.kill / +subprocess.run / interrupt_turn / _wait_turn / dispatch / +run_goal_loop / json.load (sinks, clocks and predicates). Every +assertion reads a real return value or real stdout. +""" + +import contextlib +import io +import json +import os +import queue +import signal +import sys +import tempfile +import threading +import time +import unittest +from pathlib import Path +from unittest import mock + +REPO_ROOT = Path(__file__).resolve().parents[1] +sys.path.insert(0, str(REPO_ROOT)) + +import host +from host import Server, _wait_turn + + +def _server(): + srv = object.__new__(Server) + srv.proc = None + srv.codex_bin = "codex" + srv.codex_home = None + srv.approval = "deny" + srv.model_provider = None + srv.progress_interval = 30 + srv.send_lock = threading.Lock() + srv.next_id = 1 + srv.pending = {} + srv.pending_lock = threading.Lock() + srv.turn_queues = {} + srv.events = [] + srv.events_lock = threading.Lock() + srv.restart_count = 0 + srv.trace_path = None + srv._trace_lines = 0 + return srv + + +class _FakeServer: + def __init__(self, *args, **kwargs): + pass + + def initialize(self): + pass + + def set_trace_file(self, *args, **kwargs): + pass + + def close(self): + pass + + def resume_thread(self, thread_id): + pass + + def interrupt_turn(self, thread_id, turn_id): + return {"error": "x"} + + +class WaitTurnTestCase(unittest.TestCase): + def test_first_slice_timeout(self): + # 154: elapsed already past timeout before the first slice. + q = queue.Queue() + with mock.patch.object( + host.time, "monotonic", + side_effect=[100.0, 100.5]): + self.assertEqual(_wait_turn(q, 0.1), ("timeout", None)) + + +class RunTurnInterruptRaiseTestCase(unittest.TestCase): + def test_interrupt_turn_raises(self): + # 777-780 + srv = _server() + srv.request = mock.Mock(return_value={ + "result": {"turn": {"id": "turn1"}}}) + srv.interrupt_turn = mock.Mock( + side_effect=RuntimeError("interrupt down")) + with mock.patch.object(host, "_wait_turn", + return_value=("interrupted-requested", + None)): + out = srv.run_turn("th1", "do work") + self.assertEqual(out["status"], "interrupted") + + +class MainLeftoverTestCase(unittest.TestCase): + def _state_cfg(self, tmp, tasks): + state_path = os.path.join(tmp, "host-state.json") + cfg_path = os.path.join(tmp, "host-config.json") + with open(state_path, "w", encoding="utf-8") as f: + json.dump({"tasks": tasks, "automations": []}, f) + with open(cfg_path, "w", encoding="utf-8") as f: + json.dump({}, f) + return state_path, cfg_path + + def _run_main(self, argv, patches=()): + with contextlib.ExitStack() as stack: + for p in patches: + stack.enter_context(p) + buf = io.StringIO() + with contextlib.redirect_stdout(buf): + with mock.patch.object(sys, "argv", argv): + host.main() + return buf.getvalue() + + def test_daemon_stop_inner_break(self): + # 3407-3408: SIGTERM succeeds, the liveness probe then raises. + # (A real sleep child would stay a zombie until reaped, so the + # probe would succeed and the break would never fire; the + # kill calls are therefore faked for this branch.) + def _fake_kill(pid, sig): + if sig == 0: + raise ProcessLookupError("gone") + + with tempfile.TemporaryDirectory() as tmp: + state_path, cfg_path = self._state_cfg(tmp, {}) + pidfile = os.path.join(tmp, "host-daemon.pid") + with open(pidfile, "w", encoding="utf-8") as f: + f.write("12345") + argv = ["host.py", "--state", state_path, + "--config", cfg_path, "daemon-stop"] + out = self._run_main( + argv, patches=[mock.patch("os.kill", + side_effect=_fake_kill)]) + self.assertIn("daemon 已停", out) + self.assertFalse(os.path.exists(pidfile)) + + def test_daemon_subcommand_failure(self): + # 3504-3505 + with tempfile.TemporaryDirectory() as tmp: + state_path, cfg_path = self._state_cfg(tmp, {}) + argv = ["host.py", "--state", state_path, + "--config", cfg_path, "daemon", "--foreground", + "--interval", "300"] + old_term = signal.getsignal(signal.SIGTERM) + old_int = signal.getsignal(signal.SIGINT) + + def sender(): + time.sleep(0.5) + try: + os.kill(os.getpid(), signal.SIGTERM) + except Exception: + pass + + sender_thread = threading.Thread(target=sender, daemon=True) + try: + sender_thread.start() + out = self._run_main(argv, patches=[ + mock.patch("subprocess.run", + side_effect=RuntimeError("child down")), + ]) + finally: + sender_thread.join(timeout=5) + signal.signal(signal.SIGTERM, old_term) + signal.signal(signal.SIGINT, old_int) + self.assertIn("daemon subcommand", out) + + def test_interrupt_poll_responded(self): + # 3963-3964 + with tempfile.TemporaryDirectory() as tmp: + state_path, cfg_path = self._state_cfg(tmp, { + "t1": {"name": "t1", "thread_id": "th1", + "status": "interrupted", + "turns": [{"result": {"turn_id": "turn9"}}]}}) + argv = ["host.py", "--state", state_path, + "--config", cfg_path, "interrupt", "--task", "t1"] + out = self._run_main(argv, patches=[ + mock.patch.object(host, "Server", + lambda **kw: _FakeServer()), + mock.patch("time.sleep", lambda *a, **k: None), + ]) + self.assertIn("属主已响应", out) + + def test_interrupt_poll_read_failure(self): + # 3965-3967 (+ else branch): state reads fail during polling. + with tempfile.TemporaryDirectory() as tmp: + state_path, cfg_path = self._state_cfg(tmp, { + "t1": {"name": "t1", "thread_id": "th1", + "status": "running", + "turns": [{"result": {"turn_id": "turn9"}}]}}) + argv = ["host.py", "--state", state_path, + "--config", cfg_path, "interrupt", "--task", "t1"] + real_load = json.load + calls = {"n": 0} + + def _load(fp, *a, **k): + calls["n"] += 1 + if calls["n"] <= 2: + return real_load(fp, *a, **k) + raise RuntimeError("poll read down") + + out = self._run_main(argv, patches=[ + mock.patch.object(host, "Server", + lambda **kw: _FakeServer()), + mock.patch("time.sleep", lambda *a, **k: None), + mock.patch("json.load", side_effect=_load), + ]) + self.assertIn("10 秒内未响应", out) + + def test_goal_verify_argv_warning(self): + # 4025 + with tempfile.TemporaryDirectory() as tmp: + state_path, cfg_path = self._state_cfg(tmp, {}) + argv = ["host.py", "--state", state_path, + "--config", cfg_path, "goal", "--name", "g", + "--task", "t", "--repo", "/tmp", + "--verify", "echo hi", "--verify-argv", "echo", "hi"] + out = self._run_main(argv, patches=[ + mock.patch.object(host, "Server", + lambda **kw: _FakeServer()), + mock.patch.object(host, "dispatch", + return_value={"name": "g"}), + mock.patch.object(host, "run_goal_loop", + return_value=None), + ]) + self.assertIn("verify_argv 已指定", out) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_v2cov_gaps_d6.py b/tests/test_v2cov_gaps_d6.py new file mode 100644 index 0000000..da8dedb --- /dev/null +++ b/tests/test_v2cov_gaps_d6.py @@ -0,0 +1,104 @@ +"""Coverage batch v4-D6: host.py parallel-branch fallback paths +(4186 dep-branch dispatch failure, 4198 dep cwd copy). + +Stubbing discipline: the branch code under test is never stubbed. +Patched collaborators are dispatch, dispatch_from_branch, +run_task_turn_with_retry and copy_tree_excluding_git (all of them +sinks the branch delegates to). Assertions read real stdout and the +real copy_tree call record. +""" + +import contextlib +import io +import json +import os +import sys +import tempfile +import unittest +from pathlib import Path +from unittest import mock + +REPO_ROOT = Path(__file__).resolve().parents[1] +sys.path.insert(0, str(REPO_ROOT)) + +import host + + +class _FakeServer: + def __init__(self, *args, **kwargs): + pass + + def initialize(self): + pass + + def set_trace_file(self, *args, **kwargs): + pass + + def close(self): + pass + + +class ParallelFallbackTestCase(unittest.TestCase): + def test_dep_branch_dispatch_falls_back_to_head_plus_copy(self): + # 4186 + 4198 + with tempfile.TemporaryDirectory() as tmp: + dep_dir = os.path.join(tmp, "dep-cwd") + os.makedirs(dep_dir) + with open(os.path.join(dep_dir, "seed.txt"), "w") as f: + f.write("seed") + new_dir = os.path.join(tmp, "new-cwd") + jobs_path = os.path.join(tmp, "jobs.json") + jobs = [ + {"name": "j1", "task": "t1", "repo": "/tmp/repo1"}, + {"name": "j2", "task": "t2", "repo": "/tmp/repo1", + "depends_on": ["j1"]}, + ] + with open(jobs_path, "w", encoding="utf-8") as f: + json.dump(jobs, f) + state_path = os.path.join(tmp, "host-state.json") + cfg_path = os.path.join(tmp, "host-config.json") + with open(state_path, "w", encoding="utf-8") as f: + json.dump({"tasks": {}, "automations": []}, f) + with open(cfg_path, "w", encoding="utf-8") as f: + json.dump({}, f) + + def _dispatch(srv, state, name, task, repo, *a, **k): + if name == "j1": + return {"name": "j1", "status": "completed", + "branch": "b1", "cwd": dep_dir} + return {"name": "j2", "status": "completed", + "cwd": new_dir} + + copy_calls = [] + argv = ["host.py", "--state", state_path, + "--config", cfg_path, "parallel", + "--jobs", jobs_path] + buf = io.StringIO() + with contextlib.redirect_stdout(buf): + with mock.patch.object(sys, "argv", argv): + with mock.patch.object( + host, "Server", + lambda **kw: _FakeServer()): + with mock.patch.object( + host, "dispatch", + side_effect=_dispatch): + with mock.patch.object( + host, "dispatch_from_branch", + return_value=(None, "boom")): + with mock.patch.object( + host, + "run_task_turn_with_retry", + return_value={"status": "completed", + "text": "ok"}): + with mock.patch.object( + host, + "copy_tree_excluding_git", + side_effect=lambda s, d: copy_calls.append((s, d))): + host.main() + out = buf.getvalue() + self.assertIn("从依赖分支派发失败", out) + self.assertEqual(copy_calls, [(dep_dir, new_dir)]) + + +if __name__ == "__main__": + unittest.main()