"""覆盖率 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()