diff --git a/cloud/tests/test_v6_a8_worker_gaps.py b/cloud/tests/test_v6_a8_worker_gaps.py new file mode 100644 index 0000000..12fc940 --- /dev/null +++ b/cloud/tests/test_v6_a8_worker_gaps.py @@ -0,0 +1,425 @@ +"""code-to10 v6 A8:worker 缺口补测。 + +补测基线(v5 头 10023fb 全量覆盖率,cloud/worker.py 3402 +语句缺 338 行,90%):本文件按“假执行器/假派发”补下列 +区间,其余缺口行以全量复测为准: +- 2092-2095:process_auto_memories 的 extracted 项审计。 +- 2936-2944:_tenant_dirs 技能播种失败只记日志不阻塞, + 且日志本身抛错也被吞掉。 +- 3092-3111(及 checkout 失败行):_prepare_environment + 的 clone/非 git/get-url/fetch/checkout 失败分支。 +- 6626-6652:_run_pr_jobs 的 PrSkipped/PrStepError/通用 + 异常三路失败写回。 +- 6977-7010:run_loop 退出排空(杀 proc、杀容器、标 + interrupted、审计),含 project_id 为 NULL 与审计查询 + 抛错两路 _pid 回落。 + +绝不碰(任务口径不碰项,单列): +- claim/goal_resume_pending 的 except-pass 分支。 +- MCP handler-error 通知分支。 +只加测试,不改生产代码;假的是执行器/派发/播种等外部 +依赖,不是被测代码本身;不加 no-coverage pragma、不 skip。 +""" + +import os +import sqlite3 +import sys +import tempfile +import threading +import unittest +from pathlib import Path +from types import SimpleNamespace + +try: + import cloud.worker as worker_mod + from cloud.db import connect as _db_connect + from cloud.pr import PrSkipped, PrStepError + from cloud.pr_exec import enqueue_pr_job + from cloud.server import init_db + from cloud.worker import Worker +except ImportError: # discover -s cloud/tests 时把项目根目录补进 sys.path + sys.path.insert(0, str(Path(__file__).resolve().parents[2])) + import cloud.worker as worker_mod + from cloud.db import connect as _db_connect + from cloud.pr import PrSkipped, PrStepError + from cloud.pr_exec import enqueue_pr_job + from cloud.server import init_db + from cloud.worker import Worker + + +class _FakeSync: + """同步假执行器:不走 Local/Docker/Cloud,只给 run_loop + 的排空路径一个带 kill_container 的执行器替身。""" + + def dispatch_task(self, task): + return {"status": "completed"} + + def run_task(self, task): + return {"status": "completed"} + + def kill_container(self, name): + raise RuntimeError("kill boom") + + +class _FakeProc: + """pid 不存在的假进程:_kill_proc_group 在 getpgid 处即 + 返回(不真杀),wait 抛错走排空的 except 分支。""" + + pid = 999999 + + def wait(self, timeout=None): + raise RuntimeError("wait boom") + + +class _RaisingLogger: + """warning 一律抛错的假 logger:测 _tenant_dirs 里 + “日志失败也被吞掉”的内层 except。""" + + def warning(self, *args, **kwargs): + raise RuntimeError("log boom") + + def info(self, *args, **kwargs): + pass + + def debug(self, *args, **kwargs): + pass + + def error(self, *args, **kwargs): + pass + + +def _insert_task(db_path, task_id, project_id="p1", status="queued"): + conn = sqlite3.connect(db_path) + try: + conn.execute( + "INSERT INTO tasks(id, project_id, prompt, status," + " idempotency_key, depends_on, priority, created_by," + " created_at, updated_at, last_heartbeat)" + " VALUES(?, ?, ?, ?, NULL, '[]', 0, 'tester', 0, 0, NULL)", + (task_id, project_id, "do work", status), + ) + conn.commit() + finally: + conn.close() + + +def _row(db_path, sql): + conn = sqlite3.connect(db_path) + try: + return conn.execute(sql).fetchone() + finally: + conn.close() + + +class WorkerGapsBase(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.addCleanup(self.tmp.cleanup) + self.db_path = os.path.join(self.tmp.name, "control.db") + init_db(self.db_path) + self.hb = os.path.join(self.tmp.name, "hb.json") + + def _worker(self, **kwargs): + kwargs.setdefault("env", {"XCM_API_KEY": "test-placeholder"}) + kwargs.setdefault("heartbeat_file", self.hb) + kwargs.setdefault("max_retries", 0) + kwargs.setdefault("retry_backoff_seconds", 0) + return Worker(self.db_path, "/fake/host.py", **kwargs) + + +class ProcessAutoMemoriesTestCase(WorkerGapsBase): + def test_extracted_item_audited(self): + worker = self._worker() + orig = worker_mod._auto_process_due + worker_mod._auto_process_due = lambda db_path, **kw: [ + {"status": "extracted", "task_id": "T1"}, + {"status": "other", "task_id": "T2"}, + ] + try: + results = worker.process_auto_memories() + finally: + worker_mod._auto_process_due = orig + self.assertEqual(len(results), 2) + row = _row(self.db_path, + "SELECT action, detail FROM audit_log" + " WHERE action = 'worker_memory_extracted'") + self.assertEqual(row, ("worker_memory_extracted", "T1")) + + def test_sweep_failure_returns_empty(self): + worker = self._worker() + orig = worker_mod._auto_process_due + worker_mod._auto_process_due = lambda db_path, **kw: (_ for _ in ()).throw( + RuntimeError("sweep boom")) + try: + self.assertEqual(worker.process_auto_memories(), []) + finally: + worker_mod._auto_process_due = orig + + +class TenantDirsSkillFailureTestCase(WorkerGapsBase): + def _seeded_worker(self): + tenants = os.path.join(self.tmp.name, "tenants") + template_home = os.path.join(self.tmp.name, "template-home") + os.makedirs(template_home, exist_ok=True) + with open(os.path.join(template_home, "config.toml"), "w") as f: + f.write("# template\n") + worker = self._worker(tenants_root=tenants, + tenant_template_home=template_home, + seed_skills=True) + # 预置 codex-home 配置与非空 test-repo:播种分支只测 + # 技能失败本身,不走到 test-repo 模板复制/报错。 + base = os.path.join(tenants, "p1") + os.makedirs(os.path.join(base, "codex-home"), exist_ok=True) + with open(os.path.join(base, "codex-home", "config.toml"), + "w") as f: + f.write("# existing\n") + os.makedirs(os.path.join(base, "test-repo"), exist_ok=True) + with open(os.path.join(base, "test-repo", "sample.py"), + "w") as f: + f.write("# sample\n") + return worker + + def test_skill_failure_only_logged_not_blocking(self): + worker = self._seeded_worker() + orig = worker_mod._ensure_skills + worker_mod._ensure_skills = lambda home: (_ for _ in ()).throw( + RuntimeError("skill boom")) + try: + dirs = worker._tenant_dirs("p1") + finally: + worker_mod._ensure_skills = orig + self.assertTrue(os.path.isdir(dirs["test_repo"])) + self.assertTrue(os.path.exists( + os.path.join(dirs["test_repo"], "sample.py"))) + + def test_skill_failure_and_logger_failure_swallowed(self): + worker = self._seeded_worker() + worker.logger = _RaisingLogger() + orig = worker_mod._ensure_skills + worker_mod._ensure_skills = lambda home: (_ for _ in ()).throw( + RuntimeError("skill boom")) + try: + dirs = worker._tenant_dirs("p1") + finally: + worker_mod._ensure_skills = orig + self.assertTrue(os.path.isdir(dirs["work"])) + + +class PrepareEnvironmentTestCase(WorkerGapsBase): + def _env_worker(self, runner): + repos = os.path.join(self.tmp.name, "repos") + return self._worker(repos_dir=repos, repos_runner=runner), repos + + def test_clone_failure(self): + def runner(cmd): + return SimpleNamespace(returncode=1, stdout="", + stderr="clone boom") + worker, _repos = self._env_worker(runner) + err = worker._prepare_environment( + {"id": "env1", "repo_url": "https://example.invalid/x.git"}) + self.assertIn("git clone 失败", err) + self.assertIn("clone boom", err) + + def test_existing_dir_not_a_git_repo(self): + worker, repos = self._env_worker(lambda cmd: SimpleNamespace( + returncode=0, stdout="", stderr="")) + target = os.path.join(repos, "envs", "env2") + os.makedirs(target, exist_ok=True) + err = worker._prepare_environment( + {"id": "env2", "repo_url": "https://example.invalid/x.git"}) + self.assertIn("不是 git 仓库", err) + + def _git_repo_worker(self, behaviour): + worker, repos = self._env_worker(behaviour) + target = os.path.join(repos, "envs", "env3") + os.makedirs(os.path.join(target, ".git"), exist_ok=True) + return worker + + def test_get_url_failure(self): + def runner(cmd): + return SimpleNamespace(returncode=1, stdout="", + stderr="no remote") + worker = self._git_repo_worker(runner) + err = worker._prepare_environment( + {"id": "env3", "repo_url": "https://example.invalid/x.git"}) + self.assertIn("取不到 remote origin", err) + + def test_origin_mismatch(self): + def runner(cmd): + if "get-url" in cmd: + return SimpleNamespace( + returncode=0, stdout="https://other.invalid/y.git\n", + stderr="") + return SimpleNamespace(returncode=0, stdout="", stderr="") + worker = self._git_repo_worker(runner) + err = worker._prepare_environment( + {"id": "env3", "repo_url": "https://example.invalid/x.git"}) + self.assertIn("origin 与环境 repo_url 不一致", err) + + def test_fetch_failure(self): + def runner(cmd): + if "get-url" in cmd: + return SimpleNamespace( + returncode=0, + stdout="https://example.invalid/x.git\n", stderr="") + if "fetch" in cmd: + return SimpleNamespace(returncode=1, stdout="", + stderr="fetch boom") + return SimpleNamespace(returncode=0, stdout="", stderr="") + worker = self._git_repo_worker(runner) + err = worker._prepare_environment( + {"id": "env3", "repo_url": "https://example.invalid/x.git"}) + self.assertIn("git fetch origin 失败", err) + self.assertIn("fetch boom", err) + + def test_checkout_failure_and_success(self): + def runner(cmd): + if "get-url" in cmd: + return SimpleNamespace( + returncode=0, + stdout="https://example.invalid/x.git\n", stderr="") + if "fetch" in cmd: + return SimpleNamespace(returncode=0, stdout="", stderr="") + if "checkout" in cmd: + return SimpleNamespace(returncode=1, stdout="", + stderr="no such branch") + return SimpleNamespace(returncode=0, stdout="", stderr="") + worker = self._git_repo_worker(runner) + env = {"id": "env3", "repo_url": "https://example.invalid/x.git", + "branch": "nope"} + err = worker._prepare_environment(env) + self.assertIn("git checkout nope 失败", err) + self.assertIn("no such branch", err) + # 同一 runner 改走成功 checkout:返回 None(准备完成)。 + def ok_runner(cmd): + return SimpleNamespace(returncode=0, + stdout="https://example.invalid/x.git\n", + stderr="") + worker.repos_runner = ok_runner + self.assertIsNone(worker._prepare_environment(env)) + + +class RunPrJobsFailureTestCase(WorkerGapsBase): + def _pr_worker(self, failure): + worker = self._worker(env_key="TEST_KEY", + host_name="elsewhere-host-1") + conn = _db_connect(self.db_path) + try: + worker._ensure_result_columns(conn) + conn.execute( + "INSERT INTO tasks(id, project_id, prompt, status," + " depends_on, priority, created_by, created_at," + " updated_at, last_heartbeat) VALUES('t1', 'p1', 'x'," + " 'merged', '[]', 0, 'system', 0, 0, NULL)") + conn.commit() + enqueue_pr_job(conn, "t1", "p1", "elsewhere-host-1") + finally: + conn.close() + return worker + + def _run_with_failure(self, failure): + worker = self._pr_worker(failure) + orig = worker_mod.execute_task_pr + + def _fake(db_path, task_id, user_id, **kwargs): + raise failure + worker_mod.execute_task_pr = _fake + try: + results = worker._run_pr_jobs() + finally: + worker_mod.execute_task_pr = orig + return results + + def test_pr_skipped_marks_failed(self): + results = self._run_with_failure(PrSkipped("no longer merged")) + self.assertEqual(len(results), 1) + self.assertEqual(results[0]["status"], "failed") + self.assertIn("skipped: no longer merged", results[0]["error"]) + row = _row(self.db_path, + "SELECT status FROM pr_jobs WHERE task_id = 't1'") + self.assertEqual(row[0], "failed") + + def test_pr_step_error_marks_failed(self): + results = self._run_with_failure( + PrStepError("push", "remote denied")) + self.assertEqual(results[0]["error"], "push: remote denied") + row = _row(self.db_path, + "SELECT status FROM pr_jobs WHERE task_id = 't1'") + self.assertEqual(row[0], "failed") + + def test_generic_error_marks_failed(self): + results = self._run_with_failure(RuntimeError("boom")) + self.assertEqual(results[0]["error"], "boom") + row = _row(self.db_path, + "SELECT status FROM pr_jobs WHERE task_id = 't1'") + self.assertEqual(row[0], "failed") + + +class RunLoopDrainTestCase(WorkerGapsBase): + def _drain_worker(self, task_id, project_id): + _insert_task(self.db_path, task_id, project_id=project_id, + status="running") + worker = self._worker(executor=_FakeSync()) + worker._running_set(task_id, proc=_FakeProc(), container="c1") + worker.executor.kill_container = lambda name: (_ for _ in ()).throw( + RuntimeError("kill boom")) + return worker + + def _run_drain(self, worker): + stop = threading.Event() + stop.set() + results = worker.run_loop(interval=10, drain_timeout=0, + stop_event=stop) + self.assertEqual(results, []) + return worker + + def test_drain_kills_proc_and_container_marks_interrupted(self): + worker = self._drain_worker("t-drain", "p1") + self._run_drain(worker) + row = _row(self.db_path, + "SELECT status, interrupted_reason FROM tasks" + " WHERE id = 't-drain'") + self.assertEqual(row[0], "interrupted") + self.assertEqual(row[1], "worker_shutdown") + audit = _row(self.db_path, + "SELECT action, detail FROM audit_log" + " WHERE action = 'worker_shutdown_interrupted'") + self.assertEqual(audit, + ("worker_shutdown_interrupted", "t-drain")) + + def test_drain_null_project_id_audits_empty_project(self): + worker = self._drain_worker("t-null", None) + self._run_drain(worker) + audit = _row(self.db_path, + "SELECT action, detail FROM audit_log" + " WHERE action = 'worker_shutdown_interrupted'" + " AND detail = 't-null'") + self.assertIsNotNone(audit) + + def test_drain_audit_lookup_failure_still_interrupts(self): + worker = self._drain_worker("t-dberr", "p1") + orig_connect = worker_mod._db_connect + import inspect + + def _guarded(db_path): + # 只有 run_loop 本体直接发起的审计查询才抛错: + # _set_status/_mark_interrupted_reason 调 _db_connect + # 时直接调用方是各自方法名,不受影响;排空审计的 + # except 分支据此覆盖。 + caller = inspect.currentframe().f_back + if caller is not None and \ + caller.f_code.co_name == "run_loop": + raise RuntimeError("db gone") + return orig_connect(db_path) + worker_mod._db_connect = _guarded + try: + self._run_drain(worker) + finally: + worker_mod._db_connect = orig_connect + row = _row(self.db_path, + "SELECT status FROM tasks WHERE id = 't-dberr'") + self.assertEqual(row[0], "interrupted") + + +if __name__ == "__main__": + unittest.main()