diff --git a/cloud/auth.py b/cloud/auth.py index 48c50a6..97a44ce 100644 --- a/cloud/auth.py +++ b/cloud/auth.py @@ -23,7 +23,6 @@ approve_merge -> maintainer，manage_project -> owner（仅 owner）。 import hashlib import secrets -import sqlite3 from cloud.db import connect as _db_connect import time import uuid diff --git a/cloud/cost.py b/cloud/cost.py index 6b1fb24..43b6dc5 100644 --- a/cloud/cost.py +++ b/cloud/cost.py @@ -50,7 +50,6 @@ events only. import json import os -import sqlite3 from cloud.db import connect as _db_connect import time from datetime import datetime, timezone diff --git a/cloud/db.py b/cloud/db.py index af24827..1b3f286 100644 --- a/cloud/db.py +++ b/cloud/db.py @@ -59,6 +59,7 @@ memory.py 的 summaries 完全一致，因为本模块根本不建表）。 """ import os +import re as _re import sqlite3 import time from abc import ABC, abstractmethod @@ -427,8 +428,6 @@ _OR_REPLACE_PK = { _IDENTITY_TABLES = frozenset( {"summaries", "kb_docs", "cost_records", "audit_log"}) -import re as _re - _OR_REPLACE_RE = _re.compile( r"^\s*INSERT\s+OR\s+REPLACE\s+INTO\s+(\w+)\s*\(([^)]+)\)\s*" r"VALUES\s*\((.*)\)\s*$", diff --git a/cloud/github_app.py b/cloud/github_app.py index 92a4b12..33acbe2 100644 --- a/cloud/github_app.py +++ b/cloud/github_app.py @@ -30,6 +30,7 @@ env secret，App webhook 用库里 secret，验签时两个都试（过渡 注入 fake http 客户端做单测。 """ +import base64 as _base64 import hashlib import json import secrets @@ -607,8 +608,6 @@ def set_fix_review_task_id(conn, repo, pr_number, comment_id, task_id): #（"BEGIN RSA PRIVATE KEY"），openssl 3 默认生成的是 PKCS#8 #（"BEGIN PRIVATE KEY"），两者都支持。 -import base64 as _base64 - _SHA256_DIGESTINFO_PREFIX = bytes.fromhex( "3031300d060960864801650304020105000420") diff --git a/cloud/harness2_tasks.py b/cloud/harness2_tasks.py index e42c547..1e17f97 100644 --- a/cloud/harness2_tasks.py +++ b/cloud/harness2_tasks.py @@ -73,7 +73,7 @@ import urllib.request import uuid from dataclasses import dataclass, field from pathlib import Path -from typing import Protocol +from typing import Protocol, TypedDict from .harness2 import ( EditResult, @@ -598,6 +598,12 @@ class CodexProbeAdapter: def base_harness(self) -> Tree: return dict(self.base) + def validate_tree(self, harness: Tree) -> None: + """探针无项目特有不变式：no-op。harness2._adapter_validate + 本就对缺失该方法的 adapter 跳过，故 no-op 与跳过行为一致， + 补上只是让类满足 TaskAdapter 协议（mypy arg-type）。""" + return None + def execute(self, label: str, harness: Tree) -> Rollout: files = dict(self.source_files) files.update(_tree_to_files(harness)) @@ -820,6 +826,20 @@ class CodexReflector: return parse_experience(state.result_text or "") +class _SharedRoleArgs(TypedDict): + """editor/selector/reflector 三角色共用的构造参数：键与三者 + dataclass 字段一一对应、逐键类型精确。dict 字面量会先塌成 + dict[str, int | str | None] 再按位置报错（mypy arg-type）， + TypedDict 展开时才能按字段逐个校验，运行时仍是普通 dict。""" + + environment_id: str | None + executor: str | None + effort: str | None + approval: str | None + no_time_limit: bool + timeout_seconds: int + + def run_improvement_via_tasks( client: TaskClient, *, @@ -872,9 +892,11 @@ def run_improvement_via_tasks( approval=approval, no_time_limit=no_time_limit, timeout_seconds=timeout_seconds, domain=domain) - shared = dict(environment_id=environment_id, executor=executor, - effort=effort, approval=approval, no_time_limit=no_time_limit, - timeout_seconds=timeout_seconds) + shared: _SharedRoleArgs = { + "environment_id": environment_id, "executor": executor, + "effort": effort, "approval": approval, + "no_time_limit": no_time_limit, "timeout_seconds": timeout_seconds, + } editor = CodexEditor(client=client, project_id=project_id, run_tag=run_tag, model=_role_model("editor"), **shared) selector = CodexSelector(client=client, project_id=project_id, run_tag=run_tag, diff --git a/cloud/imagegen_mcp.py b/cloud/imagegen_mcp.py index e67b3ea..c6f8eb4 100644 --- a/cloud/imagegen_mcp.py +++ b/cloud/imagegen_mcp.py @@ -15,15 +15,23 @@ copying; keys are read from the MCP server process environment by imagegen.py, never from the tool arguments. """ +import importlib import json import os import sys +# cloud 包优先、租户 flat copy 回落（与 playwright_mcp 同款理由： +# 两次 from-import 会让 mypy 报 no-redef，importlib 属性绑定 +# 在两条路径下都是同一组名字；cloud 导入本身的 ImportError +# 同样回落，与原来 except 语义一致）。 try: - from cloud.imagegen import ImageGenError, edit, generate + _mcp_backend = importlib.import_module("cloud.imagegen") except ImportError: # flat copy in a tenant codex-home sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) - from imagegen import ImageGenError, edit, generate # noqa: E402 + _mcp_backend = importlib.import_module("imagegen") +ImageGenError = _mcp_backend.ImageGenError +edit = _mcp_backend.edit +generate = _mcp_backend.generate SERVER_NAME = "imagegen" SERVER_VERSION = "0.1.0" diff --git a/cloud/marker_tools.py b/cloud/marker_tools.py index ecb87fd..e161085 100644 --- a/cloud/marker_tools.py +++ b/cloud/marker_tools.py @@ -135,7 +135,13 @@ import re import time import uuid from datetime import datetime, timedelta, timezone +from typing import Any +# ZoneInfo 在极老 Python 上不存在：先声明为 Any 再 try import， +# 这样 except 分支的 None 赋值不再是“给 type 赋值”（mypy +# misc + assignment），运行时仍是“有则类、无则 None”， +# 下游 `if ZoneInfo is not None` 守卫语义不变。 +ZoneInfo: Any = None try: from zoneinfo import ZoneInfo except Exception: # pragma: no cover - 极老 Python 无 zoneinfo diff --git a/cloud/mcp_server.py b/cloud/mcp_server.py index 29fa221..9a45ae5 100644 --- a/cloud/mcp_server.py +++ b/cloud/mcp_server.py @@ -83,7 +83,9 @@ TOOLS = [ _tool("get_task_files", "列/读任务 worktree 里的文件；不传 path 列顶层，传相对路径读文件或列目录。", {"task_id": {"type": "string"}, "path": {"type": "string"}}, ["task_id"]), - _tool("get_task_tree", "取任务 worktree 的结构化目录树（只含名、类型、大小，不读文件内容）；不传 path 列 worktree 根，传相对路径列该子目录；max_depth 1..5 缺省 3，max_entries 1..500 缺省 200，截断时 truncated=True。", + _tool("get_task_tree", "取任务 worktree 的结构化目录树（只含名、类型、大小，" + "不读文件内容）；不传 path 列 worktree 根，" + "传相对路径列该子目录；max_depth 1..5 缺省 3，max_entries 1..500 缺省 200，截断时 truncated=True。", {"task_id": {"type": "string"}, "path": {"type": "string"}, "max_depth": {"type": "integer"}, @@ -98,7 +100,8 @@ TOOLS = [ {"task_id": {"type": "string"}}, ["task_id"]), _tool("goal_resume_task", "断点续跑一个 goal 任务（服务端状态机校验）。", {"task_id": {"type": "string"}}, ["task_id"]), - _tool("create_followup", "给进行中任务发后续消息；mode=queued 排队进队尾不打断本轮，direct_send 立即送达且本轮结束后优先开下一轮。", + _tool("create_followup", "给进行中任务发后续消息；" + "mode=queued 排队进队尾不打断本轮，direct_send 立即送达且本轮结束后优先开下一轮。", {"task_id": {"type": "string"}, "body": {"type": "string"}, "mode": {"type": "string"}}, ["task_id", "body"]), @@ -112,7 +115,8 @@ TOOLS = [ {"project_id": {"type": "string"}}, ["project_id"]), _tool("get_models", "取模型一览（价目与默认模型，服务端口径）。", {}, []), - _tool("list_project_memories", "列出某项目的项目级记忆（此前任务沉淀的经验/约定，新任务会自动注入；top_k 可选，默认 20）。", + _tool("list_project_memories", "列出某项目的项目级记忆（此前任务沉淀的经验/约定，" + "新任务会自动注入；top_k 可选，默认 20）。", {"project_id": {"type": "string"}, "top_k": {"type": "integer"}}, ["project_id"]), _tool("create_project_memory", "给某项目记一条项目级记忆（kind 与 content 必填）。", diff --git a/cloud/migrations.py b/cloud/migrations.py index 81a593a..db24df5 100644 --- a/cloud/migrations.py +++ b/cloud/migrations.py @@ -29,7 +29,6 @@ 本模块不读、不拼连接串、不接受命令行密码参数。 """ -import os import re import sqlite3 diff --git a/cloud/oauth.py b/cloud/oauth.py index 85de977..a295800 100644 --- a/cloud/oauth.py +++ b/cloud/oauth.py @@ -41,7 +41,6 @@ import json import os import secrets -import sqlite3 from cloud.db import connect as _db_connect import time import urllib.parse diff --git a/cloud/playwright_mcp.py b/cloud/playwright_mcp.py index bf9d41e..b9a8044 100644 --- a/cloud/playwright_mcp.py +++ b/cloud/playwright_mcp.py @@ -23,15 +23,23 @@ each new page gets a fresh Playwright context, so cookies are isolated between pages. """ +import importlib import json import os import sys +# cloud 包优先、租户 flat copy 回落：同一段代码在两种部署形态下 +# 都要跑。importlib 动态取属性而不是两次 from-import：两次 +# from-import 会让 mypy 报 no-redef（cloud.* 与顶层模块名不同）， +# 而属性绑定在两条路径下都是同一组名字；cloud 导入本身的 +# ImportError（含内层缺依赖）同样回落，与原来 except 语义一致。 try: - from cloud.playwright_browser import Browser, BrowserError + _mcp_backend = importlib.import_module("cloud.playwright_browser") except ImportError: # flat copy in a tenant codex-home sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) - from playwright_browser import Browser, BrowserError # noqa: E402 + _mcp_backend = importlib.import_module("playwright_browser") +Browser = _mcp_backend.Browser +BrowserError = _mcp_backend.BrowserError SERVER_NAME = "playwright" SERVER_VERSION = "0.1.0" diff --git a/cloud/server.py b/cloud/server.py index c669e18..3a91214 100644 --- a/cloud/server.py +++ b/cloud/server.py @@ -96,7 +96,6 @@ from cloud.wait_ops import build_pr_diagnostics as _wait_build_diag from cloud.wait_ops import create_handoff as _handoff_create from cloud.wait_ops import decode_wait_cursor as _wait_decode_cursor from cloud.wait_ops import encode_wait_cursor as _wait_encode_cursor -from cloud.wait_ops import ensure_handoffs_table as _handoff_ensure from cloud.wait_ops import poll_for_first_change as _wait_poll_first from cloud.wait_ops import significant_now as _wait_significant_now from cloud.wait_ops import get_handoff as _handoff_get @@ -174,7 +173,6 @@ from cloud.sub2api import ensure_gateway_keys_table as _ensure_gateway_keys from cloud.sub2api import ensure_user_gateway_key as _ensure_sub2api_key from cloud.sub2api import gateway_name as _gateway_name from cloud.sub2api import gateway_origin as _gateway_origin -from cloud.sub2api import get_user_gateway_key as _get_sub2api_key from cloud.sub2api import list_gateway_models as _list_sub2api_models from cloud.sub2api import rotate_user_gateway_key as _rotate_sub2api_key from cloud.sub2api import validate_or_rotate_gateway_key as _validate_sub2api_key @@ -1769,14 +1767,11 @@ def _auth_store(db_path): pass -def _row_get(row, name): - """sqlite Row 的容错取值：老库行缺列时返回 None 不炸。""" - try: - return row[name] - except (KeyError, IndexError, TypeError): - return None - - +# 注意：此处曾有一份与 cloud.peer_messages.row_get 逐行相同的本地 +# _row_get（try return row[name] except KeyError/IndexError/TypeError +# -> None），它遮蔽了文件顶部的同名 import（F811 + mypy no-redef）。 +# 两份实现语义完全相同，删本地定义后全部调用点解析到 import 的 +# 同一份实现，行为不变；见 test_server_row_get_unified 回归测试。 def _task_to_dict(row, attachments=None, total_tokens=None): depends_on = row["depends_on"] try: diff --git a/cloud/skill_seed.py b/cloud/skill_seed.py index 97c37ae..eac78ab 100644 --- a/cloud/skill_seed.py +++ b/cloud/skill_seed.py @@ -42,7 +42,6 @@ idempotently, without clobbering tenant edits, and that skills/list sees them there. """ -import os import shutil from pathlib import Path diff --git a/cloud/sub2api.py b/cloud/sub2api.py index ffe1753..95c72bf 100644 --- a/cloud/sub2api.py +++ b/cloud/sub2api.py @@ -120,7 +120,7 @@ def get_user_gateway_key(conn, user_id): # 每次新建实例都要登录一次（实测约 111ms，占开通链路约 1/3）， # 共享后同一进程内第二次起免登录。401/刷新失败会回落到重登录， # 密码轮换后下一次登录即用新密码刷新缓存，不会卡死。 -_SHARED_ADMIN = {} +_SHARED_ADMIN: dict[tuple[str, str], tuple[str | None, str | None, float]] = {} _SHARED_ADMIN_LOCK = threading.RLock() @@ -201,7 +201,7 @@ def _call_with_login_retry(fn): # Redis 节流，控制面目前没有 Redis 客户端，未做）。 _LOGIN_BUDGET_PER_MINUTE = 15 _LOGIN_WINDOW_SECONDS = 60.0 -_login_slots = collections.deque() +_login_slots: collections.deque[float] = collections.deque() _login_slots_lock = threading.Lock() diff --git a/cloud/tests/test_automations.py b/cloud/tests/test_automations.py index fb1cb23..0c4f2b4 100644 --- a/cloud/tests/test_automations.py +++ b/cloud/tests/test_automations.py @@ -9,7 +9,6 @@ import os import sqlite3 import sys import tempfile -import time import unittest from pathlib import Path diff --git a/cloud/tests/test_chatgpt_codex_skills.py b/cloud/tests/test_chatgpt_codex_skills.py index a3334a9..ad1a1ef 100644 --- a/cloud/tests/test_chatgpt_codex_skills.py +++ b/cloud/tests/test_chatgpt_codex_skills.py @@ -5,7 +5,6 @@ frontmatter、名字与目录一致、正文非空、不含密钥占位。 """ -import os import re import unittest from pathlib import Path diff --git a/cloud/tests/test_cloud_dispatch_pr.py b/cloud/tests/test_cloud_dispatch_pr.py index 1f62d2e..6148afc 100644 --- a/cloud/tests/test_cloud_dispatch_pr.py +++ b/cloud/tests/test_cloud_dispatch_pr.py @@ -16,7 +16,6 @@ import time import unittest from pathlib import Path from types import SimpleNamespace -from unittest import mock try: from cloud.auth import AuthStore @@ -25,7 +24,6 @@ try: from cloud.server import create_server, init_db from cloud.worker import Worker from cloud.workers import ( - heartbeat_worker, list_workers, register_worker, ) @@ -37,7 +35,6 @@ except ImportError: from cloud.server import create_server, init_db from cloud.worker import Worker from cloud.workers import ( - heartbeat_worker, list_workers, register_worker, ) diff --git a/cloud/tests/test_docker_sandbox.py b/cloud/tests/test_docker_sandbox.py index 3322f09..906a8b2 100644 --- a/cloud/tests/test_docker_sandbox.py +++ b/cloud/tests/test_docker_sandbox.py @@ -169,7 +169,8 @@ class DockerCommandTestCase(unittest.TestCase): ex = self._executor(seen) dispatched = ex.dispatch_task( {"id": "t1", "prompt": "do it", "worktree": "/tmp/wt"}) - ran = ex.run_task( + # run_task 调用本身有副作用（seen 记命令）：只去绑定，不删调用。 + ex.run_task( {"id": "t1", "prompt": "again", "worktree": "/tmp/wt"}) self.assertEqual(dispatched["status"], "completed") self.assertEqual(dispatched["executor"], "docker") diff --git a/cloud/tests/test_fallback_retry.py b/cloud/tests/test_fallback_retry.py index ada9285..80b3c8f 100644 --- a/cloud/tests/test_fallback_retry.py +++ b/cloud/tests/test_fallback_retry.py @@ -22,13 +22,11 @@ from pathlib import Path from types import SimpleNamespace try: - from cloud.cost import CostStore from cloud.models import ModelRouter from cloud.server import init_db from cloud.worker import Worker except ImportError: # discover -s cloud/tests 时把项目根目录补进 sys.path sys.path.insert(0, str(Path(__file__).resolve().parents[2])) - from cloud.cost import CostStore from cloud.models import ModelRouter from cloud.server import init_db from cloud.worker import Worker diff --git a/cloud/tests/test_harness2.py b/cloud/tests/test_harness2.py index e94c808..8831b67 100644 --- a/cloud/tests/test_harness2.py +++ b/cloud/tests/test_harness2.py @@ -7,30 +7,26 @@ 账本：100.25, -20.10, 5.00 —— 带符号和 85.15，只加正数是 105.25。 """ -import csv import subprocess import sys import tempfile import unittest from pathlib import Path -from types import SimpleNamespace try: - from cloud import harness2 from cloud.harness2 import ( - Library, Rollout, EditResult, Budget, build_context, validate, + Library, Rollout, EditResult, validate, read_tree, write_tree, is_allowed, normalize_rel, autotrigger, - skill_names, diff_summary, diversity, improve_task, select_final, - assign_ids, has_experience, + diff_summary, diversity, improve_task, select_final, + assign_ids, ) except ImportError: # discover -s cloud/tests 时把项目根目录补进 sys.path sys.path.insert(0, str(Path(__file__).resolve().parents[2])) - from cloud import harness2 from cloud.harness2 import ( - Library, Rollout, EditResult, Budget, build_context, validate, + Library, Rollout, EditResult, validate, read_tree, write_tree, is_allowed, normalize_rel, autotrigger, - skill_names, diff_summary, diversity, improve_task, select_final, - assign_ids, has_experience, + diff_summary, diversity, improve_task, select_final, + assign_ids, ) BASE_SCRIPT = """import csv diff --git a/cloud/tests/test_harness2_tasks.py b/cloud/tests/test_harness2_tasks.py index 5b7fac4..54c0531 100644 --- a/cloud/tests/test_harness2_tasks.py +++ b/cloud/tests/test_harness2_tasks.py @@ -6,7 +6,6 @@ 账本与 base 树同 test_harness2.py：带符号和 85.15，只加正数 105.25。 """ -import csv import json import subprocess import sys @@ -17,26 +16,24 @@ from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path try: - from cloud import harness2_tasks from cloud.harness2 import ( parse_choice, parse_deliverable_files, parse_experience, parse_harness_files, parse_no_change, parse_notes, parse_why, ) from cloud.harness2_tasks import ( - CodexEditor, CodexProbeAdapter, HttpTaskClient, TaskState, + CodexEditor, HttpTaskClient, TaskState, build_multipart, preflight_checks, run_improvement_via_tasks, run_sweep, run_task, unwrap_result_text, validate_run_tag, ) from cloud.uploads import parse_multipart, parse_paths_field except ImportError: # discover -s cloud/tests 时把项目根目录补进 sys.path sys.path.insert(0, str(Path(__file__).resolve().parents[2])) - from cloud import harness2_tasks from cloud.harness2 import ( parse_choice, parse_deliverable_files, parse_experience, parse_harness_files, parse_no_change, parse_notes, parse_why, ) from cloud.harness2_tasks import ( - CodexEditor, CodexProbeAdapter, HttpTaskClient, TaskState, + CodexEditor, HttpTaskClient, TaskState, build_multipart, preflight_checks, run_improvement_via_tasks, run_sweep, run_task, unwrap_result_text, validate_run_tag, ) diff --git a/cloud/tests/test_mcp_coverage.py b/cloud/tests/test_mcp_coverage.py new file mode 100644 index 0000000..76c2fc8 --- /dev/null +++ b/cloud/tests/test_mcp_coverage.py @@ -0,0 +1,355 @@ +"""MCP 三模块覆盖补点（CI 第二阶段覆盖率）。 + +- websearch/imagegen：handle_request 全分支（协议方法、工具 + 分发、未知工具、工具异常、非对象请求、通知）+ main() 的 + stdio 循环（坏 JSON、通知不回包），工具执行用注入的 + fake，不联网、不调真实模型/网关。 +- playwright：全部 14 个工具分发 + 未知工具 + 工具异常 + + 协议方法 + _handle_line 的解析错误与 handler 异常分支。 + fake browser 只返回标记 dict，断言分发到了正确方法。 +- flat-copy：三个模块 + 后端文件拷到隔离临时目录，在 + PYTHONPATH 只含该目录的子进程里 import——cloud 包不可见， + 必须走 flat-copy 回落；断言后端类的 __module__ 是顶层 + 模块名（不是 cloud.*），即回落真实发生（importlib 重构 + 的回归证据，见 F401/mypy no-redef 一节）。 +""" + +import io +import json +import os +import shutil +import subprocess +import sys +import tempfile +import unittest +from pathlib import Path +from unittest import mock + +try: + from cloud.websearch_mcp import handle_request as ws_handle + from cloud.websearch_mcp import main as ws_main + from cloud.imagegen_mcp import handle_request as ig_handle + from cloud.imagegen_mcp import main as ig_main + from cloud import playwright_mcp as pw_mod + from cloud.playwright_mcp import handle_request as pw_handle + from cloud.playwright_mcp import _handle_line +except ImportError: + sys.path.insert(0, str(Path(__file__).resolve().parents[2])) + from cloud.websearch_mcp import handle_request as ws_handle + from cloud.websearch_mcp import main as ws_main + from cloud.imagegen_mcp import handle_request as ig_handle + from cloud.imagegen_mcp import main as ig_main + from cloud import playwright_mcp as pw_mod + from cloud.playwright_mcp import handle_request as pw_handle + from cloud.playwright_mcp import _handle_line + + +def _call(handle, method, req_id=1, name=None, args=None): + req = {"jsonrpc": "2.0", "id": req_id, "method": method} + if name is not None: + req["params"] = {"name": name, "arguments": args or {}} + return handle(req) + + +class WebSearchMcpTest(unittest.TestCase): + def test_protocol_methods(self): + self.assertIsNone(ws_handle({"method": "notifications/initialized"})) + self.assertIsNone(ws_handle({"method": "ping"})) + ping = _call(ws_handle, "ping") + self.assertEqual(ping["result"], {}) + init = {"jsonrpc": "2.0", "id": 1, "method": "initialize", + "params": {"protocolVersion": "2025-01-01"}} + resp = ws_handle(init) + self.assertEqual(resp["result"]["protocolVersion"], "2025-01-01") + self.assertEqual(resp["result"]["serverInfo"]["name"], "websearch") + self.assertIsNone(ws_handle({"method": "initialize"})) + listed = _call(ws_handle, "tools/list") + self.assertIn("web_search", + [t["name"] for t in listed["result"]["tools"]]) + + def test_tools_call_and_errors(self): + fake_search = lambda q, limit=5, provider="auto": { # noqa: E731 + "results": [{"title": q}]} + # 工具执行一律用注入的 fake，不调真实 search（不联网）。 + resp = ws_handle({"jsonrpc": "2.0", "id": 2, "method": "tools/call", + "params": {"name": "web_search", + "arguments": {"query": "hello"}}}, + search_fn=fake_search) + self.assertFalse(resp["result"]["isError"]) + self.assertIn("hello", resp["result"]["content"][0]["text"]) + resp = ws_handle({"jsonrpc": "2.0", "id": 3, "method": "tools/call", + "params": {"name": "web_fetch", + "arguments": {"url": "https://x/"}}}, + fetch_fn=lambda u, char_limit=15000: {"text": "T"}) + self.assertIn("T", resp["result"]["content"][0]["text"]) + unknown = _call(ws_handle, "tools/call", name="nope") + self.assertEqual(unknown["error"]["code"], -32602) + boom = ws_handle({"jsonrpc": "2.0", "id": 4, "method": "tools/call", + "params": {"name": "web_search", + "arguments": {"query": "q"}}}, + search_fn=lambda *a, **k: (_ for _ in ()).throw( + RuntimeError("boom"))) + self.assertTrue(boom["result"]["isError"]) + self.assertIn("boom", boom["result"]["content"][0]["text"]) + self.assertEqual(ws_handle("nope")["error"]["code"], -32700) + self.assertEqual( + _call(ws_handle, "bogus/method")["error"]["code"], -32601) + self.assertIsNone(ws_handle( + {"method": "tools/call", + "params": {"name": "web_search", "arguments": {}}})) + + def test_main_stdio_loop(self): + lines = ('{"bad json\n' + '{"jsonrpc": "2.0", "id": 1, "method": "initialize",' + ' "params": {}}\n' + '{"jsonrpc": "2.0", "id": 2, "method": "tools/list"}\n' + '{"jsonrpc": "2.0", "id": 3, "method": "ping"}\n' + '{"jsonrpc": "2.0", "method": "notifications/initialized"}\n' + '\n' + '{"jsonrpc": "2.0", "id": 4, "method": "bogus"}\n') + out = io.StringIO() + with mock.patch.object(sys, "stdin", io.StringIO(lines)), \ + mock.patch.object(sys, "stdout", out): + ws_main() + rows = [json.loads(x) for x in out.getvalue().splitlines()] + self.assertEqual(len(rows), 5) + self.assertEqual(rows[0]["error"]["code"], -32700) + self.assertEqual(rows[1]["result"]["serverInfo"]["name"], + "websearch") + self.assertEqual(rows[3]["result"], {}) + self.assertEqual(rows[4]["error"]["code"], -32601) + + +class ImageGenMcpTest(unittest.TestCase): + def test_tools_call_and_errors(self): + gen = lambda p, output_filename=None, size="1024x1024", model=None: { # noqa: E731 + "path": "/w/gen.png", "prompt": p} + edit = lambda p, inp, output_filename=None, size="1024x1024", model=None: { # noqa: E731 + "path": "/w/edit.png", "input": inp} + resp = ig_handle({"jsonrpc": "2.0", "id": 1, "method": "tools/call", + "params": {"name": "image_generate", + "arguments": {"prompt": "cat"}}}, + generate_fn=gen) + self.assertIn("gen.png", resp["result"]["content"][0]["text"]) + resp = ig_handle({"jsonrpc": "2.0", "id": 2, "method": "tools/call", + "params": {"name": "image_edit", + "arguments": {"prompt": "dog", + "input_filename": "in.png"}}}, + edit_fn=edit) + self.assertIn("in.png", resp["result"]["content"][0]["text"]) + unknown = _call(ig_handle, "tools/call", name="nope") + self.assertEqual(unknown["error"]["code"], -32602) + boom = ig_handle({"jsonrpc": "2.0", "id": 3, "method": "tools/call", + "params": {"name": "image_generate", + "arguments": {"prompt": "x"}}}, + generate_fn=lambda *a, **k: (_ for _ in ()).throw( + RuntimeError("gen boom"))) + self.assertTrue(boom["result"]["isError"]) + self.assertIn("gen boom", boom["result"]["content"][0]["text"]) + self.assertEqual(ig_handle(42)["error"]["code"], -32700) + self.assertIsNone(ig_handle({"method": "ping"})) + self.assertEqual(_call(ig_handle, "ping")["result"], {}) + + def test_main_stdio_loop(self): + lines = ('{"jsonrpc": "2.0", "id": 1, "method": "initialize",' + ' "params": {}}\n' + '{"jsonrpc": "2.0", "id": 2, "method": "tools/list"}\n' + 'not json\n') + out = io.StringIO() + with mock.patch.object(sys, "stdin", io.StringIO(lines)), \ + mock.patch.object(sys, "stdout", out): + ig_main() + rows = [json.loads(x) for x in out.getvalue().splitlines()] + self.assertEqual(len(rows), 3) + self.assertEqual(rows[0]["result"]["serverInfo"]["name"], + "imagegen") + self.assertEqual(rows[2]["error"]["code"], -32700) + + +class FakeBrowserFull: + def __init__(self): + self.calls = [] + + def navigate(self, url): + self.calls.append(("navigate", url)) + return {"tool": "navigate", "url": url} + + def snapshot(self): + self.calls.append(("snapshot",)) + return {"tool": "snapshot"} + + def click(self, selector): + self.calls.append(("click", selector)) + return {"tool": "click"} + + def fill(self, selector, text): + self.calls.append(("fill", selector, text)) + return {"tool": "fill"} + + def click_at(self, x, y): + self.calls.append(("click_at", x, y)) + return {"tool": "click_at"} + + def mouse_move(self, x, y): + self.calls.append(("mouse_move", x, y)) + return {"tool": "mouse_move"} + + def key_press(self, key): + self.calls.append(("key_press", key)) + return {"tool": "key_press"} + + def type_text(self, text): + self.calls.append(("type_text", text)) + return {"tool": "type_text"} + + def snapshot_refs(self): + self.calls.append(("snapshot_refs",)) + return {"tool": "snapshot_refs"} + + def click_ref(self, ref): + self.calls.append(("click_ref", ref)) + return {"tool": "click_ref"} + + def live_shot(self, name=None): + self.calls.append(("live_shot", name)) + return {"tool": "live_shot"} + + def evaluate(self, expression): + self.calls.append(("evaluate", expression)) + return {"tool": "evaluate"} + + def screenshot(self, name=None): + self.calls.append(("screenshot", name)) + return {"tool": "screenshot"} + + def close(self): + self.calls.append(("close",)) + return {"tool": "close"} + + +class PlaywrightMcpTest(unittest.TestCase): + def test_all_tools_dispatch(self): + cases = [ + ("browser_navigate", {"url": "https://x/"}, "navigate"), + ("browser_snapshot", {}, "snapshot"), + ("browser_click", {"selector": "#a"}, "click"), + ("browser_fill", {"selector": "#a", "text": "hi"}, "fill"), + ("browser_click_at", {"x": 1, "y": 2}, "click_at"), + ("browser_mouse_move", {"x": 1, "y": 2}, "mouse_move"), + ("browser_key_press", {"key": "Enter"}, "key_press"), + ("browser_type_text", {"text": "hi"}, "type_text"), + ("browser_snapshot_refs", {}, "snapshot_refs"), + ("browser_click_ref", {"ref": "e0"}, "click_ref"), + ("browser_live_shot", {}, "live_shot"), + ("browser_evaluate", {"expression": "1+1"}, "evaluate"), + ("browser_screenshot", {}, "screenshot"), + ("browser_close", {}, "close"), + ] + for i, (name, args, tool) in enumerate(cases): + fake = FakeBrowserFull() + resp = pw_handle( + {"jsonrpc": "2.0", "id": i + 1, "method": "tools/call", + "params": {"name": name, "arguments": args}}, + browser=fake) + self.assertFalse(resp["result"]["isError"], name) + self.assertIn(tool, resp["result"]["content"][0]["text"], name) + + def test_unknown_tool_and_tool_exception(self): + fake = FakeBrowserFull() + unknown = pw_handle( + {"jsonrpc": "2.0", "id": 1, "method": "tools/call", + "params": {"name": "browser_nope", "arguments": {}}}, + browser=fake) + self.assertEqual(unknown["error"]["code"], -32602) + + class Raising(FakeBrowserFull): + def navigate(self, url): + raise RuntimeError("nav boom") + + resp = pw_handle( + {"jsonrpc": "2.0", "id": 2, "method": "tools/call", + "params": {"name": "browser_navigate", + "arguments": {"url": "https://x/"}}}, + browser=Raising()) + self.assertTrue(resp["result"]["isError"]) + self.assertIn("nav boom", resp["result"]["content"][0]["text"]) + + def test_protocol_and_handle_line(self): + self.assertEqual(pw_handle("x")["error"]["code"], -32700) + self.assertIsNone(pw_handle({"method": "ping"})) + self.assertEqual(_call(pw_handle, "ping")["result"], {}) + out = io.StringIO() + with mock.patch.object(sys, "stdout", out): + _handle_line('{"jsonrpc": "2.0", "id": 9, "method": "ping"}', + FakeBrowserFull()) + _handle_line("{bad json", FakeBrowserFull()) + rows = [json.loads(x) for x in out.getvalue().splitlines()] + self.assertEqual(rows[0]["result"], {}) + self.assertEqual(rows[1]["error"]["code"], -32700) + with mock.patch.object(pw_mod, "handle_request", + side_effect=RuntimeError("handler boom")): + out2, err2 = io.StringIO(), io.StringIO() + with mock.patch.object(sys, "stdout", out2), \ + mock.patch.object(sys, "stderr", err2): + _handle_line('{"jsonrpc": "2.0", "id": 10, ' + '"method": "tools/list"}', + FakeBrowserFull()) + row = json.loads(out2.getvalue().strip()) + self.assertEqual(row["error"]["code"], -32603) + self.assertIn("handler boom", err2.getvalue()) + + +class FlatCopyTest(unittest.TestCase): + def _flat_copy(self, pairs): + td = tempfile.mkdtemp() + self.addCleanup(shutil.rmtree, td, True) + root = Path(__file__).resolve().parents[2] / "cloud" + for name in pairs: + shutil.copy(root / name, os.path.join(td, name)) + return td + + def _run_flat(self, td, module, check): + script = ("import %s as m; print(m.SERVER_NAME); print(check(m))" + % module) + env = {"PATH": os.environ.get("PATH", ""), "PYTHONPATH": td} + proc = subprocess.run( + [sys.executable, "-c", + script.replace("check(m)", check)], + cwd=td, env=env, capture_output=True, text=True, timeout=60) + self.assertEqual(proc.returncode, 0, proc.stderr) + return proc.stdout + + def test_playwright_flat_copy(self): + td = self._flat_copy(["playwright_mcp.py", + "playwright_browser.py"]) + out = self._run_flat( + td, "playwright_mcp", + "m.Browser.__module__ + '|' + " + "str(m.handle_request({'jsonrpc': '2.0', 'id': 1," + " 'method': 'tools/list'})['result']['tools'][0]['name'])") + self.assertIn("playwright", out) + self.assertIn("playwright_browser|browser_navigate", out) + + def test_websearch_flat_copy(self): + td = self._flat_copy(["websearch_mcp.py", "websearch.py"]) + out = self._run_flat( + td, "websearch_mcp", + "m.search.__module__ + '|' + " + "str(m.handle_request({'jsonrpc': '2.0', 'id': 1," + " 'method': 'ping'})['result'])") + self.assertIn("websearch", out) + self.assertIn("websearch|{}", out) + + def test_imagegen_flat_copy(self): + td = self._flat_copy(["imagegen_mcp.py", "imagegen.py"]) + out = self._run_flat( + td, "imagegen_mcp", + "m.generate.__module__ + '|' + " + "str(m.handle_request({'jsonrpc': '2.0', 'id': 1," + " 'method': 'ping'})['result'])") + self.assertIn("imagegen", out) + self.assertIn("imagegen|{}", out) + + +if __name__ == "__main__": + unittest.main() diff --git a/cloud/tests/test_migrate_cli.py b/cloud/tests/test_migrate_cli.py new file mode 100644 index 0000000..6301d73 --- /dev/null +++ b/cloud/tests/test_migrate_cli.py @@ -0,0 +1,75 @@ +"""cloud/migrate_sqlite_to_postgres.py CLI 包装层单测 +（CI 第二阶段覆盖率：基线 0%）。 + +只测包装层本身：参数解析、行数合计打印、missing 警告分支、 +unmigrated/skipped 报告行。真正的迁移逻辑在 +cloud/migrations.migrate_sqlite_to_postgres，另有 +test_p2_migrations / test_postgres_integration 覆盖； +本文件 patch 的是包装层调用的两个函数名，不碰迁移本身。 +""" + +import io +import sys +import unittest +from contextlib import redirect_stdout +from pathlib import Path +from unittest import mock + +try: + from cloud import migrate_sqlite_to_postgres as mig_cli +except ImportError: + sys.path.insert(0, str(Path(__file__).resolve().parents[2])) + from cloud import migrate_sqlite_to_postgres as mig_cli + + +def _coverage(**over): + base = {"missing_from_migration": [], + "unmigrated_no_pg_schema": ["legacy_x"], + "skipped_by_design": ["audit_log"]} + base.update(over) + return base + + +class MigrateCliTest(unittest.TestCase): + def test_totals_and_reports(self): + with mock.patch.object( + mig_cli, "migrate_sqlite_to_postgres", + return_value={"users": 2, "tasks": 0}), \ + mock.patch.object(mig_cli, "migration_coverage", + return_value=_coverage()): + out = io.StringIO() + with redirect_stdout(out): + rc = mig_cli.main(["--sqlite", "/tmp/x.db"]) + self.assertEqual(rc, 0) + text = out.getvalue() + self.assertIn("users: 2", text) + self.assertIn("tasks: 0", text) + self.assertIn("total inserted: 2", text) + self.assertNotIn("WARNING", text) + self.assertIn("unmigrated (no PG schema, NOT migrated): legacy_x", + text) + self.assertIn("skipped by design: audit_log", text) + + def test_missing_warning_branch(self): + with mock.patch.object( + mig_cli, "migrate_sqlite_to_postgres", + return_value={}), \ + mock.patch.object( + mig_cli, "migration_coverage", + return_value=_coverage( + missing_from_migration=["tasks"])): + out = io.StringIO() + with redirect_stdout(out): + rc = mig_cli.main(["--sqlite", "/tmp/x.db"]) + self.assertEqual(rc, 0) + self.assertIn("WARNING missing_from_migration: tasks", + out.getvalue()) + + def test_sqlite_arg_required(self): + with self.assertRaises(SystemExit) as ctx: + mig_cli.main([]) + self.assertEqual(ctx.exception.code, 2) + + +if __name__ == "__main__": + unittest.main() diff --git a/cloud/tests/test_oauth.py b/cloud/tests/test_oauth.py index 8e33e37..7e6ff5f 100644 --- a/cloud/tests/test_oauth.py +++ b/cloud/tests/test_oauth.py @@ -6,7 +6,6 @@ GitHub / Google 发任何请求；出现的 client_id / secret 哨兵值 """ import os -import sqlite3 import sys import tempfile import time diff --git a/cloud/tests/test_pr_jobs.py b/cloud/tests/test_pr_jobs.py index 0d77378..f6be650 100644 --- a/cloud/tests/test_pr_jobs.py +++ b/cloud/tests/test_pr_jobs.py @@ -19,7 +19,6 @@ import tempfile import threading import unittest from pathlib import Path -from types import SimpleNamespace try: from cloud.auth import AuthStore @@ -84,7 +83,8 @@ class PrJobsTableTestCase(unittest.TestCase): conn = _db_connect(self.db_path) try: job1 = enqueue_pr_job(conn, "t1", "p1", "host-a") - job2 = enqueue_pr_job(conn, "t2", "p1", "host-a") + # 第二条入队本身是测试前置（claimed2 要拿到 t2）：只去绑定，不删调用。 + enqueue_pr_job(conn, "t2", "p1", "host-a") self.assertIsNotNone(find_open_pr_job(conn, "t1")) claimed = claim_pr_job(conn, "host-a") self.assertEqual(claimed["task_id"], "t1") diff --git a/cloud/tests/test_prod_worker_timeout.py b/cloud/tests/test_prod_worker_timeout.py index 1f8cf1d..2db7f86 100644 --- a/cloud/tests/test_prod_worker_timeout.py +++ b/cloud/tests/test_prod_worker_timeout.py @@ -17,13 +17,11 @@ from unittest import mock try: from cloud import backup as backup_mod - from cloud.auth import AuthStore from cloud.sandbox import DockerExecutor from cloud.worker import Worker except ImportError: sys.path.insert(0, str(Path(__file__).resolve().parents[2])) from cloud import backup as backup_mod - from cloud.auth import AuthStore from cloud.sandbox import DockerExecutor from cloud.worker import Worker diff --git a/cloud/tests/test_server_goal_notify.py b/cloud/tests/test_server_goal_notify.py index 877686f..d5314ac 100644 --- a/cloud/tests/test_server_goal_notify.py +++ b/cloud/tests/test_server_goal_notify.py @@ -22,12 +22,12 @@ from types import SimpleNamespace try: from cloud.auth import AuthStore from cloud.push import create_subscription - from cloud.server import create_server, init_db + 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.push import create_subscription - from cloud.server import create_server, init_db + from cloud.server import create_server def _result(returncode=0, stdout="", stderr=""): diff --git a/cloud/tests/test_server_row_get.py b/cloud/tests/test_server_row_get.py new file mode 100644 index 0000000..6edff7e --- /dev/null +++ b/cloud/tests/test_server_row_get.py @@ -0,0 +1,44 @@ +"""server._row_get 统一回归测试（CI 第二阶段 F811/mypy no-redef）。 + +背景：cloud/server.py 顶部 import cloud.peer_messages.row_get +为 _row_get，文件后部又定义了一份逐行相同的本地 _row_get +（try return row[name] except KeyError/IndexError/TypeError +-> None），遮蔽 import。本轨删本地定义，全部调用点解析到 +import 的同一份实现。本测试钉住两点： + +- server._row_get 与 peer_messages.row_get 是同一对象； +- 缺列/空行/正常取值三路行为与原来本地定义一致。 +""" + +import sys +import unittest +from pathlib import Path + +try: + from cloud.peer_messages import row_get as peer_row_get + from cloud.server import _row_get +except ImportError: + sys.path.insert(0, str(Path(__file__).resolve().parents[2])) + from cloud.peer_messages import row_get as peer_row_get + from cloud.server import _row_get + + +class RowGetUnifiedTest(unittest.TestCase): + def test_same_implementation(self): + self.assertIs(_row_get, peer_row_get) + + def test_present_and_missing(self): + self.assertEqual(_row_get({"a": 1}, "a"), 1) + self.assertIsNone(_row_get({"a": 1}, "missing")) + self.assertIsNone(_row_get(None, "a")) + + def test_index_error_row(self): + class OldRow: + def __getitem__(self, name): + raise IndexError(name) + + self.assertIsNone(_row_get(OldRow(), "missing_col")) + + +if __name__ == "__main__": + unittest.main() diff --git a/cloud/tests/test_sub2api.py b/cloud/tests/test_sub2api.py index 0fd0726..fa40d24 100644 --- a/cloud/tests/test_sub2api.py +++ b/cloud/tests/test_sub2api.py @@ -494,7 +494,6 @@ class TestDockerSecretEnv(unittest.TestCase): return ex def test_secret_name_only_in_argv(self): - from cloud.sandbox import DockerExecutor seen = {} def runner(cmd, timeout=None, env=None): @@ -517,7 +516,6 @@ class TestDockerSecretEnv(unittest.TestCase): self.assertIsNone(ex._docker_cli_env(None)) def test_call_runner_env_loud_fail(self): - from cloud.sandbox import DockerExecutor def cmd_only(cmd, timeout=None): return object() diff --git a/cloud/tests/test_task_environments.py b/cloud/tests/test_task_environments.py index e1e432d..dfb478d 100644 --- a/cloud/tests/test_task_environments.py +++ b/cloud/tests/test_task_environments.py @@ -18,7 +18,6 @@ import tempfile import threading import time import unittest -import uuid from pathlib import Path from types import SimpleNamespace diff --git a/cloud/tests/test_task_links.py b/cloud/tests/test_task_links.py index 7653c61..bb90393 100644 --- a/cloud/tests/test_task_links.py +++ b/cloud/tests/test_task_links.py @@ -19,13 +19,13 @@ from types import SimpleNamespace try: from cloud.auth import AuthStore - from cloud.sandbox import DockerExecutor, LocalExecutor + from cloud.sandbox import DockerExecutor from cloud.server import create_server, init_db from cloud.worker import Worker except ImportError: # discover -s cloud/tests 时把项目根目录补进 sys.path sys.path.insert(0, str(Path(__file__).resolve().parents[2])) from cloud.auth import AuthStore - from cloud.sandbox import DockerExecutor, LocalExecutor + from cloud.sandbox import DockerExecutor from cloud.server import create_server, init_db from cloud.worker import Worker diff --git a/cloud/tests/test_task_trace.py b/cloud/tests/test_task_trace.py index 95b6b85..1918f77 100644 --- a/cloud/tests/test_task_trace.py +++ b/cloud/tests/test_task_trace.py @@ -8,7 +8,6 @@ import http.client import json import os -import socket import sqlite3 import sys import tempfile diff --git a/cloud/tests/test_webhooks.py b/cloud/tests/test_webhooks.py index 3e53a57..08971d0 100644 --- a/cloud/tests/test_webhooks.py +++ b/cloud/tests/test_webhooks.py @@ -19,7 +19,6 @@ sys.path.insert( 0, os.path.normpath(os.path.join(os.path.dirname(__file__), "..")) ) -import webhooks # noqa: E402 from webhooks import ( # noqa: E402 ALLOWED_NOTIFICATION_EVENTS, NotificationService, diff --git a/cloud/wait_ops.py b/cloud/wait_ops.py index d62b24c..9afdde8 100644 --- a/cloud/wait_ops.py +++ b/cloud/wait_ops.py @@ -36,7 +36,6 @@ automations 同款），不在 POSTGRES_SCHEMA 里，因此必须同步 import base64 import json -import os import re import time import uuid diff --git a/cloud/webhooks.py b/cloud/webhooks.py index b2b1e58..66cf978 100644 --- a/cloud/webhooks.py +++ b/cloud/webhooks.py @@ -18,7 +18,6 @@ import hashlib import hmac -import sqlite3 import time from cloud.db import connect as _db_connect diff --git a/cloud/websearch.py b/cloud/websearch.py index f1a3288..92d1930 100644 --- a/cloud/websearch.py +++ b/cloud/websearch.py @@ -58,8 +58,6 @@ import html as _html import ipaddress import json import os -import re -import socket import urllib.error import urllib.parse import urllib.request diff --git a/cloud/websearch_mcp.py b/cloud/websearch_mcp.py index 280d312..765fdd8 100644 --- a/cloud/websearch_mcp.py +++ b/cloud/websearch_mcp.py @@ -15,15 +15,23 @@ copying; it never writes API keys into config.toml -- keys are read from the MCP server process environment by websearch.py. """ +import importlib import json import os import sys +# cloud 包优先、租户 flat copy 回落（与 playwright_mcp 同款理由： +# 两次 from-import 会让 mypy 报 no-redef，importlib 属性绑定 +# 在两条路径下都是同一组名字；cloud 导入本身的 ImportError +# 同样回落，与原来 except 语义一致）。 try: - from cloud.websearch import WebSearchError, fetch_best, search + _mcp_backend = importlib.import_module("cloud.websearch") except ImportError: # flat copy in a tenant codex-home sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) - from websearch import WebSearchError, fetch_best, search # noqa: E402 + _mcp_backend = importlib.import_module("websearch") +WebSearchError = _mcp_backend.WebSearchError +fetch_best = _mcp_backend.fetch_best +search = _mcp_backend.search SERVER_NAME = "websearch" SERVER_VERSION = "0.1.0" diff --git a/deploy/deploy.py b/deploy/deploy.py index 1422e95..d282832 100644 --- a/deploy/deploy.py +++ b/deploy/deploy.py @@ -44,7 +44,6 @@ import hashlib import json import os import re -import shutil import subprocess import sys import tarfile diff --git a/deploy/tests/test_deploy.py b/deploy/tests/test_deploy.py index 3d3ff06..af63b9b 100644 --- a/deploy/tests/test_deploy.py +++ b/deploy/tests/test_deploy.py @@ -46,7 +46,8 @@ class TestDeployPure(unittest.TestCase): with tempfile.NamedTemporaryFile(suffix=".tar.gz", delete=False) as f: path = f.name try: - with tarfile.open(path, "w:gz") as tf: + # 空 tarball 文件本身是测试前置（tarball_commit 应拒绝）：上下文照跑，只去 as 绑定。 + with tarfile.open(path, "w:gz"): pass with self.assertRaises(SystemExit): deploy_mod.tarball_commit(path) diff --git a/deploy/tests/test_deploy_gates.py b/deploy/tests/test_deploy_gates.py new file mode 100644 index 0000000..568acfa --- /dev/null +++ b/deploy/tests/test_deploy_gates.py @@ -0,0 +1,148 @@ +"""deploy.py 门禁编排单测（CI 第二阶段覆盖率补点）。 + +只测编排层：run() 的 check 开关、check_clean_tree 的 +拒绝/告警/放行、deploy_via_git 的 dry-run 与各门禁拒绝、 +main() 的 git 分发与 tar 路径前置检查。真实 SSH/传输仍由 +test_deploy.py 的 transfer/remote_sha 测试覆盖，本文件不 +碰网络与主机。 +""" + +import os +import subprocess +import sys +import unittest +from types import SimpleNamespace +from unittest import mock + +REPO_ROOT = os.path.join(os.path.dirname(__file__), "..", "..") +sys.path.insert(0, os.path.join(REPO_ROOT, "deploy")) +import deploy as deploy_mod # noqa: E402 + + +def _args(**over): + base = dict(tarball=None, build_only=False, host=None, user="root", + dry_run=True, repo="git@github.com:openqj/codex-app-server", + git_key="/tmp/k", allow_dirty=False, skip_tests=False, + ref="HEAD", commit=None, source="git", + out_dir="/tmp/out") + base.update(over) + return SimpleNamespace(**base) + + +class RunAndCleanTreeTest(unittest.TestCase): + def test_run_check_exits_on_failure(self): + bad = subprocess.CompletedProcess(args=[], returncode=1, + stdout="", stderr="boom") + with mock.patch.object(deploy_mod.subprocess, "run", + return_value=bad): + with self.assertRaises(SystemExit) as ctx: + deploy_mod.run(["false"]) + self.assertEqual(ctx.exception.code, 1) + + def test_run_no_check_returns_result(self): + bad = subprocess.CompletedProcess(args=[], returncode=1, + stdout="o", stderr="e") + with mock.patch.object(deploy_mod.subprocess, "run", + return_value=bad): + self.assertIs(deploy_mod.run(["false"], check=False), bad) + + def test_clean_tree_dirty_refuses_and_warns(self): + dirty = SimpleNamespace(returncode=1) + with mock.patch.object(deploy_mod, "run", return_value=dirty): + with self.assertRaises(SystemExit) as ctx: + deploy_mod.check_clean_tree(False) + self.assertEqual(ctx.exception.code, 2) + deploy_mod.check_clean_tree(True) # 只告警，不退出 + + def test_clean_tree_clean_passes(self): + clean = SimpleNamespace(returncode=0) + with mock.patch.object(deploy_mod, "run", return_value=clean): + deploy_mod.check_clean_tree(False) + + +class DeployViaGitTest(unittest.TestCase): + def _patched(self, commit="a" * 40, on_main=True): + return (mock.patch.object(deploy_mod, "check_clean_tree"), + mock.patch.object(deploy_mod, "run_tests"), + mock.patch.object(deploy_mod, "git_commit", + return_value=commit), + mock.patch.object(deploy_mod, "commit_on_origin_main", + return_value=on_main)) + + def test_dry_run_success(self): + patches = self._patched() + for p in patches: + p.start() + try: + self.assertEqual(deploy_mod.deploy_via_git(_args()), 0) + finally: + for p in patches: + p.stop() + + def test_gates(self): + cases = [ + (_args(tarball="/tmp/x.tgz"), "tarball 冲突"), + (_args(repo="https://github.com/o/r"), "非 SSH 仓库"), + ] + for args, _why in cases: + patches = self._patched() + for p in patches: + p.start() + try: + with self.assertRaises(SystemExit) as ctx: + deploy_mod.deploy_via_git(args) + self.assertEqual(ctx.exception.code, 2) + finally: + for p in patches: + p.stop() + patches = self._patched(commit="abc") + for p in patches: + p.start() + try: + with self.assertRaises(SystemExit) as ctx: + deploy_mod.deploy_via_git(_args()) + self.assertEqual(ctx.exception.code, 2) + finally: + for p in patches: + p.stop() + patches = self._patched(on_main=False) + for p in patches: + p.start() + try: + with self.assertRaises(SystemExit) as ctx: + deploy_mod.deploy_via_git(_args()) + self.assertEqual(ctx.exception.code, 2) + finally: + for p in patches: + p.stop() + + +class MainDispatchTest(unittest.TestCase): + def test_main_dispatches_git(self): + seen = [] + with mock.patch.object( + deploy_mod, "deploy_via_git", + side_effect=lambda a: seen.append(a) or 0): + rc = deploy_mod.main(["h", "root", "--source", "git", + "--dry-run"]) + self.assertEqual(rc, 0) + self.assertEqual(seen[0].source, "git") + self.assertTrue(seen[0].dry_run) + + def test_main_tar_needs_host_or_mode(self): + with self.assertRaises(SystemExit) as ctx: + deploy_mod.main([]) + self.assertEqual(ctx.exception.code, 2) + + def test_main_build_only_tar_path(self): + with mock.patch.object(deploy_mod, "check_clean_tree"), \ + mock.patch.object(deploy_mod, "run_tests"), \ + mock.patch.object( + deploy_mod, "build_release", + return_value=("tb.tgz", "sha", "c" * 40)): + rc = deploy_mod.main(["myhost", "--build-only"]) + self.assertEqual(rc, 0) + + +if __name__ == "__main__": + unittest.main() diff --git a/deploy/tests/test_host_git_fetch.py b/deploy/tests/test_host_git_fetch.py index 6111495..9b2b7a8 100644 --- a/deploy/tests/test_host_git_fetch.py +++ b/deploy/tests/test_host_git_fetch.py @@ -6,7 +6,6 @@ host_deploy.sh 代替真实换目录脚本：假脚本只验 tar sha 与 DEPLOYED_COMMIT，打印 RELEASE_OK，不碰 systemctl、不动 live。 host_git_fetch.sh 的 CODEX_GIT_* 环境变量把工作区指到临时目录。 """ -import hashlib import os import subprocess import tarfile diff --git a/deploy/tests/test_verify_multiuser_helpers.py b/deploy/tests/test_verify_multiuser_helpers.py new file mode 100644 index 0000000..332abea --- /dev/null +++ b/deploy/tests/test_verify_multiuser_helpers.py @@ -0,0 +1,129 @@ +"""deploy/verify-multiuser.py 纯 helper 单测（CI 第二阶段覆盖率： +基线 0%）。只测 check() 与 req()： + +- check() 写 RESULTS 台账并打印 PASS/FAIL； +- req() 对本地回环 fake HTTP 成功/HTTPError/非 JSON 体/ + 连接拒绝四路的返回值与请求头。 + +main() 是主机侧对真实控制面的多用户实弹验证（建 +multiuser-test-* 用户/项目/任务、调 merge），按任务门禁 +“不要建生产任务”，本轨不以 mock 冒充它——main() 仍为 +未覆盖，如实留在缺口里。 +""" + +import http.server +import importlib.util +import io +import json +import os +import socket +import threading +import unittest +from contextlib import redirect_stdout +from pathlib import Path + +REPO_ROOT = Path(__file__).resolve().parents[2] +_SPEC = importlib.util.spec_from_file_location( + "verify_multiuser", + os.path.join(REPO_ROOT, "deploy", "verify-multiuser.py")) +vm = importlib.util.module_from_spec(_SPEC) +_SPEC.loader.exec_module(vm) + + +class Handler(http.server.BaseHTTPRequestHandler): + seen = [] + + def _send(self, code, body, ctype="application/json"): + data = body if isinstance(body, bytes) else body.encode() + self.send_response(code) + self.send_header("Content-Type", ctype) + self.send_header("Content-Length", str(len(data))) + self.end_headers() + self.wfile.write(data) + + def do_GET(self): + Handler.seen.append((self.path, dict(self.headers))) + if self.path == "/ok": + self._send(200, json.dumps({"ok": True})) + elif self.path == "/nf": + self._send(404, json.dumps({"error": "nope"})) + else: + self._send(404, "not json", "text/plain") + + def do_POST(self): + length = int(self.headers.get("Content-Length") or 0) + Handler.seen.append( + (self.path, dict(self.headers), self.rfile.read(length))) + self._send(201, json.dumps({"created": True})) + + def log_message(self, *args): + pass + + +class ReqTest(unittest.TestCase): + @classmethod + def setUpClass(cls): + cls.httpd = http.server.ThreadingHTTPServer( + ("127.0.0.1", 0), Handler) + cls.port = cls.httpd.server_address[1] + cls.thread = threading.Thread(target=cls.httpd.serve_forever, + daemon=True) + cls.thread.start() + + @classmethod + def tearDownClass(cls): + cls.httpd.shutdown() + cls.httpd.server_close() + + def setUp(self): + Handler.seen = [] + + def base(self): + return "http://127.0.0.1:%d" % self.port + + def test_get_ok(self): + code, body = vm.req(self.base(), "GET", "/ok") + self.assertEqual((code, body), (200, {"ok": True})) + + def test_http_error_json_and_non_json(self): + code, body = vm.req(self.base(), "GET", "/nf") + self.assertEqual((code, body), (404, {"error": "nope"})) + code, body = vm.req(self.base(), "GET", "/bad") + self.assertEqual((code, body), (404, {})) + + def test_post_headers_and_body(self): + code, body = vm.req(self.base(), "POST", "/ok", + key="k-123", body={"a": 1}) + self.assertEqual(code, 201) + self.assertEqual(body, {"created": True}) + path, headers, raw = Handler.seen[0] + self.assertEqual(headers.get("X-Api-Key"), "k-123") + self.assertEqual(headers.get("Content-Type"), "application/json") + self.assertEqual(json.loads(raw), {"a": 1}) + + def test_connection_refused_returns_none(self): + sock = socket.socket() + sock.bind(("127.0.0.1", 0)) + dead_port = sock.getsockname()[1] + sock.close() + code, body = vm.req("http://127.0.0.1:%d" % dead_port, + "GET", "/ok") + self.assertIsNone(code) + self.assertIn("error", body) + + +class CheckTest(unittest.TestCase): + def test_check_appends_and_prints(self): + vm.RESULTS.clear() + out = io.StringIO() + with redirect_stdout(out): + vm.check("alpha", True) + vm.check("beta", 0) + self.assertEqual(vm.RESULTS, [("alpha", True), ("beta", False)]) + self.assertIn("PASS alpha", out.getvalue()) + self.assertIn("FAIL beta", out.getvalue()) + vm.RESULTS.clear() + + +if __name__ == "__main__": + unittest.main() diff --git a/deploy/verify-multiuser.py b/deploy/verify-multiuser.py index cedad26..fc79fa0 100755 --- a/deploy/verify-multiuser.py +++ b/deploy/verify-multiuser.py @@ -56,7 +56,7 @@ P1 = PREFIX + "p1" P2 = PREFIX + "p2" NEVER_DEP = PREFIX + "never-completes" -RESULTS = [] +RESULTS: list[tuple[str, bool]] = [] def check(name, ok): diff --git a/examples/harness2_demo.py b/examples/harness2_demo.py index 53e1111..a96e00f 100644 --- a/examples/harness2_demo.py +++ b/examples/harness2_demo.py @@ -12,7 +12,6 @@ base 会漏掉它）；确定性编辑器去掉正数过滤；选择器按产物 """ import argparse -import csv import subprocess import sys import tempfile diff --git a/host.py b/host.py index d58a2f4..fa12513 100644 --- a/host.py +++ b/host.py @@ -1754,8 +1754,9 @@ def dispatch_from_branch(server, state, name, prompt, repo, dep_branch, 核对把它拦下来回滚了）。""" repo = os.path.abspath(repo) git(["add", "-A"], dep_worktree) - c = git(["commit", "-m", f"host: {name} base from {dep_branch}"], - dep_worktree) + # 提交本身是依赖派发的前置（分支须含依赖改动）：只去绑定，不删调用。 + git(["commit", "-m", f"host: {name} base from {dep_branch}"], + dep_worktree) # nothing to commit 时分支里已有全部改动，继续用它当基底 base = os.path.join(HERE, "work", "host-worktrees") os.makedirs(base, exist_ok=True) @@ -2281,16 +2282,22 @@ def main(): p.add_argument("--max-retries", type=int, default=0) sub.add_parser("list") - p = sub.add_parser("status"); p.add_argument("--task", required=True) - p = sub.add_parser("review"); p.add_argument("--task", required=True) - p = sub.add_parser("merge"); p.add_argument("--task", required=True) + p = sub.add_parser("status") + p.add_argument("--task", required=True) + p = sub.add_parser("review") + p.add_argument("--task", required=True) + p = sub.add_parser("merge") + p.add_argument("--task", required=True) p.add_argument("--verify", default=None, help="合并后验测命令；不传时仓库有 tests/ 就默认跑 unittest") p.add_argument("--no-verify", action="store_true") - p = sub.add_parser("review-comment"); p.add_argument("--task", required=True) - p.add_argument("--file", required=True); p.add_argument("--line", type=int, required=True) + p = sub.add_parser("review-comment") + p.add_argument("--task", required=True) + p.add_argument("--file", required=True) + p.add_argument("--line", type=int, required=True) p.add_argument("--comment", required=True) - p = sub.add_parser("fix-review"); p.add_argument("--task", required=True) + p = sub.add_parser("fix-review") + p.add_argument("--task", required=True) sub.add_parser("doctor") sub.add_parser("failures") sub.add_parser("health") @@ -2305,35 +2312,61 @@ def main(): help="前台跑循环（测试用，不双 fork）") sub.add_parser("daemon-status") sub.add_parser("daemon-stop") - p = sub.add_parser("discard"); p.add_argument("--task", required=True) - p = sub.add_parser("archive"); p.add_argument("--task", required=True) - p = sub.add_parser("fork"); p.add_argument("--task", required=True); p.add_argument("--name", required=True); p.add_argument("--sandbox", default=None) - p = sub.add_parser("compact"); p.add_argument("--task", required=True) - p = sub.add_parser("code-review"); p.add_argument("--task", required=True) + p = sub.add_parser("discard") + p.add_argument("--task", required=True) + p = sub.add_parser("archive") + p.add_argument("--task", required=True) + p = sub.add_parser("fork") + p.add_argument("--task", required=True) + p.add_argument("--name", required=True) + p.add_argument("--sandbox", default=None) + p = sub.add_parser("compact") + p.add_argument("--task", required=True) + p = sub.add_parser("code-review") + p.add_argument("--task", required=True) p.add_argument("--base-branch", default=None, help="评审目标：与该分支的差异（baseBranch）") p.add_argument("--commit", default=None, help="评审目标：指定提交 sha（commit）") p.add_argument("--instructions", default=None, help="评审目标：自定义评审指令（custom）") - p = sub.add_parser("thread-list"); p.add_argument("--cwd", default=None); p.add_argument("--archived", action="store_true") - p = sub.add_parser("thread-read"); p.add_argument("--thread-id", required=True); p.add_argument("--include-turns", action="store_true") - p = sub.add_parser("thread-rename"); p.add_argument("--thread-id", required=True); p.add_argument("--name", required=True) - p = sub.add_parser("thread-archive"); p.add_argument("--thread-id", required=True) - p = sub.add_parser("thread-unarchive"); p.add_argument("--thread-id", required=True) - p = sub.add_parser("thread-compact"); p.add_argument("--thread-id", required=True) - p = sub.add_parser("skills-list"); p.add_argument("--cwd", default=None) + p = sub.add_parser("thread-list") + p.add_argument("--cwd", default=None) + p.add_argument("--archived", action="store_true") + p = sub.add_parser("thread-read") + p.add_argument("--thread-id", required=True) + p.add_argument("--include-turns", action="store_true") + p = sub.add_parser("thread-rename") + p.add_argument("--thread-id", required=True) + p.add_argument("--name", required=True) + p = sub.add_parser("thread-archive") + p.add_argument("--thread-id", required=True) + p = sub.add_parser("thread-unarchive") + p.add_argument("--thread-id", required=True) + p = sub.add_parser("thread-compact") + p.add_argument("--thread-id", required=True) + p = sub.add_parser("skills-list") + p.add_argument("--cwd", default=None) p = sub.add_parser("apps-list") p = sub.add_parser("mcp-servers-list") p = sub.add_parser("mcp-servers-reload") - p = sub.add_parser("mcp-server-oauth-login"); p.add_argument("--server-name", required=True); p.add_argument("--scopes", default=None); p.add_argument("--timeout-secs", type=int, default=None) - p = sub.add_parser("run"); p.add_argument("--task", required=True); p.add_argument("--prompt", required=True) - p = sub.add_parser("interrupt"); p.add_argument("--task", required=True) + p = sub.add_parser("mcp-server-oauth-login") + p.add_argument("--server-name", required=True) + p.add_argument("--scopes", default=None) + p.add_argument("--timeout-secs", type=int, default=None) + p = sub.add_parser("run") + p.add_argument("--task", required=True) + p.add_argument("--prompt", required=True) + p = sub.add_parser("interrupt") + p.add_argument("--task", required=True) p = sub.add_parser("delegate") - p.add_argument("--parent-task", required=True); p.add_argument("--task", required=True) - p.add_argument("--name", required=True); p.add_argument("--report-to-parent", action="store_true") + p.add_argument("--parent-task", required=True) + p.add_argument("--task", required=True) + p.add_argument("--name", required=True) + p.add_argument("--report-to-parent", action="store_true") p = sub.add_parser("goal") - p.add_argument("--name", required=True); p.add_argument("--task", required=True) + p.add_argument("--name", required=True) + p.add_argument("--task", required=True) p.add_argument("--repo", default=None) p.add_argument("--seed-dir", default=None, help="附件暂存目录：派发时拷进 worktree 再跑 turn") @@ -2357,9 +2390,13 @@ def main(): help="不传时沿用原任务的 goal_max_rounds；传了则覆盖总轮次") p.add_argument("--verify-timeout", type=int, default=600, help="验测命令上限（秒）；0 表示不限时，默认 600") - p = sub.add_parser("parallel"); p.add_argument("--jobs", required=True) - p = sub.add_parser("automations"); p.add_argument("--add", action="store_true") - p.add_argument("--name"); p.add_argument("--task"); p.add_argument("--repo") + p = sub.add_parser("parallel") + p.add_argument("--jobs", required=True) + p = sub.add_parser("automations") + p.add_argument("--add", action="store_true") + p.add_argument("--name") + p.add_argument("--task") + p.add_argument("--repo") p.add_argument("--interval", type=int, default=86400) p.add_argument("--cron-line", action="store_true", help="只打印 run-due 的 crontab 行，不装机") @@ -2368,13 +2405,17 @@ def main(): p.add_argument("--cron-every-minutes", type=int, default=30) sub.add_parser("run-due") sub.add_parser("github-workflow") - p = sub.add_parser("pr"); p.add_argument("--task", required=True) + p = sub.add_parser("pr") + p.add_argument("--task", required=True) p.add_argument("--repo", default=None) - p = sub.add_parser("pr-checks"); p.add_argument("--task", required=True) + p = sub.add_parser("pr-checks") + p.add_argument("--task", required=True) p.add_argument("--repo", default=None) - p = sub.add_parser("pr-merge"); p.add_argument("--task", required=True) + p = sub.add_parser("pr-merge") + p.add_argument("--task", required=True) p.add_argument("--repo", default=None) - p = sub.add_parser("sleep"); p.add_argument("--seconds", type=float, + p = sub.add_parser("sleep") + p.add_argument("--seconds", type=float, default=5) args = ap.parse_args() diff --git a/tests/test_client.py b/tests/test_client.py new file mode 100644 index 0000000..3bfd5d1 --- /dev/null +++ b/tests/test_client.py @@ -0,0 +1,293 @@ +"""client.py 单测（CI 第二阶段覆盖率：基线 0%）。 + +不跑真实 codex 二进制：AppServer 只跟子进程的 stdin/stdout +说话，本文件用真实 os.pipe 作 stdout 通道（select + os.read +原样走）、BytesIO 作 stdin——JSON-RPC 成帧、审批自动回包、 +wait_for 事件分发、initialize/close 都是被测代码的真实逻辑， +只把“子进程是 codex”换成管道。main() 的分支用一个记录型 +FakeAppServer 测（断言它发出的请求参数与分支选择）。 +""" + +import io +import json +import os +import sys +import tempfile +import types +import unittest +from contextlib import redirect_stdout, redirect_stderr +from pathlib import Path +from unittest import mock + +REPO_ROOT = Path(__file__).resolve().parents[1] +sys.path.insert(0, str(REPO_ROOT)) + +import client as client_mod + + +class FakeProc: + """Popen 替身：stdout 是真实 pipe（可 select/os.read）， + stdin 是 BytesIO（可断言发出的每一行）。""" + + def __init__(self, chunks, wait_exc=None): + self.r, self.w = os.pipe() + for chunk in chunks: + os.write(self.w, chunk) + os.close(self.w) + self.stdin = io.BytesIO() + self.stdout = types.SimpleNamespace(fileno=lambda: self.r) + self._wait_exc = wait_exc + self.waited = False + self.killed = False + + def poll(self): + return None + + def wait(self, timeout=None): + self.waited = True + if self._wait_exc is not None: + raise self._wait_exc + return 0 + + def kill(self): + self.killed = True + + def close_fds(self): + try: + os.close(self.r) + except OSError: + pass + + +def _line(obj): + return (json.dumps(obj) + "\n").encode() + + +class AppServerPipeTest(unittest.TestCase): + def _server(self, chunks, codex_home=None, wait_exc=None): + proc = FakeProc(chunks, wait_exc=wait_exc) + self.addCleanup(proc.close_fds) + with mock.patch.object(client_mod.subprocess, "Popen", + return_value=proc): + srv = client_mod.AppServer(codex_home=codex_home) + srv._proc = proc + return srv, proc + + def test_request_and_notify_framing(self): + srv, proc = self._server([]) + rid = srv.request("thread/list", {"a": 1}) + self.assertEqual(rid, 1) + srv.notify("initialized", {}) + self.assertEqual(srv.next_id, 2) + lines = proc.stdin.getvalue().decode().splitlines() + self.assertEqual(json.loads(lines[0]), + {"jsonrpc": "2.0", "id": 1, "method": "thread/list", + "params": {"a": 1}}) + self.assertEqual(json.loads(lines[1]), + {"jsonrpc": "2.0", "method": "initialized", + "params": {}}) + + def test_read_msg_answers_approval_then_returns_response(self): + chunks = [_line({"jsonrpc": "2.0", "id": 9, + "method": "approval/request", + "params": {"x": 1}}), + _line({"jsonrpc": "2.0", "id": 1, "result": {"ok": True}})] + srv, proc = self._server(chunks) + srv.approval_policy = "deny" + msg = srv._read_msg(5) + self.assertEqual(msg, {"jsonrpc": "2.0", "id": 1, + "result": {"ok": True}}) + sent = proc.stdin.getvalue().decode().splitlines() + self.assertEqual(json.loads(sent[0]), + {"jsonrpc": "2.0", "id": 9, + "result": {"decision": "deny"}}) + + def test_read_msg_approval_auto(self): + chunks = [_line({"jsonrpc": "2.0", "id": 7, + "method": "permission/request", "params": {}}), + _line({"jsonrpc": "2.0", "id": 1, "result": {}})] + srv, proc = self._server(chunks) + srv.approval_policy = "auto" + with redirect_stderr(io.StringIO()): + msg = srv._read_msg(5) + self.assertEqual(msg["id"], 1) + sent = proc.stdin.getvalue().decode().splitlines() + self.assertEqual(json.loads(sent[0])["result"], + {"decision": "approve"}) + + def test_wait_for_dispatches_events(self): + chunks = [_line({"jsonrpc": "2.0", "method": "agent/message/delta", + "params": {"item": {"type": "agentMessage", + "text": "hi"}}}), + _line({"jsonrpc": "2.0", "id": 3, "result": {"done": 1}})] + srv, _ = self._server(chunks) + seen = [] + msg = srv.wait_for(3, timeout=5, on_event=seen.append) + self.assertEqual(msg["result"], {"done": 1}) + self.assertEqual(len(seen), 1) + self.assertEqual(seen[0]["method"], "agent/message/delta") + + def test_wait_for_eof_returns_none(self): + srv, _ = self._server([]) + self.assertIsNone(srv.wait_for(1, timeout=5)) + + def test_initialize_success_and_error(self): + chunks = [_line({"jsonrpc": "2.0", "id": 1, + "result": {"codexHome": "/tmp/ch", + "platformOs": "linux"}})] + srv, proc = self._server(chunks) + info = srv.initialize() + self.assertEqual(info["platformOs"], "linux") + sent = proc.stdin.getvalue().decode().splitlines() + self.assertEqual(json.loads(sent[0])["method"], "initialize") + self.assertEqual(json.loads(sent[1])["method"], "initialized") + + srv2, _ = self._server( + [_line({"jsonrpc": "2.0", "id": 1, + "error": {"code": -1, "message": "boom"}})]) + with self.assertRaises(RuntimeError): + srv2.initialize() + + def test_codex_home_created(self): + with tempfile.TemporaryDirectory() as tmp: + home = os.path.join(tmp, "new-home") + srv, proc = self._server([], codex_home=home) + self.assertTrue(os.path.isdir(home)) + self.assertEqual(proc.stdin.getvalue(), b"") + + def test_close_waits_and_kills_on_timeout(self): + srv, proc = self._server([]) + srv.close() + self.assertTrue(proc.waited) + self.assertFalse(proc.killed) + import subprocess as _sp + srv2, proc2 = self._server( + [], wait_exc=_sp.TimeoutExpired("codex", 5)) + srv2.close() + self.assertTrue(proc2.killed) + + +class HandleEventTest(unittest.TestCase): + def test_agent_delta_printed(self): + out = io.StringIO() + with redirect_stdout(out): + client_mod.handle_event( + {"method": "agent/message/delta", + "params": {"item": {"type": "agentMessage", + "text": "hello"}}}) + self.assertEqual(out.getvalue(), "hello") + + def test_turn_completed_and_error_paths(self): + out, err = io.StringIO(), io.StringIO() + with redirect_stdout(out), redirect_stderr(err): + client_mod.handle_event( + {"method": "turn/completed", + "params": {"status": "completed"}}) + # 首分支元组外、含 started 的方法走 stderr 分支。 + client_mod.handle_event( + {"method": "session/started", "params": {}}) + client_mod.handle_event( + {"method": "something", "params": {"error": "x"}}) + self.assertIn("[turn/completed]", out.getvalue()) + self.assertIn("[session/started]", err.getvalue()) + self.assertIn("[something]", err.getvalue()) + + +class FakeAppServer: + def __init__(self, codex_home=None, extra_args=None): + self.codex_home = codex_home + self.approval_policy = "deny" + self.requests = [] + self.notifies = [] + self.responses = {} + self.read_msgs = [] + self.closed = False + + def request(self, method, params=None): + rid = len(self.requests) + 1 + self.requests.append((method, params)) + return rid + + def notify(self, method, params=None): + self.notifies.append((method, params)) + + def wait_for(self, rid, timeout=60, on_event=None): + return self.responses.get(rid) + + def _read_msg(self, timeout): + return self.read_msgs.pop(0) if self.read_msgs else None + + def initialize(self): + return {"codexHome": "/tmp/ch", "platformOs": "linux"} + + def close(self): + self.closed = True + + +class ClientMainTest(unittest.TestCase): + def _run(self, argv, setup): + fake = FakeAppServer() + setup(fake) + err = io.StringIO() + with mock.patch.object(client_mod, "AppServer", + return_value=fake), \ + mock.patch.object(sys, "argv", argv), \ + redirect_stderr(err): + out = io.StringIO() + with redirect_stdout(out): + client_mod.main() + return fake, out.getvalue(), err.getvalue() + + def test_list_threads(self): + def setup(fake): + fake.responses[1] = {"result": {"threads": []}} + fake, out, _ = self._run( + ["client.py", "--list-threads", "--cwd", "/tmp"], + setup) + self.assertEqual(fake.requests[0][0], "thread/list") + self.assertIn("threads", out) + self.assertTrue(fake.closed) + + def test_start_and_turn_completed(self): + def setup(fake): + fake.responses[1] = {"result": {"thread": {"id": "th-9"}, + "model": "m"}} + fake.read_msgs.append( + {"method": "turn/completed", "params": {"ok": True}}) + fake, out, _ = self._run( + ["client.py", "--prompt", "fix it", "--cwd", "/tmp", + "--model", "gpt-x", "--sandbox", "read-only", + "--approval-policy", "never"], + setup) + start = fake.requests[0] + self.assertEqual(start[0], "thread/start") + self.assertEqual(start[1]["approvalPolicy"], "never") + self.assertEqual(start[1]["sandbox"], "read-only") + self.assertEqual(start[1]["model"], "gpt-x") + turn = fake.requests[1] + self.assertEqual(turn[0], "turn/start") + self.assertEqual(turn[1]["threadId"], "th-9") + self.assertEqual(turn[1]["input"][0]["text"], "fix it") + self.assertIn("[done]", out) + self.assertTrue(fake.closed) + + def test_resume_failure_prints_and_returns(self): + def setup(fake): + fake.responses[1] = {"error": {"message": "gone"}} + fake, out, err = self._run( + ["client.py", "--thread-id", "th-1", "--cwd", "/tmp"], + setup) + self.assertEqual(fake.requests[0][0], "thread/resume") + self.assertIn("thread/resume", err) + self.assertTrue(fake.closed) + + def test_no_prompt_prints_thread_id(self): + def setup(fake): + fake.responses[1] = {"result": {"thread": {"id": "th-2"}}} + _, out, _ = self._run( + ["client.py", "--cwd", "/tmp"], setup) + self.assertIn("thread_id=th-2", out) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_host.py b/tests/test_host.py index 3e4e16c..accbcbe 100644 --- a/tests/test_host.py +++ b/tests/test_host.py @@ -126,7 +126,8 @@ class MergeVerifyTest(unittest.TestCase): " self.assertEqual(f(), 1)\n") (tests_dir / "test_mod.py").write_text(body, encoding="utf-8") subprocess.run(["git", "add", "-A"], cwd=tmp, check=True) - subprocess.run(["git", "-c", "user.name=Test", "-c", "user.email=test@example.com", "commit", "-qm", "initial"], cwd=tmp, + subprocess.run(["git", "-c", "user.name=Test", "-c", "user.email=test@example.com", + "commit", "-qm", "initial"], cwd=tmp, check=True) # worktree 手动建：模拟 host 的 worktree/分支 subprocess.run(["git", "worktree", "add", os.path.join(tmp, "wt"), @@ -175,7 +176,8 @@ class MergeVerifyTest(unittest.TestCase): Path(tmp, "mod.py").write_text("def f():\n return 1\n", encoding="utf-8") subprocess.run(["git", "add", "-A"], cwd=tmp, check=True) - subprocess.run(["git", "-c", "user.name=Test", "-c", "user.email=test@example.com", "commit", "-qm", "initial"], cwd=tmp, + subprocess.run(["git", "-c", "user.name=Test", "-c", "user.email=test@example.com", + "commit", "-qm", "initial"], cwd=tmp, check=True) # 分支1：加 add2 subprocess.run(["git", "worktree", "add", @@ -643,7 +645,8 @@ class GithubPRTest(unittest.TestCase): _init_git_repo(tmp) Path(tmp, "f.txt").write_text("x\n") subprocess.run(["git", "add", "-A"], cwd=tmp, check=True) - subprocess.run(["git", "-c", "user.name=Test", "-c", "user.email=test@example.com", "commit", "-qm", "initial"], + subprocess.run(["git", "-c", "user.name=Test", "-c", "user.email=test@example.com", + "commit", "-qm", "initial"], cwd=tmp, check=True) task = {"name": "t", "branch": "host/t", "status": "completed", "prompt": "do it"} @@ -658,12 +661,14 @@ class GithubPRTest(unittest.TestCase): _init_git_repo(repo) Path(repo, "f.txt").write_text("x\n") subprocess.run(["git", "add", "-A"], cwd=repo, check=True) - subprocess.run(["git", "-c", "user.name=Test", "-c", "user.email=test@example.com", "commit", "-qm", "initial"], + subprocess.run(["git", "-c", "user.name=Test", "-c", "user.email=test@example.com", + "commit", "-qm", "initial"], cwd=repo, check=True) subprocess.run(["git", "checkout", "-qb", "host/t1"], cwd=repo, check=True) Path(repo, "f.txt").write_text("y\n") - subprocess.run(["git", "-c", "user.name=Test", "-c", "user.email=test@example.com", "commit", "-qam", "change"], + subprocess.run(["git", "-c", "user.name=Test", "-c", "user.email=test@example.com", + "commit", "-qam", "change"], cwd=repo, check=True) remote = os.path.join(tmp, "remote.git") subprocess.run(["git", "init", "--bare", "-q", remote], diff --git a/tests/test_host_dispatch_branch.py b/tests/test_host_dispatch_branch.py new file mode 100644 index 0000000..155c93f --- /dev/null +++ b/tests/test_host_dispatch_branch.py @@ -0,0 +1,94 @@ +"""dispatch_from_branch 回归测试（CI 第二阶段 F841 证据）。 + +背景：host.py 的 dispatch_from_branch 里 `git commit` 的返回值 +曾以 `c = git(...)` 绑定却从未读取（F841）。本轨只去绑定、 +不删调用——提交本身是依赖派发的前置：依赖 worktree 的改动 +必须先落到依赖分支，新 worktree 才从该分支建出、带血缘。 +本测试用真实 git 仓库断言这条链： + +- 依赖 worktree 里未提交的改动被提交到依赖分支； +- 新 worktree 的分支 host/ 由依赖分支 HEAD 派生 + （含该提交），且任务记录进 state。 +""" + +import os +import shutil +import subprocess +import sys +import tempfile +import unittest +import uuid +from pathlib import Path + +REPO_ROOT = Path(__file__).resolve().parents[1] +sys.path.insert(0, str(REPO_ROOT)) + +import host + + +def _git(args, cwd, check=True): + return subprocess.run(["git"] + args, cwd=cwd, check=check, + capture_output=True, text=True) + + +class FakeServer: + def create_thread(self, path, model=None, sandbox=None, + approval_policy=None): + self.created = (path, model, sandbox, approval_policy) + return "th-1" + + +class DispatchFromBranchTest(unittest.TestCase): + def test_commit_then_worktree_from_dep_branch(self): + name = "dep-" + uuid.uuid4().hex[:8] + server = FakeServer() + state = {"tasks": {}} + with tempfile.TemporaryDirectory() as tmp: + repo = os.path.join(tmp, "repo") + os.makedirs(repo) + _git(["init", "-q", "-b", "main"], repo) + _git(["config", "user.name", "Test"], repo) + _git(["config", "user.email", "test@example.com"], repo) + Path(repo, "f.txt").write_text("base\n", encoding="utf-8") + _git(["add", "-A"], repo) + _git(["commit", "-qm", "initial"], repo) + _git(["branch", "dep-branch"], repo) + dep_wt = os.path.join(tmp, "dep-wt") + _git(["worktree", "add", "-q", dep_wt, "dep-branch"], repo) + Path(dep_wt, "f.txt").write_text("dep change\n", + encoding="utf-8") + try: + task, err = host.dispatch_from_branch( + server, state, name, "do it", repo, "dep-branch", + dep_wt, "model-x", "medium", "workspace-write", "never") + self.assertEqual(err, "") + self.assertIsNotNone(task) + self.assertEqual(task["branch"], "host/" + name) + self.assertEqual(task["thread_id"], "th-1") + # 依赖分支 HEAD 含本次提交，且新分支由它派生。 + dep_head = _git(["rev-parse", "dep-branch"], + repo).stdout.strip() + new_head = _git(["rev-parse", "host/" + name], + repo).stdout.strip() + self.assertEqual(new_head, dep_head) + msg = _git(["log", "-1", "--format=%s", "dep-branch"], + repo).stdout.strip() + self.assertIn(name, msg) + self.assertEqual( + Path(task["cwd"], "f.txt").read_text(encoding="utf-8"), + "dep change\n") + self.assertIs(state["tasks"][name], task) + finally: + # 清理 HERE/work 下建出的 worktree（work/ 已 + # gitignore，但不留给下一轮测试；须在临时仓 + # 删除前用该仓的 git 登记移除）。 + path = os.path.join(host.HERE, "work", + "host-worktrees", name) + if os.path.exists(path): + _git(["worktree", "remove", "--force", path], repo, + check=False) + shutil.rmtree(path, ignore_errors=True) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_host_run_verify.py b/tests/test_host_run_verify.py new file mode 100644 index 0000000..488b92f --- /dev/null +++ b/tests/test_host_run_verify.py @@ -0,0 +1,53 @@ +"""host.run_verify / goal verify 的 shell 语义证据（CI 第二阶段 +bandit High triage，第 3 项白名单轨范围外、本轨只做 triage）。 + +背景：bandit High 4 项全是 host.py 的 shell=True： +run_verify（1319）、notify（1497）、goal verify 两处 +（2099/2103）。本轨不改它们，理由须有测试证据： + +- run_verify 的命令是整串 shell 字符串（配置/默认验测命令）， + 管道与复合命令必须经 shell 解释：列表式 exec 形跑不通。 + 本文件断言管道与 && 真实生效——改 shell=False 会立刻失败。 +- notify 的插值已全部 shlex.quote：证据在 tests/test_host.py + 的 test_notify_failure_quotes_shell_metacharacters（`; touch` + 与 `$(touch )` 不执行、只落日志原文），本文件不再重复。 +""" + +import sys +import tempfile +import unittest +from pathlib import Path + +REPO_ROOT = Path(__file__).resolve().parents[1] +sys.path.insert(0, str(REPO_ROOT)) + +import host + + +class RunVerifyShellTest(unittest.TestCase): + def test_pipe_requires_shell(self): + with tempfile.TemporaryDirectory() as tmp: + rc, out = host.run_verify("echo hello | tr a-z A-Z", tmp) + self.assertEqual(rc, 0) + self.assertEqual(out, "HELLO") + + def test_compound_command_requires_shell(self): + with tempfile.TemporaryDirectory() as tmp: + rc, out = host.run_verify("echo a && echo b", tmp) + self.assertEqual(rc, 0) + self.assertEqual(out, "a\nb") + + def test_empty_command_is_noop(self): + with tempfile.TemporaryDirectory() as tmp: + self.assertEqual(host.run_verify("", tmp), (0, "")) + self.assertEqual(host.run_verify(None, tmp), (0, "")) + + def test_failing_command_returns_rc_and_output(self): + with tempfile.TemporaryDirectory() as tmp: + rc, out = host.run_verify("echo oops; exit 3", tmp) + self.assertEqual(rc, 3) + self.assertEqual(out, "oops") + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_host_trace.py b/tests/test_host_trace.py index 671764c..e75e621 100644 --- a/tests/test_host_trace.py +++ b/tests/test_host_trace.py @@ -13,11 +13,9 @@ import unittest from pathlib import Path try: - import host from host import Server, TRACE_MAX_LINES, _summarize_trace_item except ImportError: sys.path.insert(0, str(Path(__file__).resolve().parents[1])) - import host from host import Server, TRACE_MAX_LINES, _summarize_trace_item diff --git a/tests/test_run_parallel.py b/tests/test_run_parallel.py new file mode 100644 index 0000000..751ec07 --- /dev/null +++ b/tests/test_run_parallel.py @@ -0,0 +1,105 @@ +"""run_parallel.py 单测（CI 第二阶段覆盖率：基线 0%）。 + +不跑真实 client.py/codex：只断言 main() 拼出的子进程命令 +（可执行文件、client.py 路径、cwd 解析、sandbox 默认值、 +approval-policy、codex-home 透传）与成功计数/返回码—— +命令构造与汇总正是本文件的全部逻辑。 +""" + +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 run_parallel as rp_mod + + +class FakePopen: + seen = [] + + def __init__(self, cmd): + self.cmd = cmd + FakePopen.seen.append(cmd) + self._rc = 0 + + def wait(self): + return self._rc + + +class RunParallelTest(unittest.TestCase): + def _run(self, jobs, extra_args): + FakePopen.seen = [] + with tempfile.TemporaryDirectory() as tmp: + jobs_path = os.path.join(tmp, "jobs.json") + with open(jobs_path, "w", encoding="utf-8") as f: + json.dump(jobs, f) + argv = ["run_parallel.py", "--jobs", jobs_path] + extra_args + with mock.patch.object(rp_mod.subprocess, "Popen", FakePopen), \ + mock.patch.object(sys, "argv", argv): + out = io.StringIO() + with redirect_stdout(out): + rc = rp_mod.main() + return rc, list(FakePopen.seen), out.getvalue() + + def test_commands_and_all_success(self): + jobs = [{"cwd": "job-a", "prompt": "review a"}, + {"cwd": "/tmp/abs-b", "prompt": "review b", + "sandbox": "workspace-write"}] + rc, seen, out = self._run(jobs, ["--codex-home", "/tmp/ch"]) + self.assertEqual(rc, 0) + self.assertEqual(len(seen), 2) + client = os.path.join(rp_mod.HERE, "client.py") + self.assertEqual(seen[0], [sys.executable, client, + "--prompt", "review a", + "--cwd", os.path.join(rp_mod.HERE, "job-a"), + "--sandbox", "read-only", + "--approval-policy", "never", + "--codex-home", "/tmp/ch"]) + self.assertEqual(seen[1][5], "/tmp/abs-b") + self.assertEqual(seen[1][7], "workspace-write") + self.assertIn("2/2", out) + + def test_one_failure_returns_1(self): + class FailingPopen(FakePopen): + def __init__(self, cmd): + super().__init__(cmd) + self._idx = len(FakePopen.seen) + + def wait(self): + return 1 if self._idx == 2 else 0 + + jobs = [{"cwd": "/tmp/a", "prompt": "x"}, + {"cwd": "/tmp/b", "prompt": "y"}] + FakePopen.seen = [] + with tempfile.TemporaryDirectory() as tmp: + jobs_path = os.path.join(tmp, "jobs.json") + with open(jobs_path, "w", encoding="utf-8") as f: + json.dump(jobs, f) + with mock.patch.object(rp_mod.subprocess, "Popen", + FailingPopen), \ + mock.patch.object(sys, "argv", + ["run_parallel.py", "--jobs", + jobs_path]): + out = io.StringIO() + with redirect_stdout(out): + rc = rp_mod.main() + self.assertEqual(rc, 1) + self.assertIn("1/2", out.getvalue()) + + def test_empty_jobs_returns_0(self): + rc, seen, out = self._run([], []) + self.assertEqual(rc, 0) + self.assertEqual(seen, []) + self.assertIn("0/0", out) + + +if __name__ == "__main__": + unittest.main()