686 lines
28 KiB
Python
686 lines
28 KiB
Python
"""覆盖率 85% 第 4 步第 6 批：host.py 纯函数与小方法长尾。
|
|
|
|
只动测试。全部是直接调用生产函数/方法：用量汇总、链路
|
|
摘要、等待轮询、状态锁与恢复、git diff/合并/验测、PR 解析
|
|
与正文、通知、doctor、Server 的 trace/close（不拉起真实
|
|
app-server 进程）。
|
|
"""
|
|
|
|
import json
|
|
import os
|
|
import queue
|
|
import subprocess
|
|
import tempfile
|
|
import time
|
|
import unittest
|
|
from unittest import mock
|
|
|
|
import host
|
|
|
|
|
|
def _git(repo, *args, **kw):
|
|
return subprocess.run(["git", "-C", repo] + list(args),
|
|
capture_output=True, text=True, **kw)
|
|
|
|
|
|
def _make_repo(base):
|
|
repo = os.path.join(base, "repo")
|
|
os.makedirs(repo)
|
|
_git(repo, "init", "-q", "-b", "main")
|
|
_git(repo, "config", "user.email", "t@t.t")
|
|
_git(repo, "config", "user.name", "t")
|
|
with open(os.path.join(repo, "a.py"), "w") as f:
|
|
f.write("x = 1\n")
|
|
_git(repo, "add", ".")
|
|
_git(repo, "commit", "-q", "-m", "init")
|
|
return repo
|
|
|
|
|
|
class UsageAndEventsTest(unittest.TestCase):
|
|
def test_sum_turn_usage(self):
|
|
events = [
|
|
{"method": "other", "params": {}},
|
|
{"method": "thread/tokenUsage/updated",
|
|
"params": {"threadId": "th-1", "turnId": "tu-1",
|
|
"tokenUsage": {"last": {
|
|
"inputTokens": 10, "outputTokens": 5,
|
|
"cachedInputTokens": 2,
|
|
"totalTokens": 15}}}},
|
|
{"method": "thread/tokenUsage/updated",
|
|
"params": {"threadId": "th-1", "turnId": "tu-2",
|
|
"tokenUsage": {"last": {
|
|
"inputTokens": 7, "outputTokens": 3}}}},
|
|
{"method": "thread/tokenUsage/updated",
|
|
"params": {"threadId": "th-2", "turnId": "tu-1",
|
|
"tokenUsage": {"last": {"inputTokens": 99}}}},
|
|
]
|
|
total = host._sum_turn_usage(events, "th-1", "tu-1")
|
|
self.assertEqual(total["input_tokens"], 10)
|
|
self.assertEqual(total["output_tokens"], 5)
|
|
self.assertEqual(total["cached_input_tokens"], 2)
|
|
self.assertEqual(total["completions"], 1)
|
|
total = host._sum_turn_usage(events, "th-1", None)
|
|
self.assertEqual(total["input_tokens"], 17)
|
|
self.assertEqual(total["completions"], 2)
|
|
self.assertIsNone(host._sum_turn_usage(events, "th-9", None))
|
|
self.assertIsNone(host._sum_turn_usage([], "th-1", None))
|
|
self.assertIsNone(host._sum_turn_usage(None, "th-1", None))
|
|
|
|
def test_partial_text_from_events(self):
|
|
events = [
|
|
{"params": {"item": {"type": "agentMessage",
|
|
"text": "first"}}},
|
|
{"params": {"item": {"type": "agentMessage",
|
|
"text": "first"}}},
|
|
{"params": {"item": {"type": "delta", "text": "skip"}}},
|
|
{"params": {"item": {"type": "agentMessage",
|
|
"text": "second"}}},
|
|
{"params": {}},
|
|
]
|
|
self.assertEqual(host._partial_text_from_events(events),
|
|
"first\nsecond")
|
|
self.assertEqual(host._partial_text_from_events([]), "")
|
|
self.assertEqual(host._partial_text_from_events(None), "")
|
|
|
|
def test_add_usage(self):
|
|
self.assertIsNone(host._add_usage(None, None))
|
|
self.assertEqual(host._add_usage(None, {"a": 1}), {"a": 1})
|
|
self.assertEqual(host._add_usage({"a": 1}, None), {"a": 1})
|
|
out = host._add_usage({"a": 1, "s": "x"}, {"a": 2, "b": 3})
|
|
self.assertEqual(out, {"a": 3, "s": "x", "b": 3})
|
|
|
|
def test_interrupt_request_path_sanitizes(self):
|
|
tmp = tempfile.TemporaryDirectory()
|
|
try:
|
|
p = host._interrupt_request_path(
|
|
os.path.join(tmp.name, "host-state.json"), "a/b c")
|
|
self.assertEqual(os.path.basename(p), "a-b-c.json")
|
|
self.assertEqual(os.path.basename(os.path.dirname(p)),
|
|
"interrupt-requests")
|
|
finally:
|
|
tmp.cleanup()
|
|
|
|
|
|
class WaitTurnTest(unittest.TestCase):
|
|
def test_completed_immediately(self):
|
|
q = queue.Queue()
|
|
q.put({"done": True})
|
|
status, done = host._wait_turn(q, timeout=5)
|
|
self.assertEqual((status, done), ("completed", {"done": True}))
|
|
|
|
def test_timeout_and_progress(self):
|
|
q = queue.Queue()
|
|
seen = []
|
|
status, done = host._wait_turn(
|
|
q, timeout=0.35, progress_interval=0.1,
|
|
on_progress=lambda e: seen.append(round(e, 1)))
|
|
self.assertEqual(status, "timeout")
|
|
self.assertIsNone(done)
|
|
self.assertGreaterEqual(len(seen), 2)
|
|
|
|
def test_interrupt_requested_and_swallow(self):
|
|
q = queue.Queue()
|
|
calls = {"n": 0}
|
|
|
|
def _check():
|
|
calls["n"] += 1
|
|
if calls["n"] == 1:
|
|
raise RuntimeError("check 炸了")
|
|
return True
|
|
status, _ = host._wait_turn(q, timeout=5,
|
|
interrupt_check=_check)
|
|
self.assertEqual(status, "interrupted-requested")
|
|
# on_progress 抛错也被吞掉，不影响 timeout 判定。
|
|
status, _ = host._wait_turn(
|
|
q, timeout=0.25, progress_interval=0.1,
|
|
on_progress=lambda e: (_ for _ in ()).throw(
|
|
RuntimeError("progress 炸了")))
|
|
self.assertEqual(status, "timeout")
|
|
|
|
def test_unlimited_waits_for_done(self):
|
|
q = queue.Queue()
|
|
|
|
def _put():
|
|
time.sleep(0.2)
|
|
q.put("x")
|
|
import threading
|
|
threading.Thread(target=_put, daemon=True).start()
|
|
status, done = host._wait_turn(q, timeout=0)
|
|
self.assertEqual((status, done), ("completed", "x"))
|
|
|
|
|
|
class SummarizeTraceTest(unittest.TestCase):
|
|
def _params(self, item, **kw):
|
|
p = {"item": item}
|
|
p.update(kw)
|
|
return p
|
|
|
|
def test_all_item_types(self):
|
|
out = host._summarize_trace_item(
|
|
"m", self._params({"type": "commandExecution", "id": "c1",
|
|
"command": "ls", "cwd": "/tmp",
|
|
"exitCode": 0, "durationMs": 5},
|
|
completedAtMs=123, threadId="th-1"))
|
|
self.assertEqual(out[0]["kind"], "exec")
|
|
self.assertEqual(out[0]["ts"], 123)
|
|
out = host._summarize_trace_item(
|
|
"m", self._params({"type": "fileChange", "id": "f1",
|
|
"status": "done",
|
|
"changes": [{"path": "a.py", "kind": "edit"},
|
|
"bad",
|
|
{"path": "b.py"}]},
|
|
thread={"id": "th-9"}))
|
|
self.assertEqual([e["id"] for e in out],
|
|
["f1:a.py", "f1:b.py"])
|
|
self.assertEqual(out[0]["thread"], "th-9")
|
|
long_text = "z" * 400
|
|
out = host._summarize_trace_item(
|
|
"m", self._params({"type": "agentMessage", "id": "m1",
|
|
"text": long_text}))
|
|
self.assertEqual(len(out[0]["text"]), 300)
|
|
self.assertTrue(out[0]["truncated"])
|
|
out = host._summarize_trace_item(
|
|
"m", self._params({"type": "reasoning", "id": "r1",
|
|
"summary": ["a", 5, "b"]}))
|
|
self.assertEqual(out[0]["summary"], "a b")
|
|
out = host._summarize_trace_item(
|
|
"m", self._params({"type": "mcpToolCall", "id": "t1",
|
|
"server": "s", "tool": "t",
|
|
"status": "ok"}))
|
|
self.assertEqual(out[0]["kind"], "tool")
|
|
out = host._summarize_trace_item(
|
|
"m", self._params({"type": "functionCallOutput", "id": "o1",
|
|
"name": "f"}))
|
|
self.assertEqual(out[0]["kind"], "call")
|
|
out = host._summarize_trace_item(
|
|
"m", self._params({"type": "plan", "id": "p1",
|
|
"text": "plan text"}))
|
|
self.assertEqual(out[0]["kind"], "plan")
|
|
|
|
def test_empty_and_unknown(self):
|
|
self.assertEqual(host._summarize_trace_item("m", {}), [])
|
|
self.assertEqual(host._summarize_trace_item(
|
|
"m", {"item": "not-a-dict"}), [])
|
|
self.assertEqual(host._summarize_trace_item(
|
|
"m", {"item": {"type": "userMessage", "text": "hi"}}), [])
|
|
self.assertEqual(host._summarize_trace_item(
|
|
"m", {"item": {"type": "agentMessage", "text": ""}}), [])
|
|
self.assertEqual(host._summarize_trace_item(
|
|
"m", {"item": {"type": "reasoning", "summary": []}}), [])
|
|
self.assertEqual(host._summarize_trace_item(
|
|
"m", {"item": {"type": "plan", "text": ""}}), [])
|
|
|
|
def test_trace_path_and_marker(self):
|
|
tmp = tempfile.TemporaryDirectory()
|
|
try:
|
|
state_path = os.path.join(tmp.name, "host-state.json")
|
|
path = host._trace_file_path(state_path, "a/b")
|
|
self.assertEqual(os.path.basename(path), "a-b.jsonl")
|
|
host._append_trace_marker(state_path, "a/b",
|
|
{"kind": "mark"})
|
|
with open(path) as f:
|
|
row = json.loads(f.readline())
|
|
self.assertEqual(row["kind"], "mark")
|
|
finally:
|
|
tmp.cleanup()
|
|
|
|
|
|
class StateLockAndReadTest(unittest.TestCase):
|
|
def setUp(self):
|
|
self.tmp = tempfile.TemporaryDirectory()
|
|
self.path = os.path.join(self.tmp.name, "host-state.json")
|
|
|
|
def tearDown(self):
|
|
self.tmp.cleanup()
|
|
|
|
def test_tmp_path_unique(self):
|
|
a = host._state_tmp_path(self.path)
|
|
b = host._state_tmp_path(self.path)
|
|
self.assertNotEqual(a, b)
|
|
self.assertTrue(a.startswith(self.path + ".tmp."))
|
|
|
|
def test_lock_exclusive_and_shared(self):
|
|
with host._state_lock(self.path):
|
|
self.assertTrue(os.path.exists(self.path + ".lockdir"))
|
|
self.assertFalse(os.path.exists(self.path + ".lockdir"))
|
|
with host._state_lock(self.path, exclusive=False):
|
|
self.assertFalse(os.path.exists(self.path + ".lockdir"))
|
|
|
|
def test_lock_stale_removed_and_timeout(self):
|
|
lockdir = self.path + ".lockdir"
|
|
os.makedirs(lockdir)
|
|
old = time.time() - 10000
|
|
os.utime(lockdir, (old, old))
|
|
with host._state_lock(self.path):
|
|
pass
|
|
self.assertFalse(os.path.exists(lockdir))
|
|
os.makedirs(lockdir)
|
|
with mock.patch.object(host, "_STATE_LOCK_TIMEOUT", 0.01):
|
|
with self.assertRaises(RuntimeError):
|
|
with host._state_lock(self.path):
|
|
pass
|
|
os.rmdir(lockdir)
|
|
|
|
def test_read_state_file_variants(self):
|
|
self.assertEqual(host._read_state_file(self.path),
|
|
{"tasks": {}, "automations": []})
|
|
with open(self.path, "w") as f:
|
|
f.write("{bad")
|
|
self.assertEqual(host._read_state_file(self.path),
|
|
{"tasks": {}, "automations": []})
|
|
with self.assertRaises(RuntimeError):
|
|
host._read_state_file_for_save(self.path)
|
|
with open(self.path, "w") as f:
|
|
json.dump(["not", "a", "dict"], f)
|
|
with self.assertRaises(RuntimeError):
|
|
host._read_state_file_for_save(self.path)
|
|
with open(self.path, "w") as f:
|
|
json.dump({"tasks": [], "automations": []}, f)
|
|
with self.assertRaises(RuntimeError):
|
|
host._read_state_file_for_save(self.path)
|
|
with open(self.path, "w") as f:
|
|
json.dump({"tasks": {}, "automations": {}}, f)
|
|
with self.assertRaises(RuntimeError):
|
|
host._read_state_file_for_save(self.path)
|
|
with open(self.path, "w") as f:
|
|
json.dump({"tasks": {}, "automations": []}, f)
|
|
self.assertEqual(host._read_state_file_for_save(self.path),
|
|
{"tasks": {}, "automations": []})
|
|
|
|
def test_load_state_strict(self):
|
|
data, err = host.load_state_strict(self.path)
|
|
self.assertIsNone(err)
|
|
with open(self.path, "w") as f:
|
|
f.write("{bad")
|
|
data, err = host.load_state_strict(self.path)
|
|
self.assertIsNone(data)
|
|
self.assertIsNotNone(err)
|
|
|
|
|
|
class RestoreAndRecoverTest(unittest.TestCase):
|
|
def setUp(self):
|
|
self.tmp = tempfile.TemporaryDirectory()
|
|
self.path = os.path.join(self.tmp.name, "host-state.json")
|
|
|
|
def tearDown(self):
|
|
self.tmp.cleanup()
|
|
|
|
def _write(self, path, obj):
|
|
with open(path, "w") as f:
|
|
json.dump(obj, f)
|
|
|
|
def test_restore_good_is_noop(self):
|
|
self._write(self.path, {"tasks": {}, "automations": [],
|
|
"version": 5})
|
|
ok, msg = host.restore_state(self.path)
|
|
self.assertTrue(ok)
|
|
self.assertIn("完好", msg)
|
|
|
|
def test_restore_from_bak1_when_bak_bad(self):
|
|
with open(self.path, "w") as f:
|
|
f.write("{bad")
|
|
with open(self.path + ".bak", "w") as f:
|
|
f.write("{also bad")
|
|
self._write(self.path + ".bak.1",
|
|
{"tasks": {}, "automations": [], "version": 7})
|
|
ok, msg = host.restore_state(self.path)
|
|
self.assertTrue(ok)
|
|
self.assertIn(".bak.1", msg)
|
|
self.assertEqual(host.load_state(self.path)["version"], 7)
|
|
leftovers = [n for n in os.listdir(self.tmp.name)
|
|
if n.startswith("host-state.json.corrupt.")]
|
|
self.assertEqual(len(leftovers), 1)
|
|
|
|
def test_restore_no_backup(self):
|
|
with open(self.path, "w") as f:
|
|
f.write("{bad")
|
|
ok, msg = host.restore_state(self.path)
|
|
self.assertFalse(ok)
|
|
self.assertIn("没有可用的备份", msg)
|
|
|
|
def test_recover_marks_only_old_running(self):
|
|
now = time.time()
|
|
state = {"tasks": {
|
|
"old": {"name": "old", "status": "running",
|
|
"created_at": now - 5000},
|
|
"fresh": {"name": "fresh", "status": "running",
|
|
"last_turn_at": now - 10},
|
|
"done": {"name": "done", "status": "completed",
|
|
"created_at": now - 5000}},
|
|
"automations": []}
|
|
got = host.recover_interrupted(state, self.path,
|
|
older_than=3600)
|
|
self.assertEqual(got, ["old"])
|
|
self.assertEqual(state["tasks"]["old"]["status"],
|
|
"interrupted")
|
|
self.assertEqual(state["tasks"]["fresh"]["status"],
|
|
"running")
|
|
self.assertEqual(state["tasks"]["done"]["status"],
|
|
"completed")
|
|
|
|
|
|
class GitHelpersTest(unittest.TestCase):
|
|
def setUp(self):
|
|
self.tmp = tempfile.TemporaryDirectory()
|
|
self.repo = _make_repo(self.tmp.name)
|
|
|
|
def tearDown(self):
|
|
self.tmp.cleanup()
|
|
|
|
def test_task_diff_tracked_and_untracked(self):
|
|
with open(os.path.join(self.repo, "a.py"), "a") as f:
|
|
f.write("y = 2\n")
|
|
with open(os.path.join(self.repo, "new.txt"), "w") as f:
|
|
f.write("hello\n")
|
|
os.makedirs(os.path.join(self.repo, "__pycache__"),
|
|
exist_ok=True)
|
|
with open(os.path.join(self.repo, "__pycache__",
|
|
"x.pyc"), "wb") as f:
|
|
f.write(b"\x00\x01")
|
|
out = host.task_diff(self.repo)
|
|
self.assertIn("y = 2", out["diff"])
|
|
self.assertIn("new.txt", out["untracked"])
|
|
self.assertEqual(out["untracked_contents"]["new.txt"],
|
|
"hello\n")
|
|
self.assertNotIn("__pycache__/x.pyc",
|
|
out["untracked_contents"])
|
|
|
|
def test_task_diff_non_git_dir(self):
|
|
plain = os.path.join(self.tmp.name, "plain")
|
|
os.makedirs(plain)
|
|
with open(os.path.join(plain, "f.txt"), "w") as f:
|
|
f.write("x")
|
|
out = host.task_diff(plain)
|
|
self.assertEqual(out["diff"], "")
|
|
self.assertIn("f.txt", out["untracked"])
|
|
|
|
def test_topo_sort_and_errors(self):
|
|
jobs = [{"name": "b", "depends_on": ["a"], "priority": 1},
|
|
{"name": "a", "priority": 5},
|
|
{"name": "c", "depends_on": ["a"], "priority": 9}]
|
|
ordered = [j["name"] for j in host.topo_sort_jobs(jobs)]
|
|
self.assertEqual(ordered, ["a", "c", "b"])
|
|
with self.assertRaises(ValueError):
|
|
host.topo_sort_jobs([{"name": "a",
|
|
"depends_on": ["ghost"]}])
|
|
with self.assertRaises(ValueError):
|
|
host.topo_sort_jobs([{"name": "a",
|
|
"depends_on": ["a"]}])
|
|
with self.assertRaises(ValueError):
|
|
host.topo_sort_jobs([{"name": "a",
|
|
"depends_on": ["b"]},
|
|
{"name": "b",
|
|
"depends_on": ["a"]}])
|
|
|
|
def test_detect_verify_and_run_verify(self):
|
|
self.assertIsNone(host.detect_verify_command(self.repo))
|
|
os.makedirs(os.path.join(self.repo, "tests"))
|
|
self.assertIn("unittest",
|
|
host.detect_verify_command(self.repo))
|
|
self.assertIsNone(host.detect_verify_command(None))
|
|
self.assertEqual(host.run_verify("", self.repo), (0, ""))
|
|
rc, out = host.run_verify("echo hi", self.repo)
|
|
self.assertEqual((rc, out), (0, "hi"))
|
|
rc, _ = host.run_verify("false", self.repo)
|
|
self.assertNotEqual(rc, 0)
|
|
|
|
def test_merge_task_success_and_verify_fail(self):
|
|
wt = os.path.join(self.tmp.name, "wt1")
|
|
_git(self.repo, "worktree", "add", wt, "-b", "host/t1")
|
|
with open(os.path.join(wt, "a.py"), "a") as f:
|
|
f.write("z = 3\n")
|
|
ok, msg = host.merge_task(self.repo, wt, "host/t1", "t1",
|
|
verify_command="true")
|
|
self.assertTrue(ok, msg)
|
|
self.assertFalse(os.path.exists(wt))
|
|
self.assertNotIn("host/t1",
|
|
_git(self.repo, "branch").stdout)
|
|
with open(os.path.join(self.repo, "a.py")) as f:
|
|
self.assertIn("z = 3", f.read())
|
|
# 验测失败：回滚主仓、分支与 worktree 保留。
|
|
wt2 = os.path.join(self.tmp.name, "wt2")
|
|
_git(self.repo, "worktree", "add", wt2, "-b", "host/t2")
|
|
with open(os.path.join(wt2, "a.py"), "a") as f:
|
|
f.write("w = 4\n")
|
|
ok, msg = host.merge_task(self.repo, wt2, "host/t2", "t2",
|
|
verify_command="false")
|
|
self.assertFalse(ok)
|
|
self.assertIn("验测失败", msg)
|
|
with open(os.path.join(self.repo, "a.py")) as f:
|
|
self.assertNotIn("w = 4", f.read())
|
|
self.assertIn("host/t2", _git(self.repo, "branch").stdout)
|
|
_git(self.repo, "worktree", "remove", "--force", wt2)
|
|
_git(self.repo, "branch", "-D", "host/t2")
|
|
|
|
|
|
class BudgetAndNotifyTest(unittest.TestCase):
|
|
def setUp(self):
|
|
self.tmp = tempfile.TemporaryDirectory()
|
|
self.path = os.path.join(self.tmp.name, "host-state.json")
|
|
|
|
def tearDown(self):
|
|
self.tmp.cleanup()
|
|
|
|
def test_record_usage_and_budget(self):
|
|
state = {}
|
|
u = host.record_turn_usage(state)
|
|
self.assertEqual(u["turns"], 1)
|
|
u = host.record_turn_usage(state, failed=True)
|
|
self.assertEqual((u["turns"], u["failed_turns"]), (2, 1))
|
|
# 跨天重置。
|
|
state["usage"] = {"date": "2000-01-01", "turns": 9,
|
|
"failed_turns": 9}
|
|
u = host.record_turn_usage(state)
|
|
self.assertEqual((u["turns"], u["failed_turns"]), (1, 0))
|
|
today = time.strftime("%Y-%m-%d", time.localtime())
|
|
self.assertFalse(host.budget_exceeded(
|
|
{"usage": {"date": today, "turns": 1}}, {}))
|
|
self.assertFalse(host.budget_exceeded(
|
|
{"usage": {"date": today, "turns": 99}},
|
|
{"max_daily_turns": "bogus"}))
|
|
self.assertFalse(host.budget_exceeded(
|
|
{"usage": {"date": "2000-01-01", "turns": 99}},
|
|
{"max_daily_turns": 5}))
|
|
self.assertTrue(host.budget_exceeded(
|
|
{"usage": {"date": today, "turns": 5}},
|
|
{"max_daily_turns": 5}))
|
|
|
|
def test_notify_failure_log_only_and_command(self):
|
|
msg = host.notify_failure(self.path, {}, "t1",
|
|
{"status": "failed", "text": "boom"})
|
|
self.assertIn("FAILED t1", msg)
|
|
log_path = os.path.join(self.tmp.name, "host-run.log")
|
|
self.assertTrue(os.path.exists(log_path))
|
|
out_file = os.path.join(self.tmp.name, "notified.txt")
|
|
cfg = {"notify_command":
|
|
"echo {task} {status} > " + out_file}
|
|
host.notify_failure(self.path, cfg, "t1",
|
|
{"status": "failed", "text": "boom"})
|
|
with open(out_file) as f:
|
|
self.assertIn("t1", f.read())
|
|
# 坏模板：记 template error，不跑命令。
|
|
before = open(log_path).read()
|
|
host.notify_failure(self.path, {"notify_command": "echo {nope}"},
|
|
"t1", {"status": "failed", "text": "x"})
|
|
after = open(log_path).read()
|
|
self.assertIn("template error", after[len(before):])
|
|
|
|
def test_build_fix_prompt_and_pr_body(self):
|
|
task = {"name": "t1", "prompt": "do it",
|
|
"review_comments": [
|
|
{"file": "a.py", "line": 3, "comment": "fix",
|
|
"status": "open"},
|
|
{"file": "b.py", "line": 1, "comment": "old",
|
|
"status": "resolved"}]}
|
|
prompt = host.build_fix_prompt(task)
|
|
self.assertIn("a.py:3: fix", prompt)
|
|
self.assertNotIn("old", prompt)
|
|
body = host.build_pr_body(task)
|
|
self.assertIn("host 任务：t1", body)
|
|
self.assertIn("a.py:3: fix", body)
|
|
body = host.build_pr_body({"name": "t2"})
|
|
self.assertIn("（无提示）", body)
|
|
self.assertIn("无未解决评论", body)
|
|
|
|
def test_check_state(self):
|
|
state = {"version": 5, "tasks": {
|
|
"bad": {"status": "failed", "cwd": "/nonexistent-x"},
|
|
"ok": {"status": "completed"},
|
|
"gone": {"status": "running", "cwd": "/nonexistent-x",
|
|
"review_comments": [
|
|
{"status": "open", "file": "a.py",
|
|
"line": 1, "comment": "c"}]}},
|
|
"automations": []}
|
|
out = host.check_state(self.path, state)
|
|
self.assertEqual(out["version"], 5)
|
|
self.assertFalse(out["backup_exists"])
|
|
self.assertEqual(out["failed_tasks"], ["bad"])
|
|
self.assertEqual(sorted(out["missing_worktrees"]),
|
|
["bad", "gone"])
|
|
self.assertEqual(out["open_review_comments"], 1)
|
|
|
|
|
|
class ParsePrChecksTest(unittest.TestCase):
|
|
def test_tab_and_human_formats(self):
|
|
checks, all_pass = host.parse_pr_checks(
|
|
"name\tstate\nbuild\tpass\nlint\tfail\n")
|
|
self.assertEqual(checks, [{"name": "build", "state": "pass"},
|
|
{"name": "lint", "state": "fail"}])
|
|
self.assertFalse(all_pass)
|
|
checks, all_pass = host.parse_pr_checks(
|
|
"✓ build pass\n✓ lint passing\n")
|
|
self.assertTrue(all_pass)
|
|
self.assertEqual([c["state"] for c in checks],
|
|
["pass", "pass"])
|
|
checks, all_pass = host.parse_pr_checks(
|
|
"no checks reported\n")
|
|
self.assertEqual((checks, all_pass), ([], False))
|
|
checks, all_pass = host.parse_pr_checks("")
|
|
self.assertEqual((checks, all_pass), ([], False))
|
|
checks, _ = host.parse_pr_checks("build queued\ntest skipped\n"
|
|
"job cancelled\nx neutral\n"
|
|
"y in_progress\nz failing\n")
|
|
self.assertEqual([c["state"] for c in checks],
|
|
["pending", "skipped", "cancelled",
|
|
"neutral", "pending", "fail"])
|
|
|
|
|
|
class CliHelpersTest(unittest.TestCase):
|
|
def test_load_config_and_effective_cfg(self):
|
|
tmp = tempfile.TemporaryDirectory()
|
|
try:
|
|
self.assertEqual(host._load_config_file(None), {})
|
|
self.assertEqual(host._load_config_file(
|
|
os.path.join(tmp.name, "nope.json")), {})
|
|
bad = os.path.join(tmp.name, "bad.json")
|
|
with open(bad, "w") as f:
|
|
f.write("{bad")
|
|
with self.assertRaises(RuntimeError):
|
|
host._load_config_file(bad)
|
|
cfg = {"max_daily_turns": 5, "other": 1}
|
|
self.assertIs(host._effective_cfg(cfg, False), cfg)
|
|
out = host._effective_cfg(cfg, True)
|
|
self.assertIsNone(out["max_daily_turns"])
|
|
self.assertEqual(out["other"], 1)
|
|
self.assertEqual(cfg["max_daily_turns"], 5)
|
|
finally:
|
|
tmp.cleanup()
|
|
|
|
def test_resolve_path_and_thread_result(self):
|
|
self.assertEqual(host._resolve_path("/base", ""), "")
|
|
self.assertEqual(host._resolve_path("/base", "/abs"), "/abs")
|
|
self.assertEqual(host._resolve_path("/base", "rel"),
|
|
"/base/rel")
|
|
self.assertEqual(host._thread_cli_result({"result": {"a": 1}},
|
|
"thread-list"),
|
|
{"a": 1})
|
|
self.assertEqual(host._thread_cli_result({"result": None},
|
|
"thread-list"), {})
|
|
with self.assertRaises(RuntimeError):
|
|
host._thread_cli_result({"error": "boom"}, "thread-list")
|
|
with self.assertRaises(RuntimeError):
|
|
host._thread_cli_result(None, "thread-list")
|
|
|
|
def test_server_command(self):
|
|
cmd = host._server_command("/bin/codex")
|
|
self.assertEqual(cmd[-1], "app-server")
|
|
self.assertIn("-c", cmd)
|
|
with self.assertRaises(ValueError):
|
|
host._server_command("/bin/codex", provider="other")
|
|
with mock.patch.dict(os.environ, {"SUB2API_BASE_URL": ""}):
|
|
with self.assertRaises(RuntimeError):
|
|
host._server_command("/bin/codex",
|
|
provider="sub2api")
|
|
with mock.patch.dict(os.environ,
|
|
{"SUB2API_BASE_URL":
|
|
"https://gw.example.com/"}):
|
|
cmd = host._server_command("/bin/codex",
|
|
provider="sub2api")
|
|
self.assertTrue(any("gw.example.com/v1" in c
|
|
for c in cmd))
|
|
|
|
|
|
class ServerTraceCloseTest(unittest.TestCase):
|
|
def test_append_trace_and_close(self):
|
|
tmp = tempfile.TemporaryDirectory()
|
|
try:
|
|
srv = host.Server(codex_bin="/bin/false")
|
|
state_path = os.path.join(tmp.name, "host-state.json")
|
|
srv.set_trace_file(state_path, "t1")
|
|
self.assertIsNotNone(srv.trace_path)
|
|
srv._append_trace({"method": "x", "params": {
|
|
"item": {"type": "agentMessage", "id": "m1",
|
|
"text": "hello"}}})
|
|
srv._append_trace({"method": "x", "params": {
|
|
"item": {"type": "userMessage", "text": "skip"}}})
|
|
with open(srv.trace_path) as f:
|
|
lines = f.readlines()
|
|
self.assertEqual(len(lines), 1)
|
|
self.assertIn("hello", lines[0])
|
|
# trace_path 为 None 时 _append_trace 是空操作。
|
|
srv.trace_path = None
|
|
srv._append_trace({"method": "x", "params": {
|
|
"item": {"type": "agentMessage", "id": "m2",
|
|
"text": "nope"}}})
|
|
finally:
|
|
tmp.cleanup()
|
|
|
|
def test_close_with_fake_proc(self):
|
|
class _Stdin:
|
|
def __init__(self):
|
|
self.closed = False
|
|
|
|
def close(self):
|
|
self.closed = True
|
|
|
|
class _Proc:
|
|
def __init__(self, wait_raises=False):
|
|
self.stdin = _Stdin()
|
|
self.stdout = _Stdin()
|
|
self.wait_raises = wait_raises
|
|
self.killed = False
|
|
|
|
def wait(self, timeout=None):
|
|
if self.wait_raises:
|
|
raise RuntimeError("not dead")
|
|
return 0
|
|
|
|
def kill(self):
|
|
self.killed = True
|
|
|
|
srv = host.Server(codex_bin="/bin/false")
|
|
proc = _Proc()
|
|
srv.proc = proc
|
|
srv.close()
|
|
self.assertTrue(proc.stdin.closed)
|
|
self.assertFalse(proc.killed)
|
|
proc2 = _Proc(wait_raises=True)
|
|
srv.proc = proc2
|
|
srv.close()
|
|
self.assertTrue(proc2.killed)
|
|
self.assertTrue(proc2.stdout.closed)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|