From 3dc4ebe27373c6792c0770d9eb8fb4b0d8b231ac Mon Sep 17 00:00:00 2001 From: host Date: Sat, 31 Oct 2026 09:28:06 +0000 Subject: [PATCH] =?UTF-8?q?coverage85-step4d:=20host=20main=20=E5=B0=BE?= =?UTF-8?q?=E9=83=A8=E5=91=BD=E4=BB=A4=E5=9D=97=E6=B5=8B=E8=AF=95=EF=BC=88?= =?UTF-8?q?+30=20=E9=A1=B9=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- tests/test_host_coverage85_3.py | 775 ++++++++++++++++++++++++++++++++ 1 file changed, 775 insertions(+) create mode 100644 tests/test_host_coverage85_3.py diff --git a/tests/test_host_coverage85_3.py b/tests/test_host_coverage85_3.py new file mode 100644 index 0000000..f1800c5 --- /dev/null +++ b/tests/test_host_coverage85_3.py @@ -0,0 +1,775 @@ +"""覆盖率 78%->85% 第 4 步:host.py main() 剩余分支的真实行为测试。 + +进程内调 host.main()(--config/--state 均指临时文件), +app-server 一律用 FakeServer(不拉真实 codex 进程), +git 操作走临时仓库;daemon 前台循环用信号停、interrupt +慢路径实测 10 秒等待;绝不跑模型。 +""" + +import io +import json +import os +import signal +import subprocess +import sys +import tempfile +import threading +import time +import unittest +from contextlib import redirect_stdout +from unittest import mock + +import host + + +class FakeServer: + def __init__(self, **kwargs): + self.calls = [] + self.results = { + "thread/list": {"data": [{"id": "th-1"}]}, + "thread/read": {"thread": {"id": "th-1"}}, + "thread/name/set": {}, + "thread/archive": {}, + "thread/unarchive": {"thread": {"id": "th-1"}}, + "thread/compact/start": {}, + "skills/list": {}, + "app/list": {}, + "mcpServerStatus/list": {}, + "config/mcpServer/reload": {}, + "mcpServer/oauth/login": {}, + } + self.error_methods = set() + self.raise_systemexit = set() + self.raise_runtime = set() + self.interrupt_result = {"result": {}} + self.trace_path = None + + def initialize(self): + self.calls.append(("initialize",)) + return {} + + def close(self): + self.calls.append(("close",)) + + def set_trace_file(self, state_path, task_name): + self.trace_path = os.path.join( + os.path.dirname(state_path), "host-traces", + task_name + ".jsonl") + + def request(self, method, params=None, timeout=120): + self.calls.append(("request", method, params)) + if method in self.raise_systemexit: + raise SystemExit(3) + if method in self.raise_runtime: + raise RuntimeError("boom-" + method) + if method in self.error_methods: + return {"error": {"message": "boom"}} + return {"result": self.results.get(method, {})} + + def resume_thread(self, thread_id): + self.calls.append(("resume", thread_id)) + return {"thread": {"id": thread_id}} + + def archive_thread(self, thread_id): + self.calls.append(("archive", thread_id)) + if "thread/archive" in self.raise_systemexit: + raise SystemExit(3) + if "thread/archive" in self.raise_runtime: + raise RuntimeError("boom-thread/archive") + return {"result": {}} + + def list_threads(self): + return [{"id": "th-1"}] + + def create_thread(self, *args, **kwargs): + self.calls.append(("create_thread", args, kwargs)) + return "th-new" + + def fork_thread(self, thread_id, **kwargs): + self.calls.append(("fork", thread_id, kwargs)) + return "th-fork" + + def interrupt_turn(self, thread_id, turn_id): + self.calls.append(("interrupt", thread_id, turn_id)) + return self.interrupt_result + + def start_review(self, thread_id, target): + self.calls.append(("review", thread_id, target)) + return {"review": "ok"} + + +class Main3Base(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.addCleanup(self.tmp.cleanup) + self.config_path = os.path.join(self.tmp.name, "cfg.json") + self.state_path = os.path.join(self.tmp.name, "host-state.json") + with open(self.config_path, "w") as f: + json.dump({}, f) + self._old_argv = sys.argv + + def tearDown(self): + sys.argv = self._old_argv + + def _save_state(self, state): + host.save_state(self.state_path, state) + + def _load_state(self): + with open(self.state_path) as f: + return json.load(f) + + def _run(self, fake, *argv, **patches): + sys.argv = ["host.py", "--config", self.config_path, + "--state", self.state_path] + list(argv) + out = io.StringIO() + code = 0 + ctxs = [mock.patch.object(host, "Server", lambda **kw: fake)] + for name, val in patches.items(): + ctxs.append(mock.patch.object(host, name, val)) + with ctxs[0]: + if len(ctxs) > 1: + with ctxs[1]: + if len(ctxs) > 2: + with ctxs[2]: + code, out = self._invoke(out) + else: + code, out = self._invoke(out) + else: + code, out = self._invoke(out) + return code, out.getvalue() + + def _invoke(self, out): + code = 0 + with redirect_stdout(out): + try: + host.main() + except SystemExit as exc: + code = exc.code if isinstance(exc.code, int) else 1 + except RuntimeError: + code = "runtime-error" + return code, out + + def _git(self, *args, cwd=None): + subprocess.run(["git"] + list(args), cwd=cwd, check=True, + capture_output=True) + + def _make_repo(self): + repo = os.path.join(self.tmp.name, "repo") + os.makedirs(repo) + self._git("init", "-q", cwd=repo) + self._git("config", "user.email", "t@t.test", cwd=repo) + self._git("config", "user.name", "t", cwd=repo) + with open(os.path.join(repo, "f.txt"), "w") as f: + f.write("hello\n") + self._git("add", ".", cwd=repo) + self._git("commit", "-qm", "init", cwd=repo) + return repo + + +class EarlyBranchesTest(Main3Base): + def test_dispatch_without_repo_raises(self): + sys.argv = ["host.py", "--config", self.config_path, + "--state", self.state_path, + "dispatch", "--name", "x", "--task", "y"] + with self.assertRaises(RuntimeError) as ctx: + host.main() + self.assertIn("缺少 --repo", str(ctx.exception)) + + def test_install_cron_new_and_existing(self): + calls = [] + + def fake_run(cmd, **kw): + calls.append(cmd) + if cmd[:2] == ["crontab", "-l"]: + return mock.Mock(stdout="") + return mock.Mock(stdout="", returncode=0) + with mock.patch.object(host.subprocess, "run", fake_run): + code, out = self._run(FakeServer(), "automations", + "--install-cron") + self.assertEqual(code, 0) + self.assertIn("已安装 crontab", out) + self.assertIn("codex-host-run-due", out) + + def fake_run2(cmd, **kw): + if cmd[:2] == ["crontab", "-l"]: + return mock.Mock(stdout="*/30 * * * * x " + "codex-host-run-due\n") + raise AssertionError("不应再写 crontab") + with mock.patch.object(host.subprocess, "run", fake_run2): + code, out = self._run(FakeServer(), "automations", + "--install-cron") + self.assertIn("已有 codex-host-run-due", out) + + def test_github_workflow_success(self): + repo = self._make_repo() + with open(self.config_path, "w") as f: + json.dump({"repo": repo}, f) + code, out = self._run(FakeServer(), "github-workflow") + self.assertEqual(code, 0) + doc = json.loads(out) + self.assertTrue(doc["changed"]) + self.assertTrue(os.path.exists(doc["path"])) + code, out = self._run(FakeServer(), "github-workflow") + self.assertFalse(json.loads(out)["changed"]) + + def test_pr_variants(self): + self._save_state({"tasks": {"t1": {"name": "t1", + "thread_id": "th-1", + "status": "completed", + "repo": "/tmp"}}, + "automations": []}) + code, out = self._run( + FakeServer(), "pr", "--task", "t1", + create_pr=lambda repo, t: {"url": "https://pr/1"}) + self.assertEqual(code, 0) + self.assertIn("https://pr/1", out) + code, out = self._run( + FakeServer(), "pr-checks", "--task", "t1", + get_pr_checks=lambda t, repo=None: {"checks": []}) + self.assertEqual(code, 0) + self.assertIn("checks", out) + code, out = self._run( + FakeServer(), "pr-merge", "--task", "t1", + merge_pr=lambda t, repo=None: (True, "merged ok")) + self.assertEqual(code, 0) + self.assertIn("merged", out) + code, out = self._run( + FakeServer(), "pr-merge", "--task", "t1", + merge_pr=lambda t, repo=None: (False, "checks pending")) + self.assertEqual(code, 1) + self.assertIn("checks pending", out) + + def test_sleep_positive_and_keyboard_interrupt(self): + code, out = self._run(FakeServer(), "sleep", "--seconds", "0.5") + self.assertEqual(code, 0) + self.assertAlmostEqual(json.loads(out)["slept_seconds"], 0.5, + places=1) + with mock.patch.object(host.time, "sleep", + side_effect=KeyboardInterrupt): + code, out = self._run(FakeServer(), "sleep", + "--seconds", "5") + self.assertEqual(code, 130) + self.assertTrue(json.loads(out)["interrupted"]) + + def test_health_not_ok_exits_1(self): + code, out = self._run(FakeServer(), "--codex-bin", + "/nonexistent-bin", "health") + self.assertEqual(code, 1) + self.assertFalse(json.loads(out)["ok"]) + + def test_restore_state_failure_exits_1(self): + with open(self.state_path, "w") as f: + f.write("{broken json") + code, out = self._run(FakeServer(), "restore-state") + self.assertEqual(code, 1) + + def test_recover_corrupt_and_recovered(self): + with open(self.state_path, "w") as f: + f.write("{broken json") + code, out = self._run(FakeServer(), "recover") + self.assertEqual(code, 1) + self.assertIn("state 已损坏", out) + # save_state 拒绝覆盖损坏文件:先删掉再写新 state。 + os.remove(self.state_path) + self._save_state({"tasks": {"old": {"name": "old", + "status": "running", + "created_at": 1000}}, + "automations": []}) + code, out = self._run(FakeServer(), "recover", + "--older-than", "0") + self.assertEqual(code, 0) + self.assertEqual(json.loads(out), {"recovered": ["old"]}) + self.assertEqual( + self._load_state()["tasks"]["old"]["status"], + "interrupted") + + +class DaemonBranchesTest(Main3Base): + def _pidfile(self): + return os.path.join(self.tmp.name, "host-daemon.pid") + + def _logfile(self): + return os.path.join(self.tmp.name, "host-daemon.log") + + def test_daemon_status_alive_with_log_tail(self): + with open(self._pidfile(), "w") as f: + f.write(str(os.getpid())) + with open(self._logfile(), "w") as f: + f.write("\n".join("line-%d" % i for i in range(7))) + code, out = self._run(FakeServer(), "daemon-status") + self.assertEqual(code, 0) + doc = json.loads(out) + self.assertTrue(doc["alive"]) + self.assertEqual(doc["pid"], os.getpid()) + self.assertIn("line-6", doc["log_tail"]) + self.assertNotIn("line-0", doc["log_tail"]) + + def test_daemon_stop_live_process(self): + proc = subprocess.Popen(["sleep", "30"]) + try: + with open(self._pidfile(), "w") as f: + f.write(str(proc.pid)) + code, out = self._run(FakeServer(), "daemon-stop") + self.assertEqual(code, 0) + self.assertIn("daemon 已停", out) + self.assertFalse(os.path.exists(self._pidfile())) + self.assertIsNotNone(proc.poll()) + finally: + if proc.poll() is None: + proc.kill() + proc.wait() + + def test_daemon_already_running(self): + with open(self._pidfile(), "w") as f: + f.write(str(os.getpid())) + code, out = self._run(FakeServer(), "daemon") + self.assertEqual(code, 0) + self.assertIn("daemon 已在运行", out) + + def test_daemon_foreground_loop_stops_on_sigterm(self): + self._save_state({"tasks": {}, "automations": []}) + # 注意:daemon 循环的子进程只带 --config 不带 --state, + # 会写 host.py 所在目录的默认 host-state.json(含 .bak + # 轮转)。测试前快照、跑后恢复/删除,不污染 worktree。 + here = os.path.dirname(os.path.abspath(host.__file__)) + leaked = [os.path.join(here, n) for n in + ["host-state.json", "host-state.json.bak", + "host-state.json.bak.1", "host-state.json.bak.2", + "host-state.json.bak.3"]] + snapshot = {} + for path in leaked: + if os.path.exists(path): + with open(path, "rb") as f: + snapshot[path] = f.read() + old_term = signal.getsignal(signal.SIGTERM) + old_int = signal.getsignal(signal.SIGINT) + try: + def _killer(): + time.sleep(0.8) + os.kill(os.getpid(), signal.SIGTERM) + th = threading.Thread(target=_killer, daemon=True) + th.start() + code, out = self._run(FakeServer(), "daemon", + "--foreground", "--interval", "0") + self.assertEqual(code, 0) + self.assertIn("host daemon stopped", out) + self.assertIn("daemon tick", out) + finally: + signal.signal(signal.SIGTERM, old_term) + signal.signal(signal.SIGINT, old_int) + for path in leaked: + if path in snapshot: + with open(path, "wb") as f: + f.write(snapshot[path]) + elif os.path.exists(path): + os.remove(path) + + +class DispatchReviewMergeTest(Main3Base): + def test_dispatch_budget_gate(self): + today = time.strftime("%Y-%m-%d", time.localtime()) + with open(self.config_path, "w") as f: + json.dump({"max_daily_turns": 5}, f) + self._save_state({"tasks": {}, "automations": [], + "usage": {"date": today, "turns": 5}}) + repo = self._make_repo() + code, out = self._run(FakeServer(), "dispatch", "--name", "x", + "--task", "y", "--repo", repo) + self.assertEqual(code, 0) + self.assertIn("预算已用完", out) + + def test_review_with_open_comments(self): + repo = self._make_repo() + with open(os.path.join(repo, "f.txt"), "a") as f: + f.write("more\n") + self._save_state({"tasks": {"t1": {"name": "t1", + "thread_id": "th-1", + "status": "completed", + "cwd": repo, + "review_comments": [ + {"file": "a.py", + "line": 3, + "comment": "fix", + "status": "open"}]}}, + "automations": []}) + code, out = self._run(FakeServer(), "review", "--task", "t1") + self.assertEqual(code, 0) + self.assertIn("open review comments", out) + self.assertIn("a.py:3: fix", out) + + def test_merge_non_git_and_failed(self): + self._save_state({"tasks": {"t1": {"name": "t1", + "thread_id": "th-1", + "status": "completed"}}, + "automations": []}) + code, out = self._run(FakeServer(), "merge", "--task", "t1") + self.assertEqual(code, 0) + self.assertIn("非 git worktree", out) + self._save_state({"tasks": {"t1": {"name": "t1", + "thread_id": "th-1", + "status": "completed", + "repo": "/tmp", + "branch": "b", + "cwd": "/tmp"}}, + "automations": []}) + code, out = self._run( + FakeServer(), "merge", "--task", "t1", + merge_task=lambda *a, **k: (False, "verify failed")) + self.assertEqual(code, 0) + self.assertIn("verify failed", out) + self.assertEqual( + self._load_state()["tasks"]["t1"]["status"], + "merge-failed") + + def test_fix_review_completed_and_failed(self): + base = {"name": "t1", "thread_id": "th-1", + "status": "completed", "cwd": "/tmp", + "review_comments": [{"file": "a.py", "line": 1, + "comment": "fix", + "status": "open"}]} + self._save_state({"tasks": {"t1": dict(base)}, + "automations": []}) + code, out = self._run( + FakeServer(), "fix-review", "--task", "t1", + run_task_turn=lambda *a, **k: {"status": "completed"}) + self.assertEqual(code, 0) + self.assertIn("fix-review completed", out) + state = self._load_state() + self.assertEqual( + state["tasks"]["t1"]["review_comments"][0]["status"], + "resolved") + self._save_state({"tasks": {"t1": dict(base)}, + "automations": []}) + code, out = self._run( + FakeServer(), "fix-review", "--task", "t1", + run_task_turn=lambda *a, **k: {"status": "failed", + "text": "bad"}) + self.assertEqual(code, 0) + self.assertIn("fix-review failed", out) + + def test_discard_with_repo_branch(self): + repo = self._make_repo() + wt = os.path.join(self.tmp.name, "wt1") + self._git("worktree", "add", wt, "-b", "host/t1", cwd=repo) + self._save_state({"tasks": {"t1": {"name": "t1", + "thread_id": "th-1", + "status": "completed", + "repo": repo, + "branch": "host/t1", + "cwd": wt}}, + "automations": []}) + code, out = self._run(FakeServer(), "discard", "--task", "t1") + self.assertEqual(code, 0) + self.assertIn("discarded t1", out) + self.assertFalse(os.path.exists(wt)) + + +class ThreadErrorBranchesTest(Main3Base): + COMMANDS = [ + (["thread-list"], "thread/list"), + (["thread-read", "--thread-id", "th-1"], "thread/read"), + (["thread-rename", "--thread-id", "th-1", "--name", "n"], + "thread/name/set"), + (["thread-archive", "--thread-id", "th-1"], "thread/archive"), + (["thread-unarchive", "--thread-id", "th-1"], + "thread/unarchive"), + (["thread-compact", "--thread-id", "th-1"], + "thread/compact/start"), + (["skills-list"], "skills/list"), + (["apps-list"], "app/list"), + (["mcp-servers-list"], "mcpServerStatus/list"), + (["mcp-servers-reload"], "config/mcpServer/reload"), + (["mcp-server-oauth-login", "--server-name", "s1"], + "mcpServer/oauth/login"), + ] + + def test_thread_commands_systemexit_reraise(self): + for argv, method in self.COMMANDS: + with self.subTest(argv=argv): + fake = FakeServer() + fake.raise_systemexit = {method} + code, out = self._run(fake, *argv) + self.assertEqual(code, 3) + + def test_thread_commands_runtime_error_json(self): + for argv, method in self.COMMANDS: + with self.subTest(argv=argv): + fake = FakeServer() + fake.raise_runtime = {method} + code, out = self._run(fake, *argv) + self.assertEqual(code, 1) + self.assertIn("boom-" + method, out) + + +class CodeReviewInterruptDelegateGoalTest(Main3Base): + def _task(self, **kw): + d = {"name": "t1", "thread_id": "th-1", + "status": "completed", "cwd": "/tmp", + "verify_command": ""} + d.update(kw) + return d + + def test_code_review_targets(self): + self._save_state({"tasks": {"t1": self._task()}, + "automations": []}) + fake = FakeServer() + code, out = self._run(fake, "code-review", "--task", "t1", + "--commit", "abc123") + self.assertEqual(code, 0) + self.assertEqual(fake.calls[-2][2]["type"], "commit") + fake = FakeServer() + code, out = self._run(fake, "code-review", "--task", "t1", + "--base-branch", "main") + self.assertEqual(code, 0) + self.assertEqual(fake.calls[-2][2]["type"], "baseBranch") + fake = FakeServer() + code, out = self._run(fake, "code-review", "--task", "t1") + self.assertEqual(code, 0) + self.assertEqual(fake.calls[-2][2]["type"], + "uncommittedChanges") + + def test_code_review_failure(self): + self._save_state({"tasks": {"t1": self._task()}, + "automations": []}) + fake = FakeServer() + + def _boom(thread_id, target): + raise RuntimeError("review 炸了") + fake.start_review = _boom + code, out = self._run(fake, "code-review", "--task", "t1") + self.assertEqual(code, 1) + self.assertIn("code-review failed", out) + + def test_interrupt_slow_path_waits_ten_seconds(self): + self._save_state({"tasks": {"it": { + "name": "it", "thread_id": "th-1", "status": "running", + "turns": [{"result": {"turn_id": "turn-1", + "status": "completed"}}]}}, + "automations": []}) + fake = FakeServer() + fake.interrupt_result = {"error": {"message": "no active turn"}} + started = time.time() + code, out = self._run(fake, "interrupt", "--task", "it") + elapsed = time.time() - started + self.assertEqual(code, 0) + self.assertGreaterEqual(elapsed, 9.5) + self.assertIn("属主 10 秒内未响应", out) + + def test_interrupt_write_failure(self): + self._save_state({"tasks": {"it": { + "name": "it", "thread_id": "th-1", "status": "running", + "turns": [{"result": {"turn_id": "turn-1", + "status": "completed"}}]}}, + "automations": []}) + req = host._interrupt_request_path(self.state_path, "it") + os.makedirs(req, exist_ok=True) + fake = FakeServer() + fake.interrupt_result = {"error": {"message": "no active turn"}} + code, out = self._run(fake, "interrupt", "--task", "it") + self.assertEqual(code, 0) + self.assertIn("写中断请求失败", out) + + def test_delegate_report_to_parent(self): + self._save_state({"tasks": {"pt": { + "name": "pt", "thread_id": "th-1", "status": "running", + "cwd": "/tmp"}}, + "automations": []}) + fake = FakeServer() + code, out = self._run( + fake, "delegate", "--parent-task", "pt", "--task", "sub", + "--name", "sub1", "--report-to-parent", + run_task_turn=lambda *a, **k: {"status": "completed", + "text": "sub done"}) + self.assertEqual(code, 0) + self.assertIn("reported to parent", out) + state = self._load_state() + self.assertEqual(state["tasks"]["sub1"]["status"], "running") + + def test_goal_branch_wiring(self): + repo = self._make_repo() + code, out = self._run( + FakeServer(), "goal", "--name", "g1", "--task", "达成目标", + "--repo", repo, + run_goal_loop=lambda *a, **k: None) + self.assertEqual(code, 0) + state = self._load_state() + self.assertIn("g1", state["tasks"]) + # 真实 dispatch 在 HERE/work 下建了 worktree + 分支, + # 本测试负责清掉,不留脏 worktree。 + wt = state["tasks"]["g1"]["cwd"] + branch = state["tasks"]["g1"]["branch"] + subprocess.run(["git", "-C", repo, "worktree", "remove", + "--force", wt], capture_output=True) + subprocess.run(["git", "-C", repo, "branch", "-D", branch], + capture_output=True) + + def test_goal_resume_guards_and_rounds(self): + # 已终局拒绝。 + self._save_state({"tasks": {"t1": self._task( + status="completed", goal="做周报")}, + "automations": []}) + code, out = self._run(FakeServer(), "goal-resume", + "--task", "t1") + self.assertIn("已终局", out) + # goal-running 但无 goal 记录拒绝。 + self._save_state({"tasks": {"t1": self._task( + status="goal-running")}, "automations": []}) + code, out = self._run(FakeServer(), "goal-resume", + "--task", "t1") + self.assertIn("没有 goal 记录", out) + # 轮次来源三路:CLI 覆盖 / 任务原值 / 缺省 3。 + seen = {} + + def _loop(srv, state, state_path, task, goal, verify, + max_rounds, *a, **k): + seen["max_rounds"] = max_rounds + seen["start_round"] = k.get("start_round") + self._save_state({"tasks": {"t1": self._task( + status="goal-running", goal="做周报", + goal_max_rounds=7, goal_rounds_done=2)}, + "automations": []}) + code, out = self._run(FakeServer(), "goal-resume", + "--task", "t1", run_goal_loop=_loop) + self.assertEqual(seen["max_rounds"], 7) + self.assertEqual(seen["start_round"], 3) + code, out = self._run(FakeServer(), "goal-resume", + "--task", "t1", "--max-rounds", "2", + run_goal_loop=_loop) + self.assertEqual(seen["max_rounds"], 2) + self._save_state({"tasks": {"t1": self._task( + status="goal-running", goal="做周报")}, + "automations": []}) + code, out = self._run(FakeServer(), "goal-resume", + "--task", "t1", run_goal_loop=_loop) + self.assertEqual(seen["max_rounds"], 3) + # 旧验收形态 0 轮次钳制到 100。 + self._save_state({"tasks": {"t1": self._task( + status="goal-running", goal="做周报", + verify_command="make test", goal_max_rounds=0)}, + "automations": []}) + code, out = self._run(FakeServer(), "goal-resume", + "--task", "t1", run_goal_loop=_loop) + self.assertEqual(seen["max_rounds"], 100) + + +class ParallelAutomationsTest(Main3Base): + def test_parallel_simple_and_blocked(self): + jobs_path = os.path.join(self.tmp.name, "jobs.json") + with open(jobs_path, "w") as f: + json.dump([ + {"name": "a", "task": "做 A"}, + {"name": "b", "task": "做 B", + "depends_on": ["a"]}, + ], f) + self._save_state({"tasks": {}, "automations": []}) + + def _run_turn(srv, state, task, prompt, model, effort, + sandbox, **kw): + if task["name"] == "a": + task["status"] = "failed" + return {"status": "failed", "text": "a bad"} + task["status"] = "completed" + return {"status": "completed", "text": "ok"} + code, out = self._run( + FakeServer(), "parallel", "--jobs", jobs_path, + run_task_turn_with_retry=_run_turn) + self.assertEqual(code, 0) + self.assertIn("skipped (依赖失败)", out) + state = self._load_state() + self.assertEqual(state["tasks"]["b"]["status"], "skipped") + + def test_parallel_unpushable_remaining(self): + # 依赖既未完成也未失败/跳过(如 interrupted):既不进 + # blocked 也不进 ready,循环以“无法推进”收尾。 + jobs_path = os.path.join(self.tmp.name, "jobs.json") + with open(jobs_path, "w") as f: + json.dump([ + {"name": "a", "task": "做 A"}, + {"name": "b", "task": "做 B", + "depends_on": ["a"]}, + ], f) + self._save_state({"tasks": {}, "automations": []}) + + def _run_turn(srv, state, task, prompt, model, effort, + sandbox, **kw): + task["status"] = "interrupted" + return {"status": "interrupted", "text": "cut"} + code, out = self._run( + FakeServer(), "parallel", "--jobs", jobs_path, + run_task_turn_with_retry=_run_turn) + self.assertEqual(code, 0) + self.assertIn("无法推进", out) + state = self._load_state() + self.assertNotIn("b", state["tasks"]) + + def test_parallel_with_repo_dependency(self): + repo = self._make_repo() + jobs_path = os.path.join(self.tmp.name, "jobs.json") + with open(jobs_path, "w") as f: + json.dump([ + {"name": "a", "task": "做 A", "repo": repo}, + {"name": "b", "task": "做 B", "repo": repo, + "depends_on": ["a"]}, + ], f) + self._save_state({"tasks": {}, "automations": []}) + + def _run_turn(srv, state, task, prompt, model, effort, + sandbox, **kw): + task["status"] = "completed" + return {"status": "completed", "text": "ok"} + code, out = self._run( + FakeServer(), "parallel", "--jobs", jobs_path, + run_task_turn_with_retry=_run_turn) + self.assertEqual(code, 0) + state = self._load_state() + self.assertEqual(state["tasks"]["a"]["status"], "completed") + self.assertEqual(state["tasks"]["b"]["status"], "completed") + # 依赖分支派发后 b 的 worktree 与 a 的分支有血缘: + # b 的 branch 应存在且不同于 a。 + self.assertNotEqual(state["tasks"]["a"]["branch"], + state["tasks"]["b"]["branch"]) + # 清理 parallel 建的 worktree(HERE/work 下)。 + for name in ("a", "b"): + wt = state["tasks"][name]["cwd"] + subprocess.run(["git", "-C", repo, "worktree", "remove", + "--force", wt], capture_output=True) + subprocess.run(["git", "-C", repo, "branch", "-D", + state["tasks"][name]["branch"]], + capture_output=True) + + def test_automations_add_list_and_run_due(self): + self._save_state({"tasks": {}, "automations": []}) + code, out = self._run(FakeServer(), "automations", "--add", + "--name", "daily", "--task", "巡检") + self.assertEqual(code, 0) + self.assertIn("automation added", out) + code, out = self._run(FakeServer(), "automations") + self.assertIn("daily", out) + # run-due:一条未到期跳过、一条到期跑(dispatch/turn 替身)。 + state = self._load_state() + state["automations"] = [ + {"name": "soon", "task": "x", "repo": None, + "interval": 86400, "last_run": time.time()}, + {"name": "due", "task": "巡检", "repo": None, + "interval": 60, "last_run": 0}, + ] + self._save_state(state) + + def _dispatch(srv, state, name, prompt, repo, model, effort, + sandbox, approval_policy, **kw): + task = {"name": name, "thread_id": "th-1", "cwd": "/tmp", + "status": "running", "prompt": prompt} + state["tasks"][name] = task + return task + code, out = self._run( + FakeServer(), "run-due", dispatch=_dispatch, + run_task_turn=lambda *a, **k: {"status": "failed", + "text": "bad"}) + self.assertEqual(code, 0) + self.assertIn("automation due ->", out) + state = self._load_state() + self.assertGreater(state["automations"][1]["last_run"], 0) + + +if __name__ == "__main__": + unittest.main()