diff --git a/cloud/tests/test_coverage85_mcp.py b/cloud/tests/test_coverage85_mcp.py new file mode 100644 index 0000000..2ec2474 --- /dev/null +++ b/cloud/tests/test_coverage85_mcp.py @@ -0,0 +1,721 @@ +"""覆盖率 78%->85% 第 2 步：三个 stdio MCP（playwright / imagegen / +websearch）的剩余缺口真实行为测试。 + +只用标准库：playwright 侧传假浏览器（handle_request 本来就接受 +browser 参数）；imagegen/websearch 侧传假 generate/edit/search/fetch； +main() 循环在进程内跑（playwright 用真实 pipe fd + 假浏览器， +imagegen/websearch 用 StringIO stdin/stdout + 模块级假依赖）。 +不跑真实浏览器、不联网、不调真实网关。flat-copy 回落导入用 +sys.modules 封堵 cloud.* 后端、重新导入模块本体系测（事后恢复）。 +""" + +import importlib +import io +import json +import os +import sys +import tempfile +import threading +import unittest +from pathlib import Path + +try: + from cloud import imagegen_mcp, playwright_mcp, websearch_mcp +except ImportError: # discover -s cloud/tests 时把项目根目录补进 sys.path + sys.path.insert(0, str(Path(__file__).resolve().parents[2])) + from cloud import imagegen_mcp, playwright_mcp, websearch_mcp + + +class FakeBrowser: + """playwright_mcp handle_request 要调的全部工具方法。""" + + def __init__(self, cwd=None): + self.calls = [] + self.cwd = cwd + self.frame_sink = None + self.closed = False + + def _ret(self, name, *args): + self.calls.append((name, args)) + return {"tool": name, "args": [str(a) for a in args]} + + def navigate(self, url): + return self._ret("navigate", url) + + def snapshot(self): + return self._ret("snapshot") + + def click(self, selector): + return self._ret("click", selector) + + def fill(self, selector, text): + return self._ret("fill", selector, text) + + def click_at(self, x, y): + return self._ret("click_at", x, y) + + def mouse_move(self, x, y): + return self._ret("mouse_move", x, y) + + def key_press(self, key): + return self._ret("key_press", key) + + def type_text(self, text): + return self._ret("type_text", text) + + def snapshot_refs(self): + return self._ret("snapshot_refs") + + def click_ref(self, ref): + return self._ret("click_ref", ref) + + def live_shot(self, output_filename=None): + return self._ret("live_shot", output_filename) + + def evaluate(self, expression): + return self._ret("evaluate", expression) + + def screenshot(self, output_filename=None): + return self._ret("screenshot", output_filename) + + def close(self): + self.closed = True + return self._ret("close") + + def set_frame_sink(self, sink): + self.frame_sink = sink + + def cached_info(self): + return {"url": "http://example.test/", "title": "t", + "started": 1} + + def pump_screencast(self): + return "" + + +class FakeBridge: + def __init__(self, browser, holder="agent", poll_error=None): + self.browser = browser + self.holder = holder + self.poll_error = poll_error + self.poll_calls = 0 + + def get_holder(self): + return self.holder + + def poll(self, browser): + self.poll_calls += 1 + if self.poll_error is not None: + raise self.poll_error + + +def _content_text(resp): + return json.loads(resp["result"]["content"][0]["text"]) + + +class PlaywrightHandleTest(unittest.TestCase): + def setUp(self): + self._old_browser = playwright_mcp._BROWSER + self._old_bridge = playwright_mcp._BRIDGE + + def tearDown(self): + playwright_mcp._BROWSER = self._old_browser + playwright_mcp._BRIDGE = self._old_bridge + + def test_non_dict_request(self): + resp = playwright_mcp.handle_request("nope") + self.assertEqual(resp["error"]["code"], -32700) + self.assertIsNone(resp["id"]) + + def test_initialize_and_notifications(self): + resp = playwright_mcp.handle_request( + {"jsonrpc": "2.0", "id": 1, "method": "initialize", + "params": {"protocolVersion": "2025-01-01"}}) + self.assertEqual(resp["result"]["protocolVersion"], "2025-01-01") + self.assertEqual(resp["result"]["serverInfo"]["name"], "playwright") + resp = playwright_mcp.handle_request( + {"jsonrpc": "2.0", "id": 2, "method": "initialize", + "params": {}}) + self.assertEqual(resp["result"]["protocolVersion"], "2024-11-05") + self.assertIsNone(playwright_mcp.handle_request( + {"jsonrpc": "2.0", "method": "initialize"})) + self.assertIsNone(playwright_mcp.handle_request( + {"jsonrpc": "2.0", "method": "notifications/initialized"})) + self.assertIsNone(playwright_mcp.handle_request( + {"jsonrpc": "2.0", "method": "notifications/cancelled"})) + + def test_ping_and_tools_list(self): + self.assertEqual(playwright_mcp.handle_request( + {"jsonrpc": "2.0", "id": 7, "method": "ping"})["result"], {}) + self.assertIsNone(playwright_mcp.handle_request( + {"jsonrpc": "2.0", "method": "ping"})) + resp = playwright_mcp.handle_request( + {"jsonrpc": "2.0", "id": 8, "method": "tools/list"}) + names = [t["name"] for t in resp["result"]["tools"]] + self.assertEqual(len(names), 14) + for want in ("browser_navigate", "browser_snapshot", "browser_click", + "browser_fill", "browser_evaluate", "browser_click_at", + "browser_live_shot", "browser_snapshot_refs", + "browser_click_ref", "browser_mouse_move", + "browser_key_press", "browser_type_text", + "browser_screenshot", "browser_close"): + self.assertIn(want, names) + self.assertIsNone(playwright_mcp.handle_request( + {"jsonrpc": "2.0", "method": "tools/list"})) + + def _call(self, browser, name, args): + return playwright_mcp.handle_request( + {"jsonrpc": "2.0", "id": 3, "method": "tools/call", + "params": {"name": name, "arguments": args}}, browser) + + def test_tool_dispatch_all(self): + cases = [ + ("browser_navigate", {"url": "http://example.test/"}, + ("navigate", ("http://example.test/",))), + ("browser_snapshot", {}, ("snapshot", ())), + ("browser_click", {"selector": "button.x"}, + ("click", ("button.x",))), + ("browser_fill", {"selector": "input", "text": "hi"}, + ("fill", ("input", "hi"))), + ("browser_click_at", {"x": 10, "y": 20}, + ("click_at", (10, 20))), + ("browser_mouse_move", {"x": 1, "y": 2}, + ("mouse_move", (1, 2))), + ("browser_key_press", {"key": "Enter"}, + ("key_press", ("Enter",))), + ("browser_type_text", {"text": "abc"}, + ("type_text", ("abc",))), + ("browser_snapshot_refs", {}, ("snapshot_refs", ())), + ("browser_click_ref", {"ref": "e3"}, ("click_ref", ("e3",))), + ("browser_live_shot", {"output_filename": "live.png"}, + ("live_shot", ("live.png",))), + ("browser_live_shot", {}, ("live_shot", (None,))), + ("browser_evaluate", {"expression": "1+1"}, + ("evaluate", ("1+1",))), + ("browser_screenshot", {"output_filename": "s.png"}, + ("screenshot", ("s.png",))), + ("browser_screenshot", {}, ("screenshot", (None,))), + ] + for name, args, want_call in cases: + with self.subTest(tool=name, args=args): + browser = FakeBrowser() + playwright_mcp._BRIDGE = FakeBridge(browser) + resp = self._call(browser, name, args) + self.assertFalse(resp["result"]["isError"]) + self.assertEqual(_content_text(resp)["tool"], + want_call[0]) + self.assertIn(want_call, browser.calls) + + def test_tool_close_is_idempotent(self): + browser = FakeBrowser() + playwright_mcp._BRIDGE = FakeBridge(browser) + self._call(browser, "browser_close", {}) + self._call(browser, "browser_close", {}) + self.assertTrue(browser.closed) + self.assertEqual( + [c for c in browser.calls if c[0] == "close"], + [("close", ()), ("close", ())]) + + def test_unknown_tool_and_unknown_method(self): + browser = FakeBrowser() + playwright_mcp._BRIDGE = FakeBridge(browser) + resp = self._call(browser, "browser_nope", {}) + self.assertEqual(resp["error"]["code"], -32602) + self.assertIn("browser_nope", resp["error"]["message"]) + resp = playwright_mcp.handle_request( + {"jsonrpc": "2.0", "id": 9, "method": "resources/list"}) + self.assertEqual(resp["error"]["code"], -32601) + self.assertIsNone(playwright_mcp.handle_request( + {"jsonrpc": "2.0", "method": "resources/list"})) + self.assertIsNone(playwright_mcp.handle_request( + {"jsonrpc": "2.0", "method": "tools/call", + "params": {"name": "browser_snapshot", "arguments": {}}}, + browser)) + + def test_tool_exception_becomes_is_error(self): + class Boom(FakeBrowser): + def snapshot(self): + raise RuntimeError("page gone") + + browser = Boom() + playwright_mcp._BRIDGE = FakeBridge(browser) + resp = self._call(browser, "browser_snapshot", {}) + self.assertTrue(resp["result"]["isError"]) + self.assertIn("page gone", + resp["result"]["content"][0]["text"]) + + def test_user_holds_browser_blocks_tools(self): + browser = FakeBrowser() + playwright_mcp._BRIDGE = FakeBridge(browser, holder="user") + resp = self._call(browser, "browser_snapshot", {}) + self.assertTrue(resp["result"]["isError"]) + self.assertIn("user holds browser control", + resp["result"]["content"][0]["text"]) + self.assertEqual(browser.calls, []) + + def test_get_browser_variants(self): + fake = FakeBrowser() + self.assertIs(playwright_mcp._get_browser(fake), fake) + playwright_mcp._BROWSER = fake + self.assertIs(playwright_mcp._get_browser(), fake) + playwright_mcp._BROWSER = None + real = playwright_mcp._get_browser() + self.assertIsInstance(real, playwright_mcp.Browser) + self.assertIs(playwright_mcp._get_browser(), real) + + def test_get_bridge_constructs_real_bridge_and_caches(self): + tmp = tempfile.TemporaryDirectory() + self.addCleanup(tmp.cleanup) + browser = FakeBrowser(cwd=tmp.name) + playwright_mcp._BRIDGE = None + bridge = playwright_mcp._get_bridge(browser) + self.assertIsNotNone(bridge) + self.assertIs(bridge.browser, browser) + self.assertEqual(bridge.get_holder(), "agent") + self.assertIs(playwright_mcp._get_bridge(browser), bridge) + resp = self._call(browser, "browser_snapshot", {}) + self.assertFalse(resp["result"]["isError"]) + playwright_mcp._drain_remote(browser) + self.assertTrue(os.listdir(tmp.name)) + + def test_get_bridge_none_and_drain(self): + browser = FakeBrowser() + saved = {} + for name in ("playwright_remote", "cloud.playwright_remote"): + saved[name] = sys.modules.get(name) + sys.modules[name] = None + try: + playwright_mcp._BRIDGE = None + self.assertIsNone(playwright_mcp._get_bridge(browser)) + playwright_mcp._drain_remote(browser) # 桥为空直接返回 + finally: + for name, val in saved.items(): + if val is None: + sys.modules.pop(name, None) + else: + sys.modules[name] = val + err = io.StringIO() + old_err = sys.stderr + sys.stderr = err + try: + playwright_mcp._BRIDGE = FakeBridge( + browser, poll_error=RuntimeError("poll boom")) + playwright_mcp._drain_remote(browser) + finally: + sys.stderr = old_err + self.assertIn("bridge poll error", err.getvalue()) + self.assertIn("poll boom", err.getvalue()) + + def test_line_reader_chunks_partial_and_errors(self): + rfd, wfd = os.pipe() + reader = playwright_mcp._LineReader(rfd) + # 空管道非阻塞读：BlockingIOError 分支，返回空且未 drained。 + lines, drained = reader.poll() + self.assertEqual(lines, []) + self.assertFalse(drained) + os.write(wfd, b'{"a": 1}\n{"b": 2}') + lines, drained = reader.poll() + self.assertEqual(lines, ['{"a": 1}']) + self.assertFalse(drained) + os.close(wfd) + lines, drained = reader.poll() + self.assertEqual(lines, ['{"b": 2}']) + self.assertTrue(drained) + os.close(rfd) + # 已关闭 fd：OSError 分支，eof 且 drained。 + rfd2, _wfd2 = os.pipe() + os.close(rfd2) + reader2 = playwright_mcp._LineReader.__new__( + playwright_mcp._LineReader) + reader2._fd = rfd2 + reader2._buf = b"" + reader2.eof = False + lines, drained = reader2.poll() + self.assertTrue(drained) + os.close(_wfd2) + + def test_handle_line_parse_and_handler_errors(self): + out = io.StringIO() + old_out = sys.stdout + sys.stdout = out + try: + playwright_mcp._handle_line("not-json{", FakeBrowser()) + playwright_mcp._handle_line( + json.dumps({"jsonrpc": "2.0", "id": 5, "method": "ping"}), + FakeBrowser()) + # initialize 且 params 不是 dict：handle_request 内抛 + # AttributeError，_handle_line 转 internal error。 + playwright_mcp._handle_line( + json.dumps({"jsonrpc": "2.0", "id": 6, "method": "initialize", + "params": ["x"]}), + FakeBrowser()) + finally: + sys.stdout = old_out + got = [json.loads(ln) for ln in out.getvalue().splitlines()] + self.assertEqual(got[0]["error"]["code"], -32700) + self.assertEqual(got[1]["result"], {}) + self.assertEqual(got[2]["error"]["code"], -32603) + self.assertEqual(got[2]["id"], 6) + + def test_main_loop_with_pipe_and_fake_browser(self): + tmp = tempfile.TemporaryDirectory() + self.addCleanup(tmp.cleanup) + browser = FakeBrowser(cwd=tmp.name) + bridge = FakeBridge(browser) + playwright_mcp._BROWSER = browser + playwright_mcp._BRIDGE = bridge + rfd, wfd = os.pipe() + lines_in = [ + "", + json.dumps({"jsonrpc": "2.0", "id": 1, "method": "initialize"}), + json.dumps({"jsonrpc": "2.0", "id": 2, "method": "tools/call", + "params": {"name": "browser_navigate", + "arguments": {"url": "http://x.test/"}}}), + "broken{", + ] + + def writer(): + import time as _time + _time.sleep(0.05) # 先空转一轮：走 main 的 idle/sleep 分支 + for ln in lines_in: + os.write(wfd, (ln + "\n").encode("utf-8")) + os.close(wfd) + + threading.Thread(target=writer, daemon=True).start() + old_in, old_out = sys.stdin, sys.stdout + new_in = os.fdopen(rfd) + sys.stdin = new_in + sys.stdout = io.StringIO() + try: + playwright_mcp.main() + out = sys.stdout.getvalue() + finally: + sys.stdin = old_in + sys.stdout = old_out + new_in.close() + got = [json.loads(ln) for ln in out.splitlines()] + self.assertEqual(got[0]["result"]["serverInfo"]["name"], "playwright") + self.assertEqual(_content_text(got[1])["tool"], "navigate") + self.assertEqual(got[2]["error"]["code"], -32700) + self.assertIsNotNone(browser.frame_sink) + self.assertGreaterEqual(bridge.poll_calls, 1) + self.assertIn(("navigate", ("http://x.test/",)), browser.calls) + + def test_main_loop_without_bridge(self): + browser = FakeBrowser() + playwright_mcp._BROWSER = browser + playwright_mcp._BRIDGE = None + saved = {} + for name in ("playwright_remote", "cloud.playwright_remote"): + saved[name] = sys.modules.get(name) + sys.modules[name] = None + rfd, wfd = os.pipe() + os.write(wfd, json.dumps( + {"jsonrpc": "2.0", "id": 1, "method": "ping"}).encode() + b"\n") + os.close(wfd) + old_in, old_out = sys.stdin, sys.stdout + new_in = os.fdopen(rfd) + sys.stdin = new_in + sys.stdout = io.StringIO() + try: + playwright_mcp.main() + out = sys.stdout.getvalue() + finally: + sys.stdin = old_in + sys.stdout = old_out + new_in.close() + for name, val in saved.items(): + if val is None: + sys.modules.pop(name, None) + else: + sys.modules[name] = val + got = [json.loads(ln) for ln in out.splitlines()] + self.assertEqual(got[0]["result"], {}) + self.assertIsNone(browser.frame_sink) + + def test_frame_queue_real_and_fallback(self): + q = playwright_mcp._frame_queue() + q.put("frame-1") + self.assertEqual(q.get_nowait(), "frame-1") + with self.assertRaises(Exception): + q.get_nowait() + saved = {} + for name in ("playwright_remote", "cloud.playwright_remote"): + saved[name] = sys.modules.get(name) + sys.modules[name] = None + try: + q2 = playwright_mcp._frame_queue() + finally: + for name, val in saved.items(): + if val is None: + sys.modules.pop(name, None) + else: + sys.modules[name] = val + q2.put("x") + self.assertEqual(q2.get_nowait(), "x") + with self.assertRaises(IndexError): + q2.get_nowait() + + def test_flat_copy_fallback_import(self): + # 封堵 cloud.playwright_browser 后重新导入：走 sys.path 回落。 + orig_browser = sys.modules.get("cloud.playwright_browser") + orig_mcp = sys.modules.get("cloud.playwright_mcp") + sys.modules.pop("cloud.playwright_mcp", None) + sys.modules["cloud.playwright_browser"] = None + try: + mod = importlib.import_module("cloud.playwright_mcp") + self.assertEqual(mod.SERVER_NAME, "playwright") + self.assertIsNotNone(mod.Browser) + finally: + sys.modules.pop("playwright_browser", None) + if orig_browser is not None: + sys.modules["cloud.playwright_browser"] = orig_browser + else: + sys.modules.pop("cloud.playwright_browser", None) + if orig_mcp is not None: + sys.modules["cloud.playwright_mcp"] = orig_mcp + else: + sys.modules.pop("cloud.playwright_mcp", None) + cloud_dir = str(Path(__file__).resolve().parents[1]) + while cloud_dir in sys.path: + sys.path.remove(cloud_dir) + + +class ImageGenMcpTest(unittest.TestCase): + def test_handle_request_dispatch_and_errors(self): + calls = {} + + def fake_generate(prompt, output_filename=None, size="1024x1024", + model=None): + calls["generate"] = (prompt, output_filename, size, model) + return {"path": "/tmp/a.png", "bytes": 10} + + def fake_edit(prompt, input_filename, output_filename=None, + size="1024x1024", model=None): + calls["edit"] = (prompt, input_filename, output_filename, size, + model) + return {"path": "/tmp/b.png", "bytes": 20} + + self.assertEqual(imagegen_mcp.handle_request("x")["error"]["code"], + -32700) + self.assertIsNone(imagegen_mcp.handle_request( + {"jsonrpc": "2.0", "method": "initialize"})) + self.assertIsNone(imagegen_mcp.handle_request( + {"jsonrpc": "2.0", "method": "notifications/cancelled"})) + self.assertEqual(imagegen_mcp.handle_request( + {"jsonrpc": "2.0", "id": 1, "method": "ping"})["result"], {}) + self.assertIsNone(imagegen_mcp.handle_request( + {"jsonrpc": "2.0", "method": "ping"})) + resp = imagegen_mcp.handle_request( + {"jsonrpc": "2.0", "id": 2, "method": "tools/list"}) + self.assertEqual([t["name"] for t in resp["result"]["tools"]], + ["image_generate", "image_edit"]) + self.assertIsNone(imagegen_mcp.handle_request( + {"jsonrpc": "2.0", "method": "tools/list"})) + resp = imagegen_mcp.handle_request( + {"jsonrpc": "2.0", "id": 3, "method": "tools/call", + "params": {"name": "image_generate", + "arguments": {"prompt": "a cat"}}}, + generate_fn=fake_generate, edit_fn=fake_edit) + self.assertFalse(resp["result"]["isError"]) + self.assertEqual(calls["generate"], + ("a cat", None, "1024x1024", None)) + resp = imagegen_mcp.handle_request( + {"jsonrpc": "2.0", "id": 4, "method": "tools/call", + "params": {"name": "image_edit", + "arguments": {"prompt": "redder", + "input_filename": "a.png", + "size": "1024x1536", + "model": "m1"}}}, + generate_fn=fake_generate, edit_fn=fake_edit) + self.assertEqual(calls["edit"], + ("redder", "a.png", None, "1024x1536", "m1")) + resp = imagegen_mcp.handle_request( + {"jsonrpc": "2.0", "id": 5, "method": "tools/call", + "params": {"name": "nope", "arguments": {}}}, + generate_fn=fake_generate, edit_fn=fake_edit) + self.assertEqual(resp["error"]["code"], -32602) + self.assertIsNone(imagegen_mcp.handle_request( + {"jsonrpc": "2.0", "method": "tools/call", + "params": {"name": "image_generate", + "arguments": {"prompt": "x"}}}, + generate_fn=fake_generate, edit_fn=fake_edit)) + self.assertIsNone(imagegen_mcp.handle_request( + {"jsonrpc": "2.0", "method": "nope"})) + resp = imagegen_mcp.handle_request( + {"jsonrpc": "2.0", "id": 6, "method": "bogus"}) + self.assertEqual(resp["error"]["code"], -32601) + + def boom(*args, **kwargs): + raise RuntimeError("relay down") + + resp = imagegen_mcp.handle_request( + {"jsonrpc": "2.0", "id": 7, "method": "tools/call", + "params": {"name": "image_generate", + "arguments": {"prompt": "x"}}}, + generate_fn=boom, edit_fn=fake_edit) + self.assertTrue(resp["result"]["isError"]) + self.assertIn("relay down", resp["result"]["content"][0]["text"]) + + def test_main_loop_with_fake_generate(self): + old_generate = imagegen_mcp.generate + imagegen_mcp.generate = lambda prompt, output_filename=None, \ + size="1024x1024", model=None: {"path": "/tmp/g.png"} + old_in, old_out = sys.stdin, sys.stdout + sys.stdin = io.StringIO("\nbroken{\n" + json.dumps( + {"jsonrpc": "2.0", "id": 1, "method": "tools/call", + "params": {"name": "image_generate", + "arguments": {"prompt": "sunset"}}}) + "\n" + json.dumps( + {"jsonrpc": "2.0", "id": 2, "method": "initialize", + "params": ["x"]}) + "\n") + sys.stdout = io.StringIO() + try: + imagegen_mcp.main() + out = sys.stdout.getvalue() + finally: + sys.stdin, sys.stdout = old_in, old_out + imagegen_mcp.generate = old_generate + got = [json.loads(ln) for ln in out.splitlines()] + self.assertEqual(got[0]["error"]["code"], -32700) + self.assertFalse(got[1]["result"]["isError"]) + self.assertEqual(got[2]["error"]["code"], -32603) + + def test_flat_copy_fallback_import(self): + orig_backend = sys.modules.get("cloud.imagegen") + orig_mcp = sys.modules.get("cloud.imagegen_mcp") + sys.modules.pop("cloud.imagegen_mcp", None) + sys.modules["cloud.imagegen"] = None + try: + mod = importlib.import_module("cloud.imagegen_mcp") + self.assertEqual(mod.SERVER_NAME, "imagegen") + finally: + sys.modules.pop("imagegen", None) + if orig_backend is not None: + sys.modules["cloud.imagegen"] = orig_backend + else: + sys.modules.pop("cloud.imagegen", None) + if orig_mcp is not None: + sys.modules["cloud.imagegen_mcp"] = orig_mcp + else: + sys.modules.pop("cloud.imagegen_mcp", None) + cloud_dir = str(Path(__file__).resolve().parents[1]) + while cloud_dir in sys.path: + sys.path.remove(cloud_dir) + + +class WebSearchMcpTest(unittest.TestCase): + def test_handle_request_dispatch_and_errors(self): + calls = {} + + def fake_search(query, limit=5, provider="auto"): + calls["search"] = (query, limit, provider) + return {"results": [{"title": "t", "url": "http://x/"}]} + + def fake_fetch(url, char_limit=15000): + calls["fetch"] = (url, char_limit) + return {"text": "hello"} + + self.assertEqual(websearch_mcp.handle_request(42)["error"]["code"], + -32700) + self.assertIsNone(websearch_mcp.handle_request( + {"jsonrpc": "2.0", "method": "initialize"})) + self.assertIsNone(websearch_mcp.handle_request( + {"jsonrpc": "2.0", "method": "notifications/initialized"})) + self.assertEqual(websearch_mcp.handle_request( + {"jsonrpc": "2.0", "id": 1, "method": "ping"})["result"], {}) + self.assertIsNone(websearch_mcp.handle_request( + {"jsonrpc": "2.0", "method": "ping"})) + resp = websearch_mcp.handle_request( + {"jsonrpc": "2.0", "id": 2, "method": "tools/list"}) + self.assertEqual([t["name"] for t in resp["result"]["tools"]], + ["web_search", "web_fetch"]) + self.assertIsNone(websearch_mcp.handle_request( + {"jsonrpc": "2.0", "method": "tools/list"})) + resp = websearch_mcp.handle_request( + {"jsonrpc": "2.0", "id": 3, "method": "tools/call", + "params": {"name": "web_search", + "arguments": {"query": "muse"}}}, + search_fn=fake_search, fetch_fn=fake_fetch) + self.assertEqual(calls["search"], ("muse", 5, "auto")) + resp = websearch_mcp.handle_request( + {"jsonrpc": "2.0", "id": 4, "method": "tools/call", + "params": {"name": "web_fetch", + "arguments": {"url": "http://x/", + "char_limit": 3000}}}, + search_fn=fake_search, fetch_fn=fake_fetch) + self.assertEqual(calls["fetch"], ("http://x/", 3000)) + resp = websearch_mcp.handle_request( + {"jsonrpc": "2.0", "id": 5, "method": "tools/call", + "params": {"name": "nope", "arguments": {}}}, + search_fn=fake_search, fetch_fn=fake_fetch) + self.assertEqual(resp["error"]["code"], -32602) + self.assertIsNone(websearch_mcp.handle_request( + {"jsonrpc": "2.0", "method": "tools/call", + "params": {"name": "web_search", + "arguments": {"query": "x"}}}, + search_fn=fake_search, fetch_fn=fake_fetch)) + self.assertIsNone(websearch_mcp.handle_request( + {"jsonrpc": "2.0", "method": "bogus"})) + resp = websearch_mcp.handle_request( + {"jsonrpc": "2.0", "id": 6, "method": "bogus"}) + self.assertEqual(resp["error"]["code"], -32601) + + def boom(*args, **kwargs): + raise RuntimeError("provider down") + + resp = websearch_mcp.handle_request( + {"jsonrpc": "2.0", "id": 7, "method": "tools/call", + "params": {"name": "web_search", + "arguments": {"query": "x"}}}, + search_fn=boom, fetch_fn=fake_fetch) + self.assertTrue(resp["result"]["isError"]) + self.assertIn("provider down", resp["result"]["content"][0]["text"]) + + def test_main_loop_with_fake_search(self): + old_search = websearch_mcp.search + websearch_mcp.search = lambda query, limit=5, provider="auto": \ + {"results": []} + old_in, old_out = sys.stdin, sys.stdout + sys.stdin = io.StringIO("\nbroken{\n" + json.dumps( + {"jsonrpc": "2.0", "id": 1, "method": "tools/call", + "params": {"name": "web_search", + "arguments": {"query": "q"}}}) + "\n" + json.dumps( + {"jsonrpc": "2.0", "id": 2, "method": "initialize", + "params": ["x"]}) + "\n") + sys.stdout = io.StringIO() + try: + websearch_mcp.main() + out = sys.stdout.getvalue() + finally: + sys.stdin, sys.stdout = old_in, old_out + websearch_mcp.search = old_search + got = [json.loads(ln) for ln in out.splitlines()] + self.assertEqual(got[0]["error"]["code"], -32700) + self.assertFalse(got[1]["result"]["isError"]) + self.assertEqual(got[2]["error"]["code"], -32603) + + def test_flat_copy_fallback_import(self): + orig_backend = sys.modules.get("cloud.websearch") + orig_mcp = sys.modules.get("cloud.websearch_mcp") + sys.modules.pop("cloud.websearch_mcp", None) + sys.modules["cloud.websearch"] = None + try: + mod = importlib.import_module("cloud.websearch_mcp") + self.assertEqual(mod.SERVER_NAME, "websearch") + finally: + sys.modules.pop("websearch", None) + if orig_backend is not None: + sys.modules["cloud.websearch"] = orig_backend + else: + sys.modules.pop("cloud.websearch", None) + if orig_mcp is not None: + sys.modules["cloud.websearch_mcp"] = orig_mcp + else: + sys.modules.pop("cloud.websearch_mcp", None) + cloud_dir = str(Path(__file__).resolve().parents[1]) + while cloud_dir in sys.path: + sys.path.remove(cloud_dir) + + +if __name__ == "__main__": + unittest.main() diff --git a/deploy/tests/test_coverage85_deploy.py b/deploy/tests/test_coverage85_deploy.py new file mode 100644 index 0000000..ef3a5da --- /dev/null +++ b/deploy/tests/test_coverage85_deploy.py @@ -0,0 +1,399 @@ +"""覆盖率 78%->85% 第 2 步：deploy.py 的剩余缺口真实行为测试。 + +只测本地逻辑与参数分发：真实临时 git 仓库建包、真实 tarball +读清单；凡是触网/触主机的步骤一律用 subprocess/ssh_cmd/transfer +替身在调用边界拦住（替身只决定命令返回值，不替换被测函数 +本身），绝不连真实主机、不跑真实部署。""" +import io +import os +import subprocess +import sys +import tarfile +import tempfile +import unittest +from contextlib import redirect_stderr, redirect_stdout +from types import SimpleNamespace +from unittest import mock + +REPO_ROOT = os.path.dirname(os.path.dirname(os.path.dirname( + os.path.abspath(__file__)))) +sys.path.insert(0, os.path.join(REPO_ROOT, "deploy")) +import deploy as deploy_mod # noqa: E402 + + +def _make_git_repo(path): + def git(*args): + subprocess.run(["git", "-C", path] + list(args), + check=True, capture_output=True, text=True) + + subprocess.run(["git", "init", "-q", path], check=True, + capture_output=True) + git("config", "user.email", "test@test.local") + git("config", "user.name", "Test") + with open(os.path.join(path, "hello.txt"), "w") as f: + f.write("hello\n") + git("add", ".") + git("commit", "-qm", "init") + head = subprocess.run(["git", "-C", path, "rev-parse", "HEAD"], + check=True, capture_output=True, + text=True).stdout.strip() + git("update-ref", "refs/remotes/origin/main", head) + return head + + +def _cp(args=(), returncode=0, **kwargs): + return subprocess.CompletedProcess(args=list(args), + returncode=returncode, + stdout="", stderr="") + + +class DeployCleanTreeTest(unittest.TestCase): + def test_clean_and_dirty_and_allow_dirty(self): + with tempfile.TemporaryDirectory() as tmp: + _make_git_repo(tmp) + old_root = deploy_mod.REPO_ROOT + deploy_mod.REPO_ROOT = tmp + try: + deploy_mod.check_clean_tree(False) # 干净：不退出 + with open(os.path.join(tmp, "hello.txt"), "a") as f: + f.write("dirty\n") + with self.assertRaises(SystemExit) as ctx: + deploy_mod.check_clean_tree(False) + self.assertEqual(ctx.exception.code, 2) + out = io.StringIO() + with redirect_stdout(out): + deploy_mod.check_clean_tree(True) # 只告警不退出 + self.assertIn("警告", out.getvalue()) + subprocess.run(["git", "-C", tmp, "add", "."], + check=True, capture_output=True) + with self.assertRaises(SystemExit): + deploy_mod.check_clean_tree(False) # 暂存区脏也拦 + finally: + deploy_mod.REPO_ROOT = old_root + + def test_run_failure_exits(self): + with self.assertRaises(SystemExit) as ctx: + deploy_mod.run(["false"], check=True) + self.assertEqual(ctx.exception.code, 1) + r = deploy_mod.run(["false"], check=False) + self.assertNotEqual(r.returncode, 0) + + +class DeployRemoteHelpersTest(unittest.TestCase): + def test_remote_sha_none_on_rc(self): + with mock.patch.object(deploy_mod.subprocess, "run", + return_value=_cp(returncode=1)): + self.assertIsNone(deploy_mod.remote_sha("h", "u", "/tmp/x")) + + def test_remote_rm_issues_rm_command(self): + seen = [] + + def fake_run(cmd, **kwargs): + seen.append(cmd) + return _cp(cmd) + + with mock.patch.object( + deploy_mod, "ssh_cmd", + side_effect=lambda h, u, c: ["ssh", "dummy", c]): + with mock.patch.object(deploy_mod.subprocess, "run", + side_effect=fake_run): + deploy_mod.remote_rm("h", "u", "/tmp/x.tar.gz") + self.assertTrue(any(c[-1] == "rm -f /tmp/x.tar.gz" for c in seen)) + + def test_build_release_archive_failure_exits(self): + with tempfile.TemporaryDirectory() as tmp: + _make_git_repo(tmp) + old_root = deploy_mod.REPO_ROOT + deploy_mod.REPO_ROOT = tmp + try: + # rev-parse 能过、archive 必败的 ref：一个 blob 的 sha + #（rev-parse 成功，git archive 拒绝非提交对象）。 + blob = subprocess.run( + ["git", "-C", tmp, "hash-object", "-w", "--stdin"], + input="blob-content\n", check=True, capture_output=True, + text=True).stdout.strip() + self.assertEqual(len(blob), 40) + with self.assertRaises(SystemExit) as ctx: + deploy_mod.build_release(blob, tmp) + self.assertEqual(ctx.exception.code, 1) + finally: + deploy_mod.REPO_ROOT = old_root + + +class DeployTransferTest(unittest.TestCase): + def _helpers(self, key=True, plain=True): + def fake_helper(name): + if name == "chunk_send_key_explicit.py" and key: + return "/helpers/chunk_send_key_explicit.py" + if name == "chunk_send.py" and plain: + return "/helpers/chunk_send.py" + return None + + return fake_helper + + def test_all_attempts_fail_exits(self): + def fake_run(cmd, **kwargs): + if cmd and cmd[0] == "scp": + return subprocess.CompletedProcess( + args=cmd, returncode=1, stdout="", stderr="reset") + return subprocess.CompletedProcess( + args=cmd, returncode=1, stdout="", stderr="chunk fail") + + with mock.patch.object(deploy_mod, "helper_path", + side_effect=self._helpers()): + with mock.patch.object(deploy_mod.subprocess, "run", + side_effect=fake_run): + with mock.patch.dict(os.environ, + {"SSH_PASSWORD": "x"}): + with self.assertRaises(SystemExit) as ctx: + deploy_mod.transfer("h", "u", "/tmp/a", "/tmp/b", + "0" * 64) + self.assertEqual(ctx.exception.code, 1) + + def test_chunk_decode_failure_continues_then_exits(self): + def fake_run(cmd, **kwargs): + if cmd and cmd[0] == "scp": + return subprocess.CompletedProcess( + args=cmd, returncode=1, stdout="", stderr="reset") + if cmd and cmd[0] == "python3": + return _cp(cmd) # 分片脚本本身成功 + return subprocess.CompletedProcess( + args=cmd, returncode=1, stdout="", stderr="decode fail") + + with mock.patch.object(deploy_mod, "helper_path", + side_effect=self._helpers()): + with mock.patch.object(deploy_mod, "ssh_cmd", + return_value=["ssh", "dummy"]): + with mock.patch.object(deploy_mod.subprocess, "run", + side_effect=fake_run): + with mock.patch.dict(os.environ, + {"SSH_PASSWORD": "x"}): + with self.assertRaises(SystemExit): + deploy_mod.transfer("h", "u", "/tmp/a", + "/tmp/b", "0" * 64) + + def test_sha_mismatch_removes_and_exits(self): + removed = [] + + def fake_run(cmd, **kwargs): + return _cp(cmd) # scp 成功 + + with mock.patch.object(deploy_mod, "helper_path", + side_effect=self._helpers(key=False, + plain=False)): + with mock.patch.object(deploy_mod.subprocess, "run", + side_effect=fake_run): + with mock.patch.object( + deploy_mod, "remote_sha", + return_value="1" * 64): + with mock.patch.object( + deploy_mod, "remote_rm", + side_effect=lambda h, u, p: removed.append(p)): + with self.assertRaises(SystemExit): + deploy_mod.transfer("h", "u", "/tmp/a", + "/tmp/b", "0" * 64) + self.assertEqual(removed, ["/tmp/b"]) + + +class DeployViaGitTest(unittest.TestCase): + def _args(self, **over): + base = dict(tarball=None, build_only=False, host="h.example", + user="root", dry_run=False, + repo="git@github.com:openqj/codex-app-server.git", + git_key="/root/.ssh/id_codex_github", + allow_dirty=True, skip_tests=True, ref="HEAD") + base.update(over) + return SimpleNamespace(**base) + + def test_conflicts_and_guards_exit_2(self): + for over, in_msg in [ + (dict(tarball="/tmp/x.tgz"), "tarball"), + (dict(build_only=True), "build-only"), + (dict(host=None, dry_run=False), "host"), + (dict(repo="https://github.com/x/y.git"), "SSH")]: + with self.subTest(over=over): + err = io.StringIO() + with redirect_stderr(err): + with self.assertRaises(SystemExit) as ctx: + deploy_mod.deploy_via_git(self._args(**over)) + self.assertEqual(ctx.exception.code, 2) + + def test_dry_run_success(self): + with tempfile.TemporaryDirectory() as tmp: + head = _make_git_repo(tmp) + old_root = deploy_mod.REPO_ROOT + deploy_mod.REPO_ROOT = tmp + try: + out = io.StringIO() + with redirect_stdout(out): + rc = deploy_mod.deploy_via_git( + self._args(dry_run=True)) + self.assertEqual(rc, 0) + self.assertIn(head, out.getvalue()) + finally: + deploy_mod.REPO_ROOT = old_root + + def _ssh_fake(self, head, script_rc=0, script_out=None, live=None): + real_run = subprocess.run + + def fake_run(cmd, **kwargs): + if cmd and cmd[0] == "git": + return real_run(cmd, **kwargs) + last = cmd[-1] if cmd else "" + if "cat /opt/codex-app-server/DEPLOYED_COMMIT" in last: + return subprocess.CompletedProcess( + args=cmd, returncode=0, + stdout=(live if live is not None else head) + "\n", + stderr="") + if last.startswith("sh "): + return subprocess.CompletedProcess( + args=cmd, returncode=script_rc, + stdout=(script_out if script_out is not None + else "RELEASE_OK commit=%s" % head), + stderr="remote-err" if script_rc != 0 else "") + return _cp(cmd) + + return fake_run + + def test_non_dry_run_success_and_failures(self): + with tempfile.TemporaryDirectory() as tmp: + head = _make_git_repo(tmp) + old_root = deploy_mod.REPO_ROOT + deploy_mod.REPO_ROOT = tmp + try: + transfers = [] + with mock.patch.object( + deploy_mod, "ssh_cmd", + side_effect=lambda h, u, c: ["ssh", "dummy", c]): + with mock.patch.object( + deploy_mod, "transfer", + side_effect=lambda *a: transfers.append(a)): + with mock.patch.object( + deploy_mod.subprocess, "run", + side_effect=self._ssh_fake(head)): + rc = deploy_mod.deploy_via_git(self._args()) + self.assertEqual(rc, 0) + self.assertEqual(len(transfers), 2) + # 主机脚本失败 -> exit 1 + with mock.patch.object( + deploy_mod, "ssh_cmd", + side_effect=lambda h, u, c: ["ssh", "dummy", c]): + with mock.patch.object(deploy_mod, "transfer"): + with mock.patch.object( + deploy_mod.subprocess, "run", + side_effect=self._ssh_fake( + head, script_rc=1, script_out="boom")): + with self.assertRaises(SystemExit): + deploy_mod.deploy_via_git(self._args()) + # live 二次核对不符 -> exit 1 + with mock.patch.object( + deploy_mod, "ssh_cmd", + side_effect=lambda h, u, c: ["ssh", "dummy", c]): + with mock.patch.object(deploy_mod, "transfer"): + with mock.patch.object( + deploy_mod.subprocess, "run", + side_effect=self._ssh_fake( + head, live="0" * 40)): + with self.assertRaises(SystemExit): + deploy_mod.deploy_via_git(self._args()) + finally: + deploy_mod.REPO_ROOT = old_root + + +class DeployMainTest(unittest.TestCase): + def _tarball(self, tmp): + head = _make_git_repo(tmp) + old_root = deploy_mod.REPO_ROOT + deploy_mod.REPO_ROOT = tmp + try: + tarball, sha, commit = deploy_mod.build_release("HEAD", tmp) + finally: + deploy_mod.REPO_ROOT = old_root + self.assertEqual(commit, head) + return tarball, head + + def test_no_host_errors(self): + with self.assertRaises(SystemExit) as ctx: + deploy_mod.main([]) + self.assertEqual(ctx.exception.code, 2) + + def test_tarball_mismatch_and_dry_run(self): + with tempfile.TemporaryDirectory() as tmp: + tarball, head = self._tarball(tmp) + with self.assertRaises(SystemExit) as ctx: + deploy_mod.main(["h.example", "--tarball", tarball, + "--commit", "0" * 40]) + self.assertEqual(ctx.exception.code, 2) + rc = deploy_mod.main(["h.example", "--tarball", tarball, + "--dry-run"]) + self.assertEqual(rc, 0) + + def test_build_only(self): + with tempfile.TemporaryDirectory() as tmp: + out = io.StringIO() + with redirect_stdout(out): + rc = deploy_mod.main(["--build-only", "--allow-dirty", + "--skip-tests", "--out-dir", tmp]) + self.assertEqual(rc, 0) + self.assertIn("仅构建", out.getvalue()) + self.assertTrue(any(n.endswith(".tar.gz") + for n in os.listdir(tmp))) + + def test_git_dry_run_via_main(self): + with tempfile.TemporaryDirectory() as tmp: + head = _make_git_repo(tmp) + old_root = deploy_mod.REPO_ROOT + deploy_mod.REPO_ROOT = tmp + try: + rc = deploy_mod.main(["--source", "git", "--dry-run", + "--skip-tests", "--allow-dirty"]) + self.assertEqual(rc, 0) + finally: + deploy_mod.REPO_ROOT = old_root + + def _ssh_fake_main(self, head, script_rc=0, live=None): + def fake_run(cmd, **kwargs): + last = cmd[-1] if cmd else "" + if "cat /opt/codex-app-server/DEPLOYED_COMMIT" in last: + return subprocess.CompletedProcess( + args=cmd, returncode=0, + stdout=(live if live is not None else head) + "\n", + stderr="") + if last.startswith("sh "): + return subprocess.CompletedProcess( + args=cmd, returncode=script_rc, + stdout="RELEASE_OK commit=%s" % head if script_rc == 0 + else "boom", + stderr="remote-err" if script_rc != 0 else "") + return _cp(cmd) + + return fake_run + + def test_tar_deploy_success_and_failures(self): + with tempfile.TemporaryDirectory() as tmp: + tarball, head = self._tarball(tmp) + for script_rc, live, ok in [(0, None, True), + (1, None, False), + (0, "0" * 40, False)]: + with self.subTest(script_rc=script_rc, live=live): + with mock.patch.object( + deploy_mod, "ssh_cmd", + side_effect=lambda h, u, c: ["ssh", "dummy", c]): + with mock.patch.object(deploy_mod, "transfer"): + with mock.patch.object( + deploy_mod.subprocess, "run", + side_effect=self._ssh_fake_main( + head, script_rc, live)): + if ok: + rc = deploy_mod.main( + ["h.example", "--tarball", tarball]) + self.assertEqual(rc, 0) + else: + with self.assertRaises(SystemExit): + deploy_mod.main( + ["h.example", "--tarball", + tarball]) + + +if __name__ == "__main__": + unittest.main() diff --git a/deploy/tests/test_coverage85_verify_multiuser.py b/deploy/tests/test_coverage85_verify_multiuser.py new file mode 100644 index 0000000..db4412f --- /dev/null +++ b/deploy/tests/test_coverage85_verify_multiuser.py @@ -0,0 +1,173 @@ +"""覆盖率 78%->85% 第 2 步：deploy/verify-multiuser.py 的 main()。 + +main() 是对真实控制面的多用户验证：本文件用本地临时 SQLite + +本地临时端口的真实控制面（test_server.py 同款 create_server）跑 +完整流程，绝不连生产、不建生产任务；失败分支用本地 stub HTTP +服务与死端口驱动。临时库里的 multiuser-test-* 行跑完必须清零。 +""" + +import http.server +import importlib.util +import io +import json +import os +import sqlite3 +import sys +import tempfile +import threading +import unittest +from contextlib import redirect_stdout +from pathlib import Path + +REPO_ROOT = Path(__file__).resolve().parents[2] +sys.path.insert(0, str(REPO_ROOT)) +_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) + +from cloud.server import create_server, init_db # noqa: E402 + + +class StubHandler(http.server.BaseHTTPRequestHandler): + mode = "wrong-id" + + def _send(self, code, body): + data = json.dumps(body).encode() + self.send_response(code) + self.send_header("Content-Type", "application/json") + self.send_header("Content-Length", str(len(data))) + self.end_headers() + self.wfile.write(data) + + def do_POST(self): + length = int(self.headers.get("Content-Length") or 0) + self.rfile.read(length) + if self.mode == "wrong-id": + self._send(201, {"id": "wrong"}) + else: + self._send(500, {"error": "boom"}) + + def do_GET(self): + self._send(404, {"error": "not found"}) + + def log_message(self, *args): + pass + + +def _start_stub(mode): + StubHandler.mode = mode + srv = http.server.HTTPServer(("127.0.0.1", 0), StubHandler) + threading.Thread(target=srv.serve_forever, daemon=True).start() + return srv + + +class VerifyMultiuserMainTest(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.db_path = os.path.join(self.tmp.name, "control.db") + self._old_argv = sys.argv + + def tearDown(self): + sys.argv = self._old_argv + self.tmp.cleanup() + + def _run_main(self, base): + vm.RESULTS.clear() + sys.argv = ["verify-multiuser.py", "--base", base, + "--db", self.db_path] + out = io.StringIO() + with redirect_stdout(out): + rc = vm.main() + return rc, out.getvalue(), list(vm.RESULTS) + + def _leftover_counts(self): + conn = sqlite3.connect(self.db_path) + try: + counts = {} + for table, col in (("users", "username"), + ("projects", "id"), + ("tasks", "project_id")): + counts[table] = conn.execute( + "SELECT COUNT(*) FROM %s WHERE %s LIKE ?" + % (table, col), ("multiuser-test-%",)).fetchone()[0] + return counts + finally: + conn.close() + + def test_happy_path_against_real_local_server(self): + server = create_server(self.db_path, "127.0.0.1", 0) + port = server.server_address[1] + thread = threading.Thread(target=server.serve_forever, daemon=True) + thread.start() + try: + rc, out, results = self._run_main( + "http://127.0.0.1:%d" % port) + finally: + server.shutdown() + server.server_close() + thread.join(timeout=5) + self.assertEqual(rc, 0, out) + self.assertEqual(len(results), 13, results) + self.assertTrue(all(ok for _, ok in results), results) + self.assertIn("13/13", out) + self.assertEqual(self._leftover_counts(), + {"users": 0, "projects": 0, "tasks": 0}) + + def test_dead_base_fails_early(self): + init_db(self.db_path) + rc, out, results = self._run_main("http://127.0.0.1:1") + self.assertEqual(rc, 1) + self.assertEqual(len(results), 2) + self.assertFalse(any(ok for _, ok in results)) + self.assertIn("项目创建失败", out) + self.assertEqual(self._leftover_counts(), + {"users": 0, "projects": 0, "tasks": 0}) + + def test_cleanup_error_reports_leftover(self): + server = create_server(self.db_path, "127.0.0.1", 0) + port = server.server_address[1] + thread = threading.Thread(target=server.serve_forever, daemon=True) + thread.start() + try: + # 删用户时触发器直接 ABORT：清理抛错 -> leftover -> 返回 1， + # 但 13 项检查本身全部通过（区分“检查失败”与“清理失败”）。 + conn = sqlite3.connect(self.db_path) + try: + conn.execute( + "CREATE TRIGGER no_user_delete BEFORE DELETE ON users " + "BEGIN SELECT RAISE(ABORT, 'trigger blocks delete'); END") + conn.commit() + finally: + conn.close() + rc, out, results = self._run_main( + "http://127.0.0.1:%d" % port) + finally: + server.shutdown() + server.server_close() + thread.join(timeout=5) + self.assertEqual(rc, 1) + self.assertTrue(all(ok for _, ok in results), results) + self.assertIn("清理未完成", out) + self.assertIn("cleanup-error", out) + + def test_stub_project_responses_fail_early(self): + init_db(self.db_path) + for mode, want in [("wrong-id", [True, True]), + ("error-500", [False, False])]: + with self.subTest(mode=mode): + stub = _start_stub(mode) + try: + rc, out, results = self._run_main( + "http://127.0.0.1:%d" % stub.server_address[1]) + finally: + stub.shutdown() + stub.server_close() + self.assertEqual(rc, 1) + self.assertEqual([ok for _, ok in results], want) + self.assertIn("项目创建失败", out) + + +if __name__ == "__main__": + unittest.main()