host
ef532a38c2
线程：host.py 新增 thread-list/read/rename/archive/unarchive/compact 子命令（纯 JSON stdout 契约，read/rename/unarchive/compact 先 resume）；服务端按任务定 scope（thread/list 的 cwd 过滤解析，Docker/cloud/异机 409，改操作仅终态，审计 thread_*）；控制台线程区（元数据、列表、重命名、归档、取消归档、压缩，终态门控）。 skills/MCP：host.py 新增 skills-list/apps-list/mcp-servers-list/mcp-servers-reload/mcp-server-oauth-login 子命令；服务端按项目 scope（读 read，改 manage_project，审计 mcp_*）；控制台技能与 MCP 区（列表、重载、OAuth 发起展示，绝不谎称登录完成）。 测试：host 84、cloud 687（4 跳过）、deploy 19 全过。诚实边界：Docker 任务线程在容器内故 409；无仓库任务推不出 cwd 时 404；OAuth 只做到发起透传，完成态靠协议通知；merged/merge-failed 改操作 409（比控制台门控窄，服务端报错原文展示）。
431 lines
17 KiB
Python
431 lines
17 KiB
Python
"""Thread ops server tests: GET/POST /tasks/<id>/thread* endpoints.
|
|
|
|
No real host.py CLI, no real codex binary: a stub script written into
|
|
tmp pretends to be host.py and prints canned JSON per the frozen CLI
|
|
contract - one JSON object on stdout, nonzero exit plus {"error": ...}
|
|
on failure. The stub reads a scenario JSON file and appends every
|
|
invocation to a log file so tests can assert the argv the server
|
|
passed, including the derived --cwd worktree path.
|
|
"""
|
|
|
|
import http.client
|
|
import json
|
|
import os
|
|
import sqlite3
|
|
import sys
|
|
import tempfile
|
|
import threading
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
try:
|
|
from cloud.auth import AuthStore
|
|
from cloud.server import create_server
|
|
except ImportError:
|
|
sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
|
|
from cloud.auth import AuthStore
|
|
from cloud.server import create_server
|
|
|
|
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
|
|
log = os.environ.get("THREADOPS_STUB_LOG")
|
|
if log:
|
|
with open(log, "a", encoding="utf-8") as f:
|
|
f.write(json.dumps({"cmd": cmd, "argv": rest}) + "\n")
|
|
scenario = {}
|
|
path = os.environ.get("THREADOPS_STUB_SCENARIO")
|
|
if path and os.path.exists(path):
|
|
with open(path, "r", encoding="utf-8") as f:
|
|
scenario = json.load(f)
|
|
if scenario.get("fail_cmd") == cmd:
|
|
print(json.dumps({"error": scenario.get("fail_error", "boom")}))
|
|
sys.exit(1)
|
|
if scenario.get("raw_stdout_for") == cmd:
|
|
print(scenario.get("raw_stdout", ""))
|
|
return
|
|
opts = {}
|
|
j = 0
|
|
while j < len(rest):
|
|
a = rest[j]
|
|
if a.startswith("--") and j + 1 < len(rest) and not rest[j + 1].startswith("--"):
|
|
opts[a] = rest[j + 1]
|
|
j += 2
|
|
elif a.startswith("--"):
|
|
opts[a] = True
|
|
j += 1
|
|
else:
|
|
j += 1
|
|
if cmd == "thread-list":
|
|
print(json.dumps({"data": scenario.get("threads", [
|
|
{"id": "t-old", "updatedAt": 100, "createdAt": 90},
|
|
{"id": "t-new", "updatedAt": 200, "createdAt": 190},
|
|
])}))
|
|
elif cmd == "thread-read":
|
|
thread = dict(scenario.get("thread", {"id": "t-new", "updatedAt": 200}))
|
|
if opts.get("--include-turns"):
|
|
thread["turns"] = [{"id": "turn-1"}]
|
|
print(json.dumps({"thread": thread}))
|
|
elif cmd == "thread-rename":
|
|
print(json.dumps({"ok": True, "name": opts.get("--name")}))
|
|
elif cmd in ("thread-archive", "thread-unarchive", "thread-compact"):
|
|
print(json.dumps({"ok": True}))
|
|
else:
|
|
print(json.dumps({"error": "unknown subcommand"}))
|
|
sys.exit(1)
|
|
|
|
main()
|
|
'''
|
|
|
|
|
|
class ThreadOpsTestCase(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.scenario_path = os.path.join(self.tmp.name, "scenario.json")
|
|
self.log_path = os.path.join(self.tmp.name, "stub.log")
|
|
self._write_scenario({})
|
|
self._old_host_py = os.environ.get("CODEX_HOST_PY")
|
|
os.environ["CODEX_HOST_PY"] = self.stub_path
|
|
os.environ["THREADOPS_STUB_SCENARIO"] = self.scenario_path
|
|
os.environ["THREADOPS_STUB_LOG"] = self.log_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")
|
|
|
|
def tearDown(self):
|
|
self.server.shutdown()
|
|
self.server.server_close()
|
|
self.thread.join(timeout=5)
|
|
try:
|
|
self.auth.conn.close()
|
|
except Exception:
|
|
pass
|
|
if self._old_host_py is None:
|
|
os.environ.pop("CODEX_HOST_PY", None)
|
|
else:
|
|
os.environ["CODEX_HOST_PY"] = self._old_host_py
|
|
os.environ.pop("THREADOPS_STUB_SCENARIO", None)
|
|
os.environ.pop("THREADOPS_STUB_LOG", None)
|
|
self.tmp.cleanup()
|
|
|
|
def _write_scenario(self, scenario):
|
|
with open(self.scenario_path, "w", encoding="utf-8") as f:
|
|
json.dump(scenario, f)
|
|
|
|
def _stub_calls(self, cmd=None):
|
|
if not os.path.exists(self.log_path):
|
|
return []
|
|
calls = []
|
|
with open(self.log_path, "r", encoding="utf-8") as f:
|
|
for line in f:
|
|
line = line.strip()
|
|
if line:
|
|
calls.append(json.loads(line))
|
|
if cmd is not None:
|
|
calls = [c for c in calls if c["cmd"] == cmd]
|
|
return calls
|
|
|
|
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, **extra):
|
|
body = {"project_id": self.project["id"], "prompt": "do work"}
|
|
body.update(extra)
|
|
status, task = self._req("POST", "/tasks", body, self.owner_key)
|
|
self.assertEqual(status, 201, task)
|
|
return task
|
|
|
|
def _set_status(self, task_id, status):
|
|
conn = sqlite3.connect(self.db_path)
|
|
try:
|
|
conn.execute("UPDATE tasks SET status = ? WHERE id = ?",
|
|
(status, task_id))
|
|
conn.commit()
|
|
finally:
|
|
conn.close()
|
|
|
|
def _audit_actions(self, project_id):
|
|
conn = sqlite3.connect(self.db_path)
|
|
try:
|
|
rows = conn.execute(
|
|
"SELECT action FROM audit_log WHERE project_id = ?",
|
|
(project_id,)).fetchall()
|
|
finally:
|
|
conn.close()
|
|
return [r[0] for r in rows]
|
|
|
|
def test_unauthorized_without_key(self):
|
|
task = self._create_task(executor="local")
|
|
status, body = self._req(
|
|
"GET", "/tasks/%s/thread" % task["id"], None, None)
|
|
self.assertEqual(status, 401, body)
|
|
|
|
def test_forbidden_outsider_read(self):
|
|
task = self._create_task(executor="local")
|
|
status, body = self._req(
|
|
"GET", "/tasks/%s/threads" % task["id"], None,
|
|
self.outsider_key)
|
|
self.assertEqual(status, 403, body)
|
|
|
|
def test_forbidden_reader_cannot_mutate(self):
|
|
task = self._create_task(executor="local")
|
|
self._set_status(task["id"], "completed")
|
|
status, body = self._req(
|
|
"POST", "/tasks/%s/thread/archive" % task["id"], {},
|
|
self.reader_key)
|
|
self.assertEqual(status, 403, body)
|
|
|
|
def test_forbidden_outsider_cannot_mutate(self):
|
|
# P2 接线新增：非成员改操作同样 403，且不留审计。
|
|
task = self._create_task(executor="local")
|
|
self._set_status(task["id"], "completed")
|
|
status, body = self._req(
|
|
"POST", "/tasks/%s/thread/archive" % task["id"], {},
|
|
self.outsider_key)
|
|
self.assertEqual(status, 403, body)
|
|
self.assertNotIn("thread_archived",
|
|
self._audit_actions(self.project["id"]))
|
|
|
|
def test_task_not_found(self):
|
|
status, body = self._req(
|
|
"GET", "/tasks/nope/thread", None, self.owner_key)
|
|
self.assertEqual(status, 404, body)
|
|
status, body = self._req(
|
|
"POST", "/tasks/nope/thread/archive", {}, self.owner_key)
|
|
self.assertEqual(status, 404, body)
|
|
|
|
def test_docker_task_409(self):
|
|
task = self._create_task(executor="docker")
|
|
self._set_status(task["id"], "completed")
|
|
status, body = self._req(
|
|
"GET", "/tasks/%s/threads" % task["id"], None, self.owner_key)
|
|
self.assertEqual(status, 409, body)
|
|
status, body = self._req(
|
|
"POST", "/tasks/%s/thread/archive" % task["id"], {},
|
|
self.owner_key)
|
|
self.assertEqual(status, 409, body)
|
|
self.assertEqual(self._stub_calls(), [])
|
|
|
|
def test_terminal_gate_409_and_read_allowed(self):
|
|
task = self._create_task(executor="local")
|
|
self._set_status(task["id"], "running")
|
|
status, body = self._req(
|
|
"POST", "/tasks/%s/thread/archive" % task["id"], {},
|
|
self.owner_key)
|
|
self.assertEqual(status, 409, body)
|
|
self.assertEqual(body["status"], "running")
|
|
status, body = self._req(
|
|
"GET", "/tasks/%s/thread" % task["id"], None, self.owner_key)
|
|
self.assertEqual(status, 200, body)
|
|
|
|
def test_merged_task_mutation_409_but_read_ok(self):
|
|
task = self._create_task(executor="local")
|
|
self._set_status(task["id"], "merged")
|
|
status, _body = self._req(
|
|
"POST", "/tasks/%s/thread/archive" % task["id"], {},
|
|
self.owner_key)
|
|
self.assertEqual(status, 409)
|
|
status, body = self._req(
|
|
"GET", "/tasks/%s/thread" % task["id"], None, self.owner_key)
|
|
self.assertEqual(status, 200, body)
|
|
|
|
def test_threads_list_passthrough_and_cwd(self):
|
|
task = self._create_task(executor="local")
|
|
status, body = self._req(
|
|
"GET", "/tasks/%s/threads" % task["id"], None, self.owner_key)
|
|
self.assertEqual(status, 200, body)
|
|
self.assertEqual(
|
|
[t["id"] for t in body["threads"]], ["t-old", "t-new"])
|
|
expected_cwd = os.path.join(
|
|
self.tmp.name, "worktrees", task["id"])
|
|
self.assertEqual(body["cwd"], expected_cwd)
|
|
calls = self._stub_calls("thread-list")
|
|
self.assertEqual(len(calls), 1)
|
|
self.assertIn("--cwd", calls[0]["argv"])
|
|
idx = calls[0]["argv"].index("--cwd")
|
|
self.assertEqual(calls[0]["argv"][idx + 1], expected_cwd)
|
|
|
|
def test_thread_latest_non_archived_rule(self):
|
|
self._write_scenario({"threads": [
|
|
{"id": "t-old", "updatedAt": 100, "createdAt": 90},
|
|
{"id": "t-arch", "updatedAt": 300, "createdAt": 290,
|
|
"archived": True},
|
|
{"id": "t-new", "updatedAt": 200, "createdAt": 190},
|
|
]})
|
|
task = self._create_task(executor="local")
|
|
status, body = self._req(
|
|
"GET", "/tasks/%s/thread" % task["id"], None, self.owner_key)
|
|
self.assertEqual(status, 200, body)
|
|
self.assertEqual(body["thread_id"], "t-new")
|
|
|
|
def test_thread_tie_break_later_in_list_wins(self):
|
|
self._write_scenario({"threads": [
|
|
{"id": "t-a", "updatedAt": 50, "createdAt": 50},
|
|
{"id": "t-b", "updatedAt": 50, "createdAt": 50},
|
|
]})
|
|
task = self._create_task(executor="local")
|
|
status, body = self._req(
|
|
"GET", "/tasks/%s/thread" % task["id"], None, self.owner_key)
|
|
self.assertEqual(status, 200, body)
|
|
self.assertEqual(body["thread_id"], "t-b")
|
|
|
|
def test_thread_not_found_when_empty(self):
|
|
self._write_scenario({"threads": []})
|
|
task = self._create_task(executor="local")
|
|
status, body = self._req(
|
|
"GET", "/tasks/%s/thread" % task["id"], None, self.owner_key)
|
|
self.assertEqual(status, 404, body)
|
|
status, body = self._req(
|
|
"GET", "/tasks/%s/threads" % task["id"], None, self.owner_key)
|
|
self.assertEqual(status, 200, body)
|
|
self.assertEqual(body["threads"], [])
|
|
|
|
def test_thread_read_include_turns(self):
|
|
task = self._create_task(executor="local")
|
|
status, body = self._req(
|
|
"GET", "/tasks/%s/thread" % task["id"], None, self.owner_key)
|
|
self.assertEqual(status, 200, body)
|
|
self.assertNotIn("turns", body["thread"])
|
|
status, body = self._req(
|
|
"GET", "/tasks/%s/thread?include_turns=1" % task["id"],
|
|
None, self.owner_key)
|
|
self.assertEqual(status, 200, body)
|
|
self.assertEqual(body["thread"]["turns"], [{"id": "turn-1"}])
|
|
calls = self._stub_calls("thread-read")
|
|
self.assertIn("--include-turns", calls[1]["argv"])
|
|
self.assertNotIn("--include-turns", calls[0]["argv"])
|
|
|
|
def test_rename_ok_and_audit(self):
|
|
task = self._create_task(executor="local")
|
|
self._set_status(task["id"], "completed")
|
|
status, body = self._req(
|
|
"POST", "/tasks/%s/thread/rename" % task["id"],
|
|
{"name": "new name"}, self.owner_key)
|
|
self.assertEqual(status, 200, body)
|
|
self.assertEqual(body["result"]["name"], "new name")
|
|
self.assertIn("thread_renamed",
|
|
self._audit_actions(self.project["id"]))
|
|
calls = self._stub_calls("thread-rename")
|
|
self.assertEqual(len(calls), 1)
|
|
self.assertIn("new name", calls[0]["argv"])
|
|
|
|
def test_rename_validation(self):
|
|
task = self._create_task(executor="local")
|
|
self._set_status(task["id"], "completed")
|
|
status, _body = self._req(
|
|
"POST", "/tasks/%s/thread/rename" % task["id"],
|
|
{}, self.owner_key)
|
|
self.assertEqual(status, 400)
|
|
status, _body = self._req(
|
|
"POST", "/tasks/%s/thread/rename" % task["id"],
|
|
{"name": " "}, self.owner_key)
|
|
self.assertEqual(status, 400)
|
|
|
|
def test_archive_unarchive_compact_ok_and_audit(self):
|
|
task = self._create_task(executor="local")
|
|
self._set_status(task["id"], "failed")
|
|
for action in ("archive", "unarchive", "compact"):
|
|
status, body = self._req(
|
|
"POST", "/tasks/%s/thread/%s" % (task["id"], action),
|
|
{}, self.owner_key)
|
|
self.assertEqual(status, 200, body)
|
|
actions = self._audit_actions(self.project["id"])
|
|
self.assertIn("thread_archived", actions)
|
|
self.assertIn("thread_unarchived", actions)
|
|
self.assertIn("thread_compacted", actions)
|
|
|
|
def test_cli_failure_passthrough_502(self):
|
|
self._write_scenario(
|
|
{"fail_cmd": "thread-list", "fail_error": "list boom"})
|
|
task = self._create_task(executor="local")
|
|
status, body = self._req(
|
|
"GET", "/tasks/%s/threads" % task["id"], None, self.owner_key)
|
|
self.assertEqual(status, 502, body)
|
|
self.assertEqual(body["error"], "list boom")
|
|
self._write_scenario(
|
|
{"fail_cmd": "thread-archive", "fail_error": "arch boom"})
|
|
self._set_status(task["id"], "completed")
|
|
status, body = self._req(
|
|
"POST", "/tasks/%s/thread/archive" % task["id"], {},
|
|
self.owner_key)
|
|
self.assertEqual(status, 502, body)
|
|
self.assertEqual(body["error"], "arch boom")
|
|
|
|
def test_cli_bad_json_502(self):
|
|
self._write_scenario(
|
|
{"raw_stdout_for": "thread-list", "raw_stdout": "not json{"})
|
|
task = self._create_task(executor="local")
|
|
status, body = self._req(
|
|
"GET", "/tasks/%s/threads" % task["id"], None, self.owner_key)
|
|
self.assertEqual(status, 502, body)
|
|
|
|
def test_host_py_missing_502(self):
|
|
os.environ["CODEX_HOST_PY"] = os.path.join(
|
|
self.tmp.name, "does-not-exist.py")
|
|
task = self._create_task(executor="local")
|
|
status, body = self._req(
|
|
"GET", "/tasks/%s/threads" % task["id"], None, self.owner_key)
|
|
self.assertEqual(status, 502, body)
|
|
self.assertIn("host.py", body["error"])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|