From 3d2df7dfe698dc621cc84810ee556a493eeeac60 Mon Sep 17 00:00:00 2001 From: host Date: Sat, 31 Oct 2026 08:25:00 +0000 Subject: [PATCH] =?UTF-8?q?coverage85-step2b:=20server/worker=20=E5=89=A9?= =?UTF-8?q?=E4=BD=99=E7=BC=BA=E5=8F=A3=E7=9C=9F=E5=AE=9E=E8=A1=8C=E4=B8=BA?= =?UTF-8?q?=E6=B5=8B=E8=AF=95=EF=BC=88+24=20=E9=A1=B9=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - server：_dispatch_project（项目 PATCH repo_url、成员邀请/改角 色/删除与 404/403/409 族）、_create_task 校验 400 族与合法变体 （goal/协调员/plan/环境）、trace/agents/trace-stream（终态 SSE snapshot+done）、GitHub webhook 去重与坏体。 - worker：_gateway_context（xcmapi/缺 origin/缺属主/假 sub2api key 与协调员父属主回查）、_ensure_gateway 缓存、_docker_task_ context（代理/租户网络失败/搜索透传）、_deliver_marker_tools （守卫空跑、创建派生、goal 转换、轮询快照）、协调员孤儿扫尾、 run_one/run_all/run_loop/main 空库路径。 生产代码零改动。 --- cloud/tests/test_coverage85_server2.py | 420 +++++++++++++++++++++++++ cloud/tests/test_coverage85_worker2.py | 387 +++++++++++++++++++++++ 2 files changed, 807 insertions(+) create mode 100644 cloud/tests/test_coverage85_server2.py create mode 100644 cloud/tests/test_coverage85_worker2.py diff --git a/cloud/tests/test_coverage85_server2.py b/cloud/tests/test_coverage85_server2.py new file mode 100644 index 0000000..0e5e0c0 --- /dev/null +++ b/cloud/tests/test_coverage85_server2.py @@ -0,0 +1,420 @@ +"""覆盖率 78%->85% 第 2 步：cloud/server.py 剩余缺口的真实 HTTP 测试。 + +沿用 test_server.py 的 create_server + http.client 模式：每个测试 +用独立临时 SQLite 与临时端口，不跑模型、不联网。重点： +_dispatch_project（项目 PATCH / 成员增删改查）、_create_task 的 +校验分支与合法变体、任务 trace / trace-stream、GitHub webhook +去重与坏体分支。 +""" + +import hashlib +import hmac +import http.client +import json +import os +import sqlite3 +import sys +import tempfile +import threading +import unittest +from pathlib import Path +from unittest import mock + +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 + + +class Server2TestCase(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._old_admins = os.environ.get("CODEX_GLOBAL_ADMINS") + os.environ["CODEX_GLOBAL_ADMINS"] = self.owner_id + self.viewer_id, self.viewer_key = self.auth.create_user("viewer") + self.outsider_id, self.outsider_key = self.auth.create_user( + "outsider") + self.server = create_server(self.db_path, "127.0.0.1", 0) + 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.auth.set_member( + self.project["id"], self.owner_id, self.viewer_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 + self.tmp.cleanup() + if self._old_admins is None: + os.environ.pop("CODEX_GLOBAL_ADMINS", None) + else: + os.environ["CODEX_GLOBAL_ADMINS"] = self._old_admins + + def _req(self, method, path, body=None, key=None, headers=None, + raw_body=None): + conn = http.client.HTTPConnection( + "127.0.0.1", self.port, timeout=10) + hdrs = dict(headers or {}) + if raw_body is not None: + payload = raw_body + elif body is not None: + payload = json.dumps(body).encode("utf-8") + hdrs.setdefault("Content-Type", "application/json") + else: + payload = None + 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, key, **extra): + body = {"project_id": self.project["id"], "prompt": "做一个加法函数"} + body.update(extra) + status, task = self._req("POST", "/tasks", body, key) + self.assertEqual(status, 201, task) + return task + + def _task_count(self): + conn = sqlite3.connect(self.db_path) + try: + return conn.execute("SELECT COUNT(*) FROM tasks").fetchone()[0] + finally: + conn.close() + + def _members_path(self): + return "/projects/%s/members" % self.project["id"] + + # -- _dispatch_project：项目 PATCH -------------------------------- + def test_project_patch_repo_url(self): + pid = self.project["id"] + status, body = self._req( + "PATCH", "/projects/" + pid, + {"repo_url": "https://github.com/org/repo.git"}, self.owner_key) + self.assertEqual(status, 200, body) + self.assertEqual(body["project"]["repo_url"], + "https://github.com/org/repo.git") + status, body = self._req( + "PATCH", "/projects/" + pid, {"repo_url": ""}, self.owner_key) + self.assertEqual(status, 200) + self.assertIsNone(body["project"]["repo_url"]) + status, body = self._req( + "PATCH", "/projects/" + pid, {}, self.owner_key) + self.assertEqual(status, 400) + self.assertEqual(body["error"], "repo_url required") + status, body = self._req( + "PATCH", "/projects/" + pid, {"repo_url": 123}, self.owner_key) + self.assertEqual(status, 400) + status, body = self._req( + "PATCH", "/projects/" + pid, + {"repo_url": "https://user:pass@github.com/org/repo.git"}, + self.owner_key) + self.assertEqual(status, 400) + status, body = self._req( + "PATCH", "/projects/" + pid, + {"repo_url": "https://github.com/org/repo.git"}, + self.viewer_key) + self.assertEqual(status, 403) + status, body = self._req( + "PATCH", "/projects/nope", + {"repo_url": "https://github.com/org/repo.git"}, + self.owner_key) + self.assertEqual(status, 404) + + # -- _dispatch_project：成员 -------------------------------------- + def test_project_members_flow(self): + path = self._members_path() + status, body = self._req("GET", path, key=self.owner_key) + self.assertEqual(status, 200) + roles = {m["username"]: m["role"] for m in body["members"]} + self.assertEqual(roles, {"owner": "owner", "viewer": "viewer"}) + status, body = self._req("GET", path, key=self.outsider_key) + self.assertEqual(status, 403) + # 邀请：按 username、默认角色 contributor。 + status, body = self._req( + "POST", path, {"username": "outsider"}, self.owner_key) + self.assertEqual(status, 201, body) + self.assertEqual(body["role"], "contributor") + self.assertEqual(body["user_id"], self.outsider_id) + # 外人已是成员：再邀 409；邀不存在的用户 404；坏角色 400； + # 无目标 400；viewer 邀人 403。 + status, _ = self._req( + "POST", path, {"username": "outsider"}, self.owner_key) + self.assertEqual(status, 409) + status, body = self._req( + "POST", path, {"username": "ghost"}, self.owner_key) + self.assertEqual(status, 404) + status, body = self._req( + "POST", path, {"username": "viewer", "role": "super"}, + self.owner_key) + self.assertEqual(status, 400) + status, body = self._req("POST", path, {}, self.owner_key) + self.assertEqual(status, 400) + status, _ = self._req( + "POST", path, {"username": "viewer"}, self.viewer_key) + self.assertEqual(status, 403) + # 新成员立刻能建任务（角色真实生效）。 + self._create_task(self.outsider_key) + # 改角色：viewer -> contributor，之后 viewer 也能建任务。 + status, body = self._req( + "PATCH", path + "/" + self.viewer_id, + {"role": "contributor"}, self.owner_key) + self.assertEqual(status, 200) + self.assertEqual(body["role"], "contributor") + self._create_task(self.viewer_key) + # 唯一 owner 降级 409；删非成员 404；viewer 改角色 403。 + status, body = self._req( + "PATCH", path + "/" + self.owner_id, {"role": "viewer"}, + self.owner_key) + self.assertEqual(status, 409) + status, _ = self._req( + "PATCH", path + "/nobody", {"role": "viewer"}, self.owner_key) + self.assertEqual(status, 404) + status, _ = self._req( + "PATCH", path + "/" + self.viewer_id, {"role": "viewer"}, + self.viewer_key) + self.assertEqual(status, 403) + # 删成员：删后读该项目任务 403、建任务 403；删唯一 owner 409。 + status, body = self._req( + "DELETE", path + "/" + self.outsider_id, key=self.owner_key) + self.assertEqual(status, 200) + task = self._create_task(self.owner_key) + status, _ = self._req( + "GET", "/tasks/" + task["id"], key=self.outsider_key) + self.assertEqual(status, 403) + status, _ = self._req( + "POST", "/tasks", + {"project_id": self.project["id"], "prompt": "x"}, + self.outsider_key) + self.assertEqual(status, 403) + status, _ = self._req( + "DELETE", path + "/" + self.owner_id, key=self.owner_key) + self.assertEqual(status, 409) + status, _ = self._req( + "DELETE", path + "/nobody", key=self.owner_key) + self.assertEqual(status, 404) + status, _ = self._req("PUT", path, {}, self.owner_key) + self.assertEqual(status, 404) + status, _ = self._req( + "GET", path + "/" + self.viewer_id, key=self.owner_key) + self.assertEqual(status, 404) + status, _ = self._req("GET", "/projects/nope/members", + key=self.owner_key) + self.assertEqual(status, 404) + + # -- _create_task：校验分支 --------------------------------------- + def test_create_task_validation_errors(self): + pid = self.project["id"] + cases = [ + ({"prompt": "x"}, "project_id required"), + ({"project_id": pid}, "prompt required"), + ({"project_id": pid, "prompt": "/goal"}, + "goal 目标不能为空"), + ({"project_id": pid, "prompt": "x", "depends_on": {"a": 1}}, + "depends_on must be a list"), + ({"project_id": pid, "prompt": "x", "priority": "abc"}, + "priority must be an integer"), + ({"project_id": pid, "prompt": "x", "executor": "bogus"}, + "executor must be local, docker or cloud"), + ({"project_id": pid, "prompt": "x", "model": " "}, + "model must be a non-empty string"), + ({"project_id": pid, "prompt": "x", "effort": "ultra"}, + "effort must be low, medium, high, xhigh or max"), + ({"project_id": pid, "prompt": "x", "approval": "sometimes"}, + "approval must be ask, auto or full"), + ({"project_id": pid, "prompt": "x", "child_model": " "}, + "child_model must be a non-empty string"), + ({"project_id": pid, "prompt": "x", "child_effort": "ultra"}, + "child_effort must be low, medium, high, xhigh or max"), + ({"project_id": pid, "prompt": "x", "plan": True, + "approval": "full"}, "plan 模式与 approval=full 冲突"), + ({"project_id": pid, "prompt": "x", "goal_mode": True, + "verify_command": "make test"}, + "goal 模式不需要验收命令"), + ({"project_id": pid, "prompt": "x", "goal_mode": True, + "plan": True}, "plan 模式与 goal 模式冲突"), + ({"project_id": pid, "prompt": "x", "goal_max_rounds": 0}, + "goal_max_rounds must be >= 1"), + ({"project_id": pid, "prompt": "x", "goal_max_rounds": 101}, + "goal_max_rounds 超过上限 100 轮"), + ({"project_id": pid, "prompt": "x", "goal_max_rounds": "abc"}, + "goal_max_rounds must be an integer"), + ({"project_id": pid, "prompt": "x", "goal": 123}, + "goal must be a string"), + ({"project_id": pid, "prompt": "x", "verify_command": 123}, + "verify_command must be a string"), + ({"project_id": pid, "prompt": "x", "plan": True, + "verify_command": "make test"}, + "plan 模式与 goal 模式冲突"), + ({"project_id": pid, "prompt": "x", + "coordinator_parts": "not-a-list"}, + "coordinator_parts must be a list"), + ({"project_id": pid, "prompt": "x", + "coordinator_parts": []}, + "coordinator_parts must not be empty"), + ({"project_id": pid, "prompt": "x", + "coordinator_parts": [{"name": "a", "prompt": "do a"}], + "goal_mode": True}, + "协调员任务不跑模型回合"), + ({"project_id": pid, "prompt": "x", "idempotency_key": 123}, + "idempotency_key must be a string"), + ({"project_id": pid, "prompt": "x", "environment_id": " "}, + "environment_id must be a non-empty string"), + ] + for body, want_err in cases: + with self.subTest(body=body): + status, resp = self._req("POST", "/tasks", body, + self.owner_key) + self.assertEqual(status, 400, resp) + self.assertIn(want_err, resp["error"]) + # environment 不存在走 404（与上面 400 族不同）。 + status, resp = self._req( + "POST", "/tasks", + {"project_id": pid, "prompt": "x", "environment_id": "nope"}, + self.owner_key) + self.assertEqual(status, 404) + self.assertEqual(resp["error"], "environment not found") + self.assertEqual(self._task_count(), 0) + # 坏 JSON 体也 400。 + status, resp = self._req("POST", "/tasks", key=self.owner_key, + raw_body=b"{not json") + self.assertEqual(status, 400) + + def test_create_task_valid_variants(self): + pid = self.project["id"] + task = self._create_task(self.owner_key, depends_on="dep-1") + self.assertEqual(task["depends_on"], ["dep-1"]) + task = self._create_task(self.owner_key, prompt="/goal 收邮件") + self.assertEqual(task["goal_mode"], 1) + self.assertEqual(task["goal"], "收邮件") + task = self._create_task( + self.owner_key, + coordinator_parts=[{"name": "a", "prompt": "do a"}, + {"name": "b", "prompt": "do b"}]) + self.assertTrue(task["coordinator_parts"]) + task = self._create_task(self.owner_key, plan=True, + approval="ask") + self.assertEqual(task["plan"], 1) + task = self._create_task(self.owner_key, executor="local", + effort="high", model=" m1 ", + child_model="cm", child_effort="low") + self.assertEqual(task["model"], "m1") + self.assertEqual(task["executor"], "local") + + # -- trace / trace-stream ----------------------------------------- + def test_task_trace_and_stream(self): + task = self._create_task(self.owner_key) + tid = task["id"] + status, body = self._req("GET", "/tasks/%s/trace" % tid, + key=self.owner_key) + self.assertEqual(status, 200, body) + self.assertEqual(body["task_id"], tid) + self.assertIn("entries", body) + status, body = self._req("GET", "/tasks/%s/trace" % tid, + key=self.outsider_key) + self.assertEqual(status, 403) + status, body = self._req("GET", "/tasks/nope/trace", + key=self.owner_key) + self.assertEqual(status, 404) + status, body = self._req("GET", "/tasks/%s/agents" % tid, + key=self.owner_key) + self.assertEqual(status, 200) + self.assertIn("agents", body) + # 终态任务的 stream 立刻以 done 收尾（不进 3600 秒轮询）。 + conn = sqlite3.connect(self.db_path) + try: + conn.execute("UPDATE tasks SET status = 'completed' " + "WHERE id = ?", (tid,)) + conn.commit() + finally: + conn.close() + conn = http.client.HTTPConnection( + "127.0.0.1", self.port, timeout=10) + try: + conn.request("GET", "/tasks/%s/trace/stream" % tid, + headers={"X-API-Key": self.owner_key}) + resp = conn.getresponse() + self.assertEqual(resp.status, 200) + self.assertIn("text/event-stream", + resp.getheader("Content-Type")) + raw = resp.read().decode("utf-8") + finally: + conn.close() + events = [json.loads(ln[6:]) for ln in raw.splitlines() + if ln.startswith("data: ")] + self.assertEqual(events[0]["type"], "snapshot") + self.assertEqual(events[-1]["type"], "done") + self.assertEqual(events[-1]["status"], "completed") + status, _ = self._req("GET", "/tasks/nope/trace/stream", + key=self.owner_key) + self.assertEqual(status, 404) + status, _ = self._req("GET", "/tasks/%s/trace/stream" % tid, + key=self.outsider_key) + self.assertEqual(status, 403) + + # -- GitHub webhook：去重与坏体 ------------------------------------ + def _github_post(self, payload, raw=None, delivery="d-1", + event="issue_comment"): + if raw is None: + raw = json.dumps(payload).encode("utf-8") + sig = "sha256=" + hmac.new( + b"test-secret", raw, hashlib.sha256).hexdigest() + return self._req( + "POST", "/webhooks/github", raw_body=raw, + headers={"Content-Type": "application/json", + "X-Hub-Signature-256": sig, + "X-GitHub-Event": event, + "X-GitHub-Delivery": delivery}) + + def test_github_webhook_duplicate_and_bad_bodies(self): + payload = { + "action": "created", + "comment": {"body": "请 /fix-review 处理", + "user": {"login": "alice"}}, + "repository": {"full_name": "org/repo"}, + "issue": {"number": 7}, + } + with mock.patch.dict( + os.environ, {"GITHUB_WEBHOOK_SECRET": "test-secret"}): + status, body = self._github_post(payload) + self.assertEqual(status, 200) + self.assertEqual(body["type"], "fix_review") + # 无匹配项目：不建任务，task_id 为 None。 + self.assertIsNone(body.get("task_id")) + status, body = self._github_post(payload) + self.assertEqual(status, 200) + self.assertTrue(body["duplicate"]) + self.assertEqual(body["delivery_id"], "d-1") + status, body = self._github_post( + payload, raw=b"{broken", delivery="d-2") + self.assertEqual(status, 400) + status, body = self._github_post( + payload, raw=b"[1, 2]", delivery="d-3") + self.assertEqual(status, 400) + + +if __name__ == "__main__": + unittest.main() diff --git a/cloud/tests/test_coverage85_worker2.py b/cloud/tests/test_coverage85_worker2.py new file mode 100644 index 0000000..d8d4524 --- /dev/null +++ b/cloud/tests/test_coverage85_worker2.py @@ -0,0 +1,387 @@ +"""覆盖率 78%->85% 第 2 步：cloud/worker.py 剩余缺口的真实行为测试。 + +只用临时 SQLite（server.init_db 建表）与假执行器/假网关： +_gateway_context 的 sub2api 分支用模块级 _ensure_sub2api_key 替身 +（假网关，绝不碰真实 sub2api/生产网关）；_docker_task_context +用假执行器；_deliver_marker_tools 用真实父子任务行；main/run_loop +用空库。绝不跑模型、不建生产任务。 +""" + +import io +import json +import os +import sqlite3 +import sys +import tempfile +import unittest +from contextlib import redirect_stdout +from pathlib import Path +from unittest import mock + +try: + from cloud.server import init_db + from cloud.worker import Worker, main as worker_main + import cloud.worker as worker_mod +except ImportError: + sys.path.insert(0, str(Path(__file__).resolve().parents[2])) + from cloud.server import init_db + from cloud.worker import Worker, main as worker_main + import cloud.worker as worker_mod + + +class _Env: + def __init__(self, **kw): + self.kw = kw + self.old = {} + + def __enter__(self): + for k, v in self.kw.items(): + self.old[k] = os.environ.get(k) + if v is None: + os.environ.pop(k, None) + else: + os.environ[k] = v + return self + + def __exit__(self, *exc): + for k, v in self.old.items(): + if v is None: + os.environ.pop(k, None) + else: + os.environ[k] = v + return False + + +def _insert_task(db_path, task_id, status="queued", **kw): + conn = sqlite3.connect(db_path) + try: + cols = {"id": task_id, "project_id": kw.pop("project_id", "proj"), + "prompt": kw.pop("prompt", "do it"), "status": status, + "created_by": kw.pop("created_by", "owner1"), + "created_at": kw.pop("created_at", 1000.0), + "updated_at": kw.pop("updated_at", 1000.0)} + cols.update(kw) + names = ", ".join(cols) + placeholders = ", ".join("?" * len(cols)) + conn.execute("INSERT INTO tasks(%s) VALUES(%s)" + % (names, placeholders), tuple(cols.values())) + conn.commit() + finally: + conn.close() + + +def _rows(db_path, sql, params=()): + conn = sqlite3.connect(db_path) + try: + return conn.execute(sql, params).fetchall() + finally: + conn.close() + + +class Worker2Base(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.addCleanup(self.tmp.cleanup) + self.db_path = os.path.join(self.tmp.name, "control.db") + init_db(self.db_path) + self.hb = os.path.join(self.tmp.name, "hb.json") + self.worker = Worker(self.db_path, "/fake/host.py", + env={"XCM_API_KEY": "test-placeholder"}, + heartbeat_file=self.hb, + worker_id="cov85-worker2") + + +class GatewayContextTest(Worker2Base): + def test_xcmapi_returns_none_none(self): + with _Env(CLOUD_GATEWAY="xcmapi", SUB2API_BASE_URL=None): + self.assertEqual( + self.worker._gateway_context({"created_by": "owner1"}), + (None, None)) + with _Env(CLOUD_GATEWAY=None, SUB2API_BASE_URL=None): + self.assertEqual( + self.worker._gateway_context({"created_by": "owner1"}), + (None, None)) + + def test_sub2api_missing_origin_raises(self): + with _Env(CLOUD_GATEWAY="sub2api", SUB2API_BASE_URL=None): + with self.assertRaises(RuntimeError) as ctx: + self.worker._gateway_context({"created_by": "owner1"}) + self.assertIn("SUB2API_BASE_URL", str(ctx.exception)) + + def test_sub2api_missing_owner_raises(self): + with _Env(CLOUD_GATEWAY="sub2api", + SUB2API_BASE_URL="https://gw.example.test"): + with self.assertRaises(RuntimeError) as ctx: + self.worker._gateway_context({}) + self.assertIn("created_by", str(ctx.exception)) + + def test_sub2api_fake_gateway_and_parent_owner(self): + _insert_task(self.db_path, "parent1", status="completed", + created_by="owner9") + calls = [] + with _Env(CLOUD_GATEWAY="sub2api", + SUB2API_BASE_URL="https://gw.example.test:8443"): + with mock.patch.object( + worker_mod, "_ensure_sub2api_key", + side_effect=lambda conn, owner: ( + calls.append(owner), "fake-key-123")[1]): + provider, env = self.worker._gateway_context( + {"created_by": "owner1"}) + self.assertEqual(provider, "sub2api") + self.assertEqual(env["SUB2API_API_KEY"], "fake-key-123") + self.assertEqual(env["SUB2API_BASE_URL"], + "https://gw.example.test:8443") + self.assertIn("gw.example.test", env["NO_PROXY"]) + self.assertEqual(env["NO_PROXY"], env["no_proxy"]) + self.assertEqual(calls, ["owner1"]) + # 协调员子任务无 created_by 时跟父任务属主走。 + provider, env = self.worker._gateway_context( + {"parent_task_id": "parent1"}) + self.assertEqual(provider, "sub2api") + self.assertEqual(calls, ["owner1", "owner9"]) + + def test_sub2api_default_model_env_and_missing_key(self): + with _Env(CLOUD_SUB2API_DEFAULT_MODEL="model-x"): + self.assertEqual( + self.worker._sub2api_default_model({}), "model-x") + with _Env(CLOUD_SUB2API_DEFAULT_MODEL=None, + SUB2API_BASE_URL=None): + with self.assertRaises(RuntimeError): + self.worker._sub2api_default_model({}) + + def test_ensure_gateway_caches(self): + task = {"created_by": "owner1"} + with _Env(CLOUD_GATEWAY="xcmapi"): + first = self.worker._ensure_gateway(task) + self.assertEqual(first, (None, None)) + self.assertIn("_gateway_provider", task) + second = self.worker._ensure_gateway(task) + self.assertEqual(second, first) + + +class FakeExecutor: + def __init__(self, gateway=None, network_ok=True): + self.gateway = gateway + self.network_ok = network_ok + self.networks = [] + + def ensure_network(self, name): + self.networks.append(name) + return self.network_ok + + def network_gateway(self, network): + return self.gateway + + +class DockerTaskContextTest(Worker2Base): + def test_plain_no_tenant_no_proxy(self): + ok, mounts, network, proxy, err = self.worker._docker_task_context( + {"project_id": "proj"}) + self.assertEqual((ok, mounts, network, proxy, err), + (True, None, None, None, None)) + + def test_egress_proxy_gateway(self): + w = Worker(self.db_path, "/fake/host.py", + env={"XCM_API_KEY": "test-placeholder"}, + heartbeat_file=self.hb, worker_id="cov85-w2", + egress_proxy_port=8888, + executor=FakeExecutor(gateway="10.0.0.1")) + ok, _, _, proxy, err = w._docker_task_context({"project_id": "p"}) + self.assertTrue(ok) + self.assertIsNone(err) + self.assertEqual(proxy["HTTP_PROXY"], "http://10.0.0.1:8888") + self.assertEqual(proxy["NO_PROXY"], "localhost,127.0.0.1") + w2 = Worker(self.db_path, "/fake/host.py", + env={"XCM_API_KEY": "test-placeholder"}, + heartbeat_file=self.hb, worker_id="cov85-w2", + egress_proxy_port=8888, + executor=FakeExecutor(gateway=None)) + ok, _, _, proxy, err = w2._docker_task_context({"project_id": "p"}) + self.assertFalse(ok) + self.assertIsNone(proxy) + self.assertIn("egress 代理网关解析失败", err) + + def test_tenant_network_failure_and_search_passthrough(self): + tenants = os.path.join(self.tmp.name, "tenants") + template_home = os.path.join(self.tmp.name, "template-home") + os.makedirs(template_home) + with open(os.path.join(template_home, "config.toml"), "w") as f: + f.write("[mcp_servers]\n") + template_repo = os.path.join(self.tmp.name, "template-repo") + os.makedirs(template_repo) + with open(os.path.join(template_repo, "sample.py"), "w") as f: + f.write("x = 1\n") + w = Worker(self.db_path, "/fake/host.py", + env={"XCM_API_KEY": "test-placeholder"}, + heartbeat_file=self.hb, worker_id="cov85-w2", + tenant_isolation=True, tenants_root=tenants, + tenant_template_home=template_home, + tenant_template_repo=template_repo, + executor=FakeExecutor(network_ok=False)) + ok, _, _, _, err = w._docker_task_context({"project_id": "proj"}) + self.assertFalse(ok) + self.assertIn("租户 Docker 网络创建失败", err) + w2 = Worker(self.db_path, "/fake/host.py", + env={"XCM_API_KEY": "test-placeholder"}, + heartbeat_file=self.hb, worker_id="cov85-w2", + executor=FakeExecutor()) + with _Env(SEARXNG_URL="http://search.example.test:8080"): + ok, _, _, proxy, err = w2._docker_task_context( + {"project_id": "proj"}) + self.assertTrue(ok) + self.assertEqual(proxy["SEARXNG_URL"], + "http://search.example.test:8080") + + def test_tenant_safe_name_errors(self): + with self.assertRaises(ValueError): + self.worker._tenant_safe_name(None) + with self.assertRaises(ValueError): + self.worker._tenant_safe_name("...") + self.assertEqual(self.worker._tenant_network("proj"), + "codex-proj") + + +class MarkerToolsTest(Worker2Base): + def _followups(self, task_id): + return _rows(self.db_path, + "SELECT body FROM task_followups WHERE task_id = ?", + (task_id,)) + + def test_guards_are_noops(self): + _insert_task(self.db_path, "parent1", status="queued") + parent = self.worker.get_task("parent1") + for result, patch in [ + (None, {}), + ({"status": "failed"}, {}), + ({"status": "completed", "task_creates": [ + {"prompt": "x", "name": "c", "model": None, + "effort": None, "executor": None}]}, + {"verify_command": "make test"}), + ({"status": "completed", "task_creates": [ + {"prompt": "x", "name": "c", "model": None, + "effort": None, "executor": None}]}, + {"goal_mode": 1}), + ({"status": "completed", "task_creates": [ + {"prompt": "x", "name": "c", "model": None, + "effort": None, "executor": None}]}, + {"coordinator_parts": [{"name": "a"}]}), + ({"status": "completed"}, {}), + ]: + with self.subTest(patch=patch): + task = dict(parent) + task.update(patch) + self.worker._deliver_marker_tools(task, result) + self.assertEqual(self._followups("parent1"), []) + self.assertEqual(_rows(self.db_path, + "SELECT COUNT(*) FROM tasks"), [(1,)]) + + def test_creates_child_and_report(self): + _insert_task(self.db_path, "parent1", status="completed") + parent = self.worker.get_task("parent1") + self.worker._deliver_marker_tools(parent, { + "status": "completed", + "task_creates": [{"prompt": "child do", "name": "c1", + "model": None, "effort": None, + "executor": None}]}) + children = _rows( + self.db_path, + "SELECT prompt, status, parent_task_id FROM tasks " + "WHERE parent_task_id = ?", ("parent1",)) + self.assertEqual(children, [("child do", "queued", "parent1")]) + followups = self._followups("parent1") + self.assertEqual(len(followups), 1) + self.assertIn("任务工具报告", followups[0][0]) + + def test_goal_convert_queues_goal_followup(self): + _insert_task(self.db_path, "parent1", status="completed") + parent = self.worker.get_task("parent1") + self.worker._deliver_marker_tools(parent, { + "status": "completed", + "goal_converts": [{"goal": "收齐周报"}]}) + followups = self._followups("parent1") + self.assertEqual(len(followups), 1) + self.assertIn("/goal", followups[0][0]) + self.assertIn("收齐周报", followups[0][0]) + + def test_poll_snapshot_report(self): + _insert_task(self.db_path, "parent1", status="completed") + _insert_task(self.db_path, "child1", status="completed") + parent = self.worker.get_task("parent1") + self.worker._deliver_marker_tools(parent, { + "status": "completed", + "task_polls": [{"task_id": "child1", "interval_seconds": 0.05, + "timeout_seconds": 5, "chars": 200}]}) + followups = self._followups("parent1") + self.assertEqual(len(followups), 1) + self.assertIn("child1", followups[0][0]) + + +class OrphanSweepTest(Worker2Base): + def test_sweep_marks_orphans_only(self): + _insert_task(self.db_path, "p-done", status="completed") + _insert_task(self.db_path, "p-run", status="running") + _insert_task(self.db_path, "c1", status="queued", + parent_task_id="p-done") + _insert_task(self.db_path, "c2", status="queued", + parent_task_id="p-run") + _insert_task(self.db_path, "c3", status="queued", + parent_task_id="p-missing") + _insert_task(self.db_path, "c4", status="failed", + parent_task_id="p-done") + marked = self.worker.sweep_orphaned_coordinator_children() + self.assertEqual(sorted(marked), + [("p-done", "c1"), ("p-missing", "c3")]) + statuses = dict(_rows( + self.db_path, "SELECT id, status FROM tasks")) + self.assertEqual(statuses["c1"], "interrupted") + self.assertEqual(statuses["c3"], "interrupted") + self.assertEqual(statuses["c2"], "queued") + self.assertEqual(statuses["c4"], "failed") + reasons = _rows( + self.db_path, + "SELECT interrupted_reason FROM tasks WHERE id = 'c1'") + self.assertEqual(reasons, [("coordinator_orphan",)]) + + +class RunLoopMainTest(Worker2Base): + def test_run_one_no_ready_and_budget(self): + result = self.worker.run_one() + self.assertEqual(result, {"dispatched": False, + "reason": "no_ready_task"}) + self.assertEqual(self.worker.run_all(), []) + tight = Worker(self.db_path, "/fake/host.py", + env={"XCM_API_KEY": "test-placeholder"}, + heartbeat_file=self.hb, worker_id="cov85-w2", + max_cost_usd=0) + result = tight.run_one() + self.assertEqual(result["dispatched"], False) + self.assertEqual(result["reason"], "budget_exceeded") + + def test_run_loop_max_seconds_zero(self): + results = self.worker.run_loop(max_seconds=0) + self.assertEqual(results, []) + self.assertTrue(os.path.exists(self.hb)) + + def test_main_single_and_loop_and_bad_executor(self): + out = io.StringIO() + with redirect_stdout(out): + worker_main(["--db", self.db_path, + "--host-py", "/fake/host.py", + "--heartbeat-file", self.hb, + "--worker-id", "cov85-main"]) + self.assertEqual(json.loads(out.getvalue()), []) + out = io.StringIO() + with redirect_stdout(out): + worker_main(["--db", self.db_path, + "--host-py", "/fake/host.py", + "--heartbeat-file", self.hb, + "--worker-id", "cov85-main", + "--loop", "--max-seconds", "0"]) + self.assertEqual(json.loads(out.getvalue()), []) + with self.assertRaises(SystemExit) as ctx: + worker_main(["--db", self.db_path, "--executor", "bogus"]) + self.assertEqual(ctx.exception.code, 2) + + +if __name__ == "__main__": + unittest.main()