codex-app-server/cloud/tests/test_coverage85_mcp.py

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