host
755029f220
对照 Codex app：wait_threads 首变+afterCursor+8上限、 read_thread 详情摘要、send_message_to_thread 复活失败任务、 handoff 登记与上报轮询、worktree 派生状态、PR checks JSON+ check-runs 日志与注释、automation GET/PATCH、clock.sleep （API+host）。诚实边界见 docs/thread-wait-ops.md：无 worktree 阶段流、无 automation 变化静默、无独立测试结果 记录、handoff 服务端不执行迁移、wait 仅数据库级变化检测。 三套件：cloud 940 OK（4 跳过）、host 91 OK、deploy 29 OK
53 lines
1.7 KiB
Python
53 lines
1.7 KiB
Python
import io
|
|
import json
|
|
import os
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
from contextlib import redirect_stdout
|
|
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 HostSleepTest(unittest.TestCase):
|
|
def _run(self, cli_args):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
state_path = os.path.join(tmp, "host-state.json")
|
|
with open(state_path, "w", encoding="utf-8") as f:
|
|
json.dump({"tasks": {}, "automations": []}, f)
|
|
cfg_path = os.path.join(tmp, "cfg.json")
|
|
with open(cfg_path, "w", encoding="utf-8") as f:
|
|
json.dump({"state": state_path}, f)
|
|
argv = ["host.py", "--config", cfg_path,
|
|
"--state", state_path] + cli_args
|
|
out = io.StringIO()
|
|
code = 0
|
|
with mock.patch.object(host, "Server",
|
|
lambda **kw: mock.MagicMock()):
|
|
with mock.patch.object(sys, "argv", argv):
|
|
with redirect_stdout(out):
|
|
try:
|
|
host.main()
|
|
except SystemExit as e:
|
|
code = e.code if isinstance(e.code, int) else 1
|
|
return code, out.getvalue()
|
|
|
|
def test_sleep_zero(self):
|
|
code, out = self._run(["sleep", "--seconds", "0"])
|
|
self.assertEqual(code, 0)
|
|
self.assertEqual(json.loads(out),
|
|
{"slept_seconds": 0.0, "interrupted": False})
|
|
|
|
def test_sleep_out_of_range(self):
|
|
code, _out = self._run(["sleep", "--seconds", "61"])
|
|
self.assertEqual(code, 1)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|