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
669 lines
29 KiB
Python
669 lines
29 KiB
Python
"""线程等待与移交操作单测：cloud/wait_ops 纯逻辑 + server 端点。
|
||
|
||
不跑真实模型、不联网、不调真实 gh：thread-read 用 stub
|
||
host.py（与 test_thread_ops.py 同款契约），pr checks 用
|
||
注入的 fake gh runner，worktree 状态用临时目录。覆盖鉴权、
|
||
权限、校验、状态机（移交转移、终态不可动）、超时语义与
|
||
诚实降级（异机 worktree / Docker 任务的 thread-read 仍回
|
||
200，只 thread 为空）。
|
||
"""
|
||
|
||
import http.client
|
||
import json
|
||
import os
|
||
import sqlite3
|
||
import sys
|
||
import tempfile
|
||
import threading
|
||
import unittest
|
||
from pathlib import Path
|
||
from types import SimpleNamespace
|
||
|
||
try:
|
||
from cloud.auth import AuthStore
|
||
from cloud.server import create_server
|
||
from cloud.wait_ops import (build_pr_diagnostics, create_handoff,
|
||
decode_wait_cursor,
|
||
detect_significant_change,
|
||
encode_wait_cursor, interruptible_sleep,
|
||
latest_handoff, parse_pr_checks,
|
||
poll_for_first_change, poll_task_statuses,
|
||
report_handoff, worktree_status)
|
||
except ImportError:
|
||
sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
|
||
from cloud.auth import AuthStore
|
||
from cloud.server import create_server
|
||
from cloud.wait_ops import (build_pr_diagnostics, create_handoff,
|
||
decode_wait_cursor,
|
||
detect_significant_change,
|
||
encode_wait_cursor, interruptible_sleep,
|
||
latest_handoff, parse_pr_checks,
|
||
poll_for_first_change, poll_task_statuses,
|
||
report_handoff, worktree_status)
|
||
|
||
STUB = r'''import json
|
||
import os
|
||
import sys
|
||
|
||
def main():
|
||
argv = sys.argv[1:]
|
||
cmd = None
|
||
rest = []
|
||
i = 0
|
||
while i < len(argv):
|
||
a = argv[i]
|
||
if a == "--codex-home":
|
||
i += 2
|
||
continue
|
||
if a.startswith("--codex-home="):
|
||
i += 1
|
||
continue
|
||
if cmd is None:
|
||
cmd = a
|
||
i += 1
|
||
continue
|
||
rest.append(a)
|
||
i += 1
|
||
if cmd == "thread-list":
|
||
print(json.dumps({"data": [
|
||
{"id": "t-old", "updatedAt": 100, "createdAt": 90},
|
||
{"id": "t-new", "updatedAt": 200, "createdAt": 190},
|
||
]}))
|
||
elif cmd == "thread-read":
|
||
print(json.dumps({"thread": {"id": "t-new", "updatedAt": 200}}))
|
||
else:
|
||
print(json.dumps({"error": "unknown subcommand"}))
|
||
sys.exit(1)
|
||
|
||
main()
|
||
'''
|
||
|
||
|
||
class WaitOpsPureTestCase(unittest.TestCase):
|
||
def test_parse_pr_checks_tabular_and_human(self):
|
||
checks, all_pass, failed, pending = parse_pr_checks(
|
||
"build\tpass\ntest\tfail\nlint\tpending\n")
|
||
self.assertEqual([c["state"] for c in checks],
|
||
["pass", "fail", "pending"])
|
||
self.assertFalse(all_pass)
|
||
self.assertEqual(failed, ["test"])
|
||
self.assertEqual(pending, ["lint"])
|
||
checks, all_pass, _, _ = parse_pr_checks(
|
||
"\u2713 build pass\n\u2713 test pass\n")
|
||
self.assertTrue(all_pass)
|
||
checks, all_pass, _, _ = parse_pr_checks("no checks reported")
|
||
self.assertEqual(checks, [])
|
||
self.assertFalse(all_pass)
|
||
|
||
def test_worktree_status_matrix(self):
|
||
self.assertEqual(
|
||
worktree_status("running", "/tmp/x", False, False)["status"],
|
||
"creating")
|
||
self.assertEqual(
|
||
worktree_status("completed", "/tmp/x", False, False)["status"],
|
||
"missing")
|
||
self.assertEqual(
|
||
worktree_status("running", "/tmp/x", True, True)["status"],
|
||
"ready")
|
||
self.assertEqual(
|
||
worktree_status("running", "/tmp/x", True, False)["status"],
|
||
"present_no_git")
|
||
remote = worktree_status("running", "/tmp/x", False, False,
|
||
remote_host="host-b")
|
||
self.assertEqual(remote["status"], "remote")
|
||
self.assertIsNone(remote["exists"])
|
||
|
||
def test_handoff_transitions(self):
|
||
conn = sqlite3.connect(":memory:")
|
||
conn.row_factory = sqlite3.Row
|
||
h = create_handoff(conn, "t1", "p1", "host-b", "u1",
|
||
source_host="host-a")
|
||
self.assertEqual(h["status"], "pending")
|
||
with self.assertRaises(ValueError):
|
||
create_handoff(conn, "t1", "p1", "host-c", "u1")
|
||
ok, h = report_handoff(conn, h["id"], "running", progress=40)
|
||
self.assertTrue(ok)
|
||
self.assertEqual((h["status"], h["progress"]), ("running", 40))
|
||
ok, h = report_handoff(conn, h["id"], "completed")
|
||
self.assertTrue(ok)
|
||
ok, err = report_handoff(conn, h["id"], "running")
|
||
self.assertFalse(ok)
|
||
self.assertIn("cannot move", err)
|
||
ok, err = report_handoff(conn, h["id"], "bogus")
|
||
self.assertFalse(ok)
|
||
self.assertEqual(latest_handoff(conn, "t1")["status"], "completed")
|
||
conn.close()
|
||
|
||
def test_wait_cursor_and_first_change(self):
|
||
base = {"a": {"status": "running", "pending_approvals": 0,
|
||
"pending_input": 0, "pending_followups": 0},
|
||
"b": {"status": "running", "pending_approvals": 0,
|
||
"pending_input": 0, "pending_followups": 0}}
|
||
cursor = encode_wait_cursor(base)
|
||
self.assertEqual(decode_wait_cursor(cursor), base)
|
||
with self.assertRaises(ValueError):
|
||
decode_wait_cursor("not-a-cursor!!")
|
||
cur = dict(base)
|
||
cur["b"] = {"status": "running", "pending_approvals": 0,
|
||
"pending_input": 1, "pending_followups": 0}
|
||
found = detect_significant_change(base, cur, ["a", "b"])
|
||
self.assertEqual(found[0], "b")
|
||
self.assertEqual(found[1], "needs_input")
|
||
cur2 = dict(base)
|
||
cur2["a"] = {"status": "failed", "pending_approvals": 0,
|
||
"pending_input": 0, "pending_followups": 0}
|
||
found = detect_significant_change(base, cur2, ["a", "b"])
|
||
self.assertEqual(found[1], "terminal")
|
||
# 基线里没有的任务不误报；无游标时已终态立即算变化。
|
||
found = detect_significant_change(
|
||
base, {"c": {"status": "completed", "pending_approvals": 0,
|
||
"pending_input": 0, "pending_followups": 0}},
|
||
["c"])
|
||
self.assertIsNone(found)
|
||
found = detect_significant_change(
|
||
None, {"c": {"status": "completed", "pending_approvals": 0,
|
||
"pending_input": 0, "pending_followups": 0}},
|
||
["c"])
|
||
self.assertEqual(found[1], "terminal")
|
||
changed, summaries, _w, timed_out = poll_for_first_change(
|
||
lambda: {"a": {"status": "completed",
|
||
"pending_approvals": 0, "pending_input": 0,
|
||
"pending_followups": 0}},
|
||
["a"], baseline=None, timeout_seconds=5)
|
||
self.assertFalse(timed_out)
|
||
self.assertEqual(changed["kind"], "terminal")
|
||
|
||
def test_build_pr_diagnostics_json_and_check_runs(self):
|
||
def run(cmd):
|
||
if "--json" in cmd:
|
||
return (0, json.dumps([
|
||
{"name": "build", "bucket": "pass",
|
||
"workflow": "CI",
|
||
"link": "https://example.invalid/ci/1",
|
||
"description": "ok"},
|
||
{"name": "test", "bucket": "fail",
|
||
"workflow": "CI",
|
||
"link": "https://example.invalid/ci/2",
|
||
"description": "1 failed"}]), "")
|
||
if "view" in cmd:
|
||
return (0, "abc123\n", "")
|
||
if any("commits" in c for c in cmd):
|
||
return (0, json.dumps({"check_runs": [
|
||
{"id": 99, "name": "test",
|
||
"conclusion": "failure"}]}), "")
|
||
if any("check-runs" in c for c in cmd):
|
||
return (0, json.dumps({
|
||
"id": 99, "conclusion": "failure",
|
||
"output": {
|
||
"title": "Test failed",
|
||
"summary": "1 test failed",
|
||
"text": "FAILED test_x: assert failed",
|
||
"annotations": [{
|
||
"path": "src/a.py", "start_line": 12,
|
||
"annotation_level": "failure",
|
||
"message": "assert failed"}]}}), "")
|
||
return (1, "", "unknown cmd")
|
||
diag = build_pr_diagnostics(run, 42, owner_repo="o/r",
|
||
head_sha="abc123")
|
||
self.assertFalse(diag["all_pass"])
|
||
self.assertEqual(diag["failed"], ["test"])
|
||
self.assertEqual(diag["checks"][1]["workflow"], "CI")
|
||
self.assertTrue(diag["check_runs"]["available"])
|
||
run_detail = diag["check_runs"]["runs"][0]
|
||
self.assertIn("FAILED test_x", run_detail["log_text"])
|
||
self.assertEqual(run_detail["annotations"][0]["path"], "src/a.py")
|
||
self.assertEqual(run_detail["annotations"][0]["start_line"], 12)
|
||
# 制表符回落：--json 失败时仍能解析，且无仓库时不查 API。
|
||
diag2 = build_pr_diagnostics(
|
||
lambda cmd: (1, "", "boom") if "--json" in cmd
|
||
else (0, "build\tpass\ntest\tfail\n", ""),
|
||
7)
|
||
self.assertEqual(diag2["failed"], ["test"])
|
||
self.assertFalse(diag2["check_runs"]["available"])
|
||
|
||
def test_poll_and_sleep(self):
|
||
calls = {"n": 0}
|
||
|
||
def fetch():
|
||
calls["n"] += 1
|
||
return {"a": "running"} if calls["n"] < 3 else {"a": "completed"}
|
||
|
||
ok, statuses, waited, timed_out = poll_task_statuses(
|
||
fetch, timeout_seconds=5)
|
||
self.assertTrue(ok)
|
||
self.assertFalse(timed_out)
|
||
self.assertEqual(statuses, {"a": "completed"})
|
||
ok, _, _, timed_out = poll_task_statuses(
|
||
lambda: {"a": "running"}, timeout_seconds=0)
|
||
self.assertFalse(ok)
|
||
self.assertTrue(timed_out)
|
||
ok, _, _, _ = poll_task_statuses(lambda: {}, timeout_seconds=5)
|
||
self.assertTrue(ok)
|
||
slept, interrupted = interruptible_sleep(0)
|
||
self.assertEqual((slept, interrupted), (0.0, False))
|
||
|
||
|
||
class ThreadWaitServerTestCase(unittest.TestCase):
|
||
def setUp(self):
|
||
self.tmp = tempfile.TemporaryDirectory()
|
||
self.db_path = os.path.join(self.tmp.name, "control.db")
|
||
self.auth = AuthStore(self.db_path)
|
||
self.owner_id, self.owner_key = self.auth.create_user("owner")
|
||
self.reader_id, self.reader_key = self.auth.create_user("reader")
|
||
self.outsider_id, self.outsider_key = self.auth.create_user(
|
||
"outsider")
|
||
self.stub_path = os.path.join(self.tmp.name, "stub_host.py")
|
||
with open(self.stub_path, "w", encoding="utf-8") as f:
|
||
f.write(STUB)
|
||
self._old_host_py = os.environ.get("CODEX_HOST_PY")
|
||
os.environ["CODEX_HOST_PY"] = self.stub_path
|
||
self.server = create_server(self.db_path, "127.0.0.1", 0)
|
||
self.server.tenants_root = os.path.join(self.tmp.name, "tenants")
|
||
self.server.host_worktrees_dir = os.path.join(
|
||
self.tmp.name, "worktrees")
|
||
self.port = self.server.server_address[1]
|
||
self.thread = threading.Thread(
|
||
target=self.server.serve_forever, daemon=True)
|
||
self.thread.start()
|
||
status, self.project = self._req(
|
||
"POST", "/projects", {"name": "demo"}, self.owner_key)
|
||
self.assertEqual(status, 201, self.project)
|
||
self.auth.set_member(self.project["id"], self.owner_id,
|
||
self.reader_id, "viewer")
|
||
status, self.project2 = self._req(
|
||
"POST", "/projects", {"name": "other"}, self.owner_key)
|
||
self.assertEqual(status, 201, self.project2)
|
||
|
||
def tearDown(self):
|
||
self.server.shutdown()
|
||
self.server.server_close()
|
||
self.thread.join(timeout=5)
|
||
if self._old_host_py is None:
|
||
os.environ.pop("CODEX_HOST_PY", None)
|
||
else:
|
||
os.environ["CODEX_HOST_PY"] = self._old_host_py
|
||
try:
|
||
self.auth.conn.close()
|
||
except Exception:
|
||
pass
|
||
self.tmp.cleanup()
|
||
|
||
def _req(self, method, path, body=None, key=None):
|
||
conn = http.client.HTTPConnection(
|
||
"127.0.0.1", self.port, timeout=10)
|
||
hdrs = {}
|
||
payload = None
|
||
if body is not None:
|
||
payload = json.dumps(body).encode("utf-8")
|
||
hdrs["Content-Type"] = "application/json"
|
||
if key is not None:
|
||
hdrs["X-API-Key"] = key
|
||
try:
|
||
conn.request(method, path, body=payload, headers=hdrs)
|
||
resp = conn.getresponse()
|
||
data = resp.read()
|
||
finally:
|
||
conn.close()
|
||
try:
|
||
parsed = json.loads(data.decode("utf-8")) if data else None
|
||
except ValueError:
|
||
parsed = None
|
||
return resp.status, parsed
|
||
|
||
def _create_task(self, project_id=None, **extra):
|
||
body = {"project_id": project_id or self.project["id"],
|
||
"prompt": "做事"}
|
||
body.update(extra)
|
||
status, task = self._req("POST", "/tasks", body, self.owner_key)
|
||
self.assertEqual(status, 201, task)
|
||
return task
|
||
|
||
def _set_task(self, task_id, **fields):
|
||
conn = sqlite3.connect(self.db_path)
|
||
try:
|
||
for key, value in fields.items():
|
||
conn.execute(
|
||
"UPDATE tasks SET %s = ? WHERE id = ?" % key,
|
||
(value, task_id))
|
||
conn.commit()
|
||
finally:
|
||
conn.close()
|
||
|
||
def test_thread_read_summary_and_docker_downgrade(self):
|
||
# 默认执行器是 docker（线程守卫会挡），读摘要的 thread
|
||
# 部分必须用显式 local 任务测；docker 任务测降级分支。
|
||
task = self._create_task(executor="local")
|
||
self._set_task(task["id"], status="running")
|
||
status, body = self._req(
|
||
"GET", "/tasks/%s/thread/read" % task["id"],
|
||
None, self.reader_key)
|
||
self.assertEqual(status, 200, body)
|
||
self.assertEqual(body["status"], "running")
|
||
self.assertEqual(body["state"], "running")
|
||
self.assertEqual(body["thread_id"], "t-new")
|
||
self.assertEqual(body["thread"]["id"], "t-new")
|
||
self.assertEqual(body["progress"]["trace_entries"], 0)
|
||
self.assertIsNone(body["result"]["text"])
|
||
self.assertEqual(body["changed_files"], [])
|
||
self.assertEqual(body["commands"], [])
|
||
self.assertEqual(body["recent_activity"], [])
|
||
self.assertFalse(body["attention"]["blocked"])
|
||
self.assertFalse(body["attention"]["needs_instruction"])
|
||
self.assertIsNone(body["test_results"])
|
||
docker = self._create_task()
|
||
status, body = self._req(
|
||
"GET", "/tasks/%s/thread/read" % docker["id"],
|
||
None, self.owner_key)
|
||
self.assertEqual(status, 200, body)
|
||
self.assertIsNone(body["thread"])
|
||
self.assertIn("docker", body["note"])
|
||
status, _ = self._req(
|
||
"GET", "/tasks/%s/thread/read" % task["id"],
|
||
None, self.outsider_key)
|
||
self.assertEqual(status, 403)
|
||
status, _ = self._req(
|
||
"GET", "/tasks/nope/thread/read", None, self.owner_key)
|
||
self.assertEqual(status, 404)
|
||
|
||
def test_thread_message_followup_semantics(self):
|
||
task = self._create_task()
|
||
self._set_task(task["id"], status="running")
|
||
status, body = self._req(
|
||
"POST", "/tasks/%s/thread/message" % task["id"],
|
||
{"body": "下一步改 README"}, self.owner_key)
|
||
self.assertEqual(status, 201, body)
|
||
self.assertEqual(body["delivery"], "queued")
|
||
status, body = self._req(
|
||
"POST", "/tasks/%s/thread/message" % task["id"],
|
||
{"body": "立刻改", "mode": "direct"}, self.owner_key)
|
||
self.assertEqual(status, 201, body)
|
||
self.assertEqual(body["delivery"], "direct")
|
||
status, body = self._req(
|
||
"POST", "/tasks/%s/thread/message" % task["id"],
|
||
{"body": "坏模式", "mode": "bogus"}, self.owner_key)
|
||
self.assertEqual(status, 400, body)
|
||
# 失败任务发修复指令：复活（不是 409），状态回 queued。
|
||
done = self._create_task()
|
||
self._set_task(done["id"], status="failed",
|
||
error_text="compile error")
|
||
status, body = self._req(
|
||
"POST", "/tasks/%s/thread/message" % done["id"],
|
||
{"body": "继续修复编译错误并重跑测试"}, self.owner_key)
|
||
self.assertEqual(status, 201, body)
|
||
self.assertEqual(body["delivery"], "revived")
|
||
status, task_after = self._req(
|
||
"GET", "/tasks/%s" % done["id"], None, self.owner_key)
|
||
self.assertEqual(task_after["status"], "queued")
|
||
self.assertIsNone(task_after["error_text"])
|
||
# 已合并任务不能复活（worktree 已删），应 409。
|
||
merged = self._create_task()
|
||
self._set_task(merged["id"], status="merged")
|
||
status, body = self._req(
|
||
"POST", "/tasks/%s/thread/message" % merged["id"],
|
||
{"body": "再改一下"}, self.owner_key)
|
||
self.assertEqual(status, 409, body)
|
||
|
||
def test_handoff_lifecycle_and_cross_project(self):
|
||
task = self._create_task()
|
||
status, body = self._req(
|
||
"GET", "/tasks/%s/handoff" % task["id"], None, self.owner_key)
|
||
self.assertEqual(status, 200, body)
|
||
self.assertIsNone(body["handoff"])
|
||
status, h = self._req(
|
||
"POST", "/tasks/%s/handoff" % task["id"],
|
||
{"target_host": "host-b"}, self.owner_key)
|
||
self.assertEqual(status, 201, h)
|
||
self.assertEqual(h["status"], "pending")
|
||
status, body = self._req(
|
||
"POST", "/tasks/%s/handoff" % task["id"],
|
||
{"target_host": "host-c"}, self.owner_key)
|
||
self.assertEqual(status, 409, body)
|
||
status, body = self._req(
|
||
"POST", "/tasks/%s/handoff/%s/report" % (task["id"], h["id"]),
|
||
{"status": "running", "progress": 40}, self.owner_key)
|
||
self.assertEqual(status, 200, body)
|
||
self.assertEqual((body["status"], body["progress"]),
|
||
("running", 40))
|
||
status, body = self._req(
|
||
"GET", "/tasks/%s/handoff" % task["id"], None, self.owner_key)
|
||
self.assertEqual(body["handoff"]["status"], "running")
|
||
other = self._create_task(self.project2["id"])
|
||
status, body = self._req(
|
||
"POST", "/tasks/%s/handoff/%s/report" % (other["id"], h["id"]),
|
||
{"status": "completed"}, self.owner_key)
|
||
self.assertEqual(status, 404, body)
|
||
status, body = self._req(
|
||
"POST", "/tasks/%s/handoff/%s/report" % (task["id"], h["id"]),
|
||
{"status": "completed"}, self.owner_key)
|
||
self.assertEqual(status, 200, body)
|
||
status, body = self._req(
|
||
"POST", "/tasks/%s/handoff/%s/report" % (task["id"], h["id"]),
|
||
{"status": "running"}, self.owner_key)
|
||
self.assertEqual(status, 409, body)
|
||
|
||
def test_worktree_status_derived(self):
|
||
task = self._create_task()
|
||
status, body = self._req(
|
||
"GET", "/tasks/%s/worktree" % task["id"], None, self.owner_key)
|
||
self.assertEqual(status, 200, body)
|
||
self.assertEqual(body["status"], "creating")
|
||
import re
|
||
safe = re.sub(r"[^A-Za-z0-9_.-]", "-", task["id"])
|
||
path = os.path.join(self.server.host_worktrees_dir, safe)
|
||
os.makedirs(os.path.join(path, ".git"), exist_ok=True)
|
||
status, body = self._req(
|
||
"GET", "/tasks/%s/worktree" % task["id"], None, self.owner_key)
|
||
self.assertEqual(body["status"], "ready")
|
||
self.assertTrue(body["exists"])
|
||
self._set_task(task["id"], worktree_host="other-host")
|
||
status, body = self._req(
|
||
"GET", "/tasks/%s/worktree" % task["id"], None, self.owner_key)
|
||
self.assertEqual(body["status"], "remote")
|
||
self.assertIsNone(body["exists"])
|
||
|
||
def test_pr_checks_and_no_pr(self):
|
||
task = self._create_task()
|
||
status, body = self._req(
|
||
"GET", "/tasks/%s/pr/checks" % task["id"], None, self.owner_key)
|
||
self.assertEqual(status, 409, body)
|
||
self._set_task(task["id"], pr_number=42,
|
||
pr_url="https://example.invalid/pr/42")
|
||
# 制表符回落 + 无仓库时不查 check-runs API。
|
||
self.server.pr_gh_runner = lambda cmd, cwd: SimpleNamespace(
|
||
returncode=0, stdout="build\tpass\ntest\tfail\n",
|
||
stderr="")
|
||
status, body = self._req(
|
||
"GET", "/tasks/%s/pr/checks" % task["id"], None, self.owner_key)
|
||
self.assertEqual(status, 200, body)
|
||
self.assertFalse(body["all_pass"])
|
||
self.assertEqual(body["failed"], ["test"])
|
||
self.assertIn("test", body["diagnostics"])
|
||
self.assertFalse(body["check_runs"]["available"])
|
||
# 有仓库时：JSON checks + check-runs 日志与行号注释。
|
||
status, repo_proj = self._req(
|
||
"POST", "/projects",
|
||
{"name": "repo-proj",
|
||
"repo_url": "https://github.com/openqj/demo.git"},
|
||
self.owner_key)
|
||
self.assertEqual(status, 201, repo_proj)
|
||
repo_task = self._create_task(repo_proj["id"])
|
||
self._set_task(repo_task["id"], pr_number=43,
|
||
pr_url="https://example.invalid/pr/43")
|
||
|
||
def _gh(cmd, cwd):
|
||
if "--json" in cmd:
|
||
return SimpleNamespace(returncode=0, stdout=json.dumps([
|
||
{"name": "build", "bucket": "pass",
|
||
"workflow": "CI",
|
||
"link": "https://example.invalid/ci/1",
|
||
"description": "ok"},
|
||
{"name": "test", "bucket": "fail",
|
||
"workflow": "CI",
|
||
"link": "https://example.invalid/ci/2",
|
||
"description": "1 failed"}]), stderr="")
|
||
if "view" in cmd:
|
||
return SimpleNamespace(returncode=0, stdout="abc123\n",
|
||
stderr="")
|
||
if any("commits" in c for c in cmd):
|
||
return SimpleNamespace(returncode=0, stdout=json.dumps(
|
||
{"check_runs": [{"id": 99, "name": "test",
|
||
"conclusion": "failure"}]}),
|
||
stderr="")
|
||
if any("check-runs" in c for c in cmd):
|
||
return SimpleNamespace(returncode=0, stdout=json.dumps({
|
||
"id": 99, "conclusion": "failure",
|
||
"output": {
|
||
"title": "Test failed",
|
||
"summary": "1 test failed",
|
||
"text": "FAILED test_x: assert failed",
|
||
"annotations": [{
|
||
"path": "src/a.py", "start_line": 12,
|
||
"annotation_level": "failure",
|
||
"message": "assert failed"}]}}),
|
||
stderr="")
|
||
return SimpleNamespace(returncode=1, stdout="", stderr="boom")
|
||
|
||
self.server.pr_gh_runner = _gh
|
||
status, body = self._req(
|
||
"GET", "/tasks/%s/pr/checks" % repo_task["id"],
|
||
None, self.owner_key)
|
||
self.assertEqual(status, 200, body)
|
||
self.assertEqual(body["checks"][1]["workflow"], "CI")
|
||
self.assertIn("example.invalid/ci/2", body["checks"][1]["link"])
|
||
self.assertTrue(body["check_runs"]["available"])
|
||
run_detail = body["check_runs"]["runs"][0]
|
||
self.assertIn("FAILED test_x", run_detail["log_text"])
|
||
self.assertEqual(run_detail["annotations"][0]["path"], "src/a.py")
|
||
self.assertEqual(run_detail["annotations"][0]["start_line"], 12)
|
||
self._set_task(task["id"], worktree_host="other-host")
|
||
status, body = self._req(
|
||
"GET", "/tasks/%s/pr/checks" % task["id"], None, self.owner_key)
|
||
self.assertEqual(status, 409, body)
|
||
|
||
def test_sleep_and_wait(self):
|
||
task = self._create_task()
|
||
status, body = self._req(
|
||
"POST", "/tasks/%s/sleep" % task["id"],
|
||
{"seconds": 0}, self.owner_key)
|
||
self.assertEqual(status, 200, body)
|
||
self.assertEqual(body["slept_seconds"], 0)
|
||
status, body = self._req(
|
||
"POST", "/tasks/%s/sleep" % task["id"],
|
||
{"seconds": 61}, self.owner_key)
|
||
self.assertEqual(status, 400, body)
|
||
parent = self._create_task()
|
||
child = self._create_task()
|
||
self._set_task(child["id"], status="completed",
|
||
parent_task_id=parent["id"])
|
||
# 无游标：已终态的子任务立即作为首变返回，不等满超时。
|
||
status, body = self._req(
|
||
"POST", "/tasks/%s/wait" % parent["id"],
|
||
{"timeout_seconds": 30}, self.owner_key)
|
||
self.assertEqual(status, 200, body)
|
||
self.assertTrue(body["all_terminal"])
|
||
self.assertFalse(body["timed_out"])
|
||
self.assertEqual(body["children"][0]["task_id"], child["id"])
|
||
self.assertEqual(body["changed_task"]["task_id"], child["id"])
|
||
self.assertEqual(body["changed_task"]["kind"], "terminal")
|
||
cursor = body["cursor"]
|
||
# 同一游标再等：已终态不再重复报，超时回进度 + 新游标。
|
||
status, body = self._req(
|
||
"POST", "/tasks/%s/wait" % parent["id"],
|
||
{"timeout_seconds": 0, "after_cursor": cursor},
|
||
self.owner_key)
|
||
self.assertEqual(status, 200, body)
|
||
self.assertTrue(body["timed_out"])
|
||
self.assertIsNone(body["changed_task"])
|
||
# 非法游标 400；超 8 个 400；运行中 + 超时 0 回超时。
|
||
status, body = self._req(
|
||
"POST", "/tasks/%s/wait" % parent["id"],
|
||
{"timeout_seconds": 0, "after_cursor": "bogus!!"},
|
||
self.owner_key)
|
||
self.assertEqual(status, 400, body)
|
||
running = self._create_task()
|
||
status, body = self._req(
|
||
"POST", "/tasks/%s/wait" % parent["id"],
|
||
{"task_ids": [running["id"]], "timeout_seconds": 0},
|
||
self.owner_key)
|
||
self.assertEqual(status, 200, body)
|
||
self.assertTrue(body["timed_out"])
|
||
self.assertFalse(body["all_terminal"])
|
||
status, body = self._req(
|
||
"POST", "/tasks/%s/wait" % parent["id"],
|
||
{"task_ids": ["x"] * 9, "timeout_seconds": 0},
|
||
self.owner_key)
|
||
self.assertEqual(status, 400, body)
|
||
status, body = self._req(
|
||
"POST", "/tasks/%s/wait" % parent["id"],
|
||
{"timeout_seconds": 121}, self.owner_key)
|
||
self.assertEqual(status, 400, body)
|
||
|
||
def test_wait_cursor_approval_trigger(self):
|
||
# 基线游标（运行中、无待审批）-> 新建 elicitation 审批
|
||
# -> 再等同一游标，立即报 needs_input（不是等满超时）。
|
||
parent = self._create_task()
|
||
target = self._create_task()
|
||
status, body = self._req(
|
||
"POST", "/tasks/%s/wait" % parent["id"],
|
||
{"task_ids": [target["id"]], "timeout_seconds": 0},
|
||
self.owner_key)
|
||
self.assertEqual(status, 200, body)
|
||
self.assertTrue(body["timed_out"])
|
||
cursor = body["cursor"]
|
||
status, appr = self._req(
|
||
"POST", "/tasks/%s/approvals" % target["id"],
|
||
{"kind": "elicitation", "summary": "请确认是否继续"},
|
||
self.owner_key)
|
||
self.assertEqual(status, 201, appr)
|
||
status, body = self._req(
|
||
"POST", "/tasks/%s/wait" % parent["id"],
|
||
{"task_ids": [target["id"]], "timeout_seconds": 30,
|
||
"after_cursor": cursor},
|
||
self.owner_key)
|
||
self.assertEqual(status, 200, body)
|
||
self.assertFalse(body["timed_out"])
|
||
self.assertEqual(body["changed_task"]["kind"], "needs_input")
|
||
self.assertEqual(body["changed_task"]["task_id"], target["id"])
|
||
|
||
def test_automation_update_and_get(self):
|
||
status, auto = self._req(
|
||
"POST", "/projects/%s/automations" % self.project["id"],
|
||
{"name": "daily", "prompt": "做每日检查"}, self.owner_key)
|
||
self.assertEqual(status, 201, auto)
|
||
status, body = self._req(
|
||
"PATCH", "/projects/%s/automations/%s"
|
||
% (self.project["id"], auto["id"]),
|
||
{"interval_seconds": 7200}, self.owner_key)
|
||
self.assertEqual(status, 200, body)
|
||
self.assertEqual(body["interval_seconds"], 7200)
|
||
status, body = self._req(
|
||
"GET", "/projects/%s/automations/%s"
|
||
% (self.project["id"], auto["id"]),
|
||
None, self.owner_key)
|
||
self.assertEqual(status, 200, body)
|
||
self.assertEqual(body["runs"], [])
|
||
status, body = self._req(
|
||
"PATCH", "/projects/%s/automations/%s"
|
||
% (self.project["id"], auto["id"]),
|
||
{}, self.owner_key)
|
||
self.assertEqual(status, 400, body)
|
||
status, body = self._req(
|
||
"PATCH", "/projects/%s/automations/%s"
|
||
% (self.project2["id"], auto["id"]),
|
||
{"interval_seconds": 3600}, self.owner_key)
|
||
self.assertEqual(status, 404, body)
|
||
status, body = self._req(
|
||
"POST", "/projects/%s/automations/%s/cancel"
|
||
% (self.project["id"], auto["id"]),
|
||
{}, self.owner_key)
|
||
self.assertEqual(status, 200, body)
|
||
status, body = self._req(
|
||
"PATCH", "/projects/%s/automations/%s"
|
||
% (self.project["id"], auto["id"]),
|
||
{"interval_seconds": 3600}, self.owner_key)
|
||
self.assertEqual(status, 409, body)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main()
|