host
6bcdeb8da1
根因（2026-10-30 活体实证）：#181 的 .browser-remote.sock 假设
宿主能连容器内 socket，但生产 Docker 任务走 runsc/gVisor，
其 gofer 挂载下容器内建的 socket 只在容器内可见，宿主目录
里只有普通文件（browser-live.png 一直能透传）。后果：server
代理对真实任务一律 503，会话根本没活过。
修法：传输层改四个普通文件（worktree 内，原子写 + 追加日志）：
- .browser-remote-state.json：MCP 每轮重写，mtime 即存活信号，
server 判 30 秒无心跳为会话死亡（503）；
- .browser-remote-frame.jpg：最新 screencast 帧（base64 文本）；
- .browser-remote-input.jsonl：server 追加，MCP 按偏移量消费
完整行，执行时再判 holder，半行等下一轮；
- .browser-remote-control.json：{holder, seq}，MCP 首轮即应用，
接管可先于浏览器启动。
- holder 竞态：control 写下后 state 文件约 50ms 才刷新，server
的 input/stream 改判 effective_holder（控制文件 mtime 新者胜），
否则接管后立刻点会误 409。
- 线程契约与控制权模型不变；MCP 主循环仍是唯一触页线程。
- 播种刷新（#181）让老租户自动拿到新版脚本。
测试：socket 测试改文件桥测试（桥轮询/偏移量/半行/陈旧态/
effective_holder + server 代理全链），全量 1357 项通过（skipped 4）。
364 lines
12 KiB
Python
364 lines
12 KiB
Python
"""Tests for the Playwright browser MCP (fake browser only).
|
|
|
|
No real Chromium is launched here: CI has no browsers, so the
|
|
Browser class is exercised through a fake sync_playwright
|
|
factory, and the MCP server through a fake browser object.
|
|
Real-browser proof lives in the production verification task.
|
|
"""
|
|
|
|
import json
|
|
import os
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
|
|
sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
|
|
|
|
from playwright_browser import Browser, BrowserError # noqa: E402
|
|
from playwright_mcp import handle_request # noqa: E402
|
|
|
|
|
|
class _FakeLocator:
|
|
def __init__(self, text):
|
|
self._text = text
|
|
|
|
def inner_text(self, timeout=None):
|
|
return self._text
|
|
|
|
@property
|
|
def first(self):
|
|
return self
|
|
|
|
def click(self, timeout=None):
|
|
pass
|
|
|
|
|
|
class _FakeMouse:
|
|
def __init__(self, page):
|
|
self._page = page
|
|
|
|
def click(self, x, y):
|
|
self._page.calls.append(("mouse_click", x, y))
|
|
|
|
def move(self, x, y):
|
|
self._page.calls.append(("mouse_move", x, y))
|
|
|
|
|
|
class _FakeKeyboard:
|
|
def __init__(self, page):
|
|
self._page = page
|
|
|
|
def press(self, key):
|
|
self._page.calls.append(("key_press", key))
|
|
|
|
def type(self, text):
|
|
self._page.calls.append(("key_type", text))
|
|
|
|
|
|
class _FakeCDP:
|
|
def __init__(self):
|
|
self.sent = []
|
|
self.handlers = {}
|
|
|
|
def on(self, event, handler):
|
|
self.handlers[event] = handler
|
|
|
|
def send(self, method, params=None):
|
|
self.sent.append((method, params or {}))
|
|
|
|
|
|
class _FakeContext:
|
|
def __init__(self, cdp):
|
|
self.cdp = cdp
|
|
|
|
def new_cdp_session(self, page):
|
|
return self.cdp
|
|
|
|
|
|
class _FakePage:
|
|
def __init__(self):
|
|
self.url = "https://example.com/"
|
|
self.calls = []
|
|
self.shot_path = None
|
|
self.mouse = _FakeMouse(self)
|
|
self.keyboard = _FakeKeyboard(self)
|
|
self.cdp = _FakeCDP()
|
|
self.context = _FakeContext(self.cdp)
|
|
|
|
def title(self):
|
|
return "Example Domain"
|
|
|
|
def goto(self, url, timeout=None, wait_until=None):
|
|
self.calls.append(("goto", url))
|
|
self.url = url
|
|
|
|
def locator(self, selector):
|
|
self.calls.append(("locator", selector))
|
|
return _FakeLocator("hello " * 3000)
|
|
|
|
def click(self, selector, timeout=None):
|
|
self.calls.append(("click", selector))
|
|
|
|
def fill(self, selector, text, timeout=None):
|
|
self.calls.append(("fill", selector, text))
|
|
|
|
def evaluate(self, expression):
|
|
self.calls.append(("evaluate", expression))
|
|
if "querySelectorAll" in expression:
|
|
return [{"tag": "button", "role": "", "name": "GO",
|
|
"path": "button"},
|
|
{"tag": "a", "role": "link", "name": "Home",
|
|
"path": "a"}]
|
|
return 42
|
|
|
|
def screenshot(self, path=None):
|
|
self.shot_path = path
|
|
with open(path, "wb") as fh:
|
|
fh.write(b"\x89PNG\r\n\x1a\n" + b"\x00" * 100)
|
|
|
|
def close(self):
|
|
pass
|
|
|
|
|
|
class _FakeBrowserHandle:
|
|
def __init__(self, page):
|
|
self._page = page
|
|
|
|
def new_page(self, viewport=None):
|
|
return self._page
|
|
|
|
def close(self):
|
|
pass
|
|
|
|
|
|
class _FakePW:
|
|
def __init__(self, page):
|
|
self._page = page
|
|
self.launch_kwargs = None
|
|
|
|
def start(self):
|
|
return self
|
|
|
|
@property
|
|
def chromium(self):
|
|
outer = self
|
|
|
|
class _Chromium:
|
|
def launch(self, **kwargs):
|
|
outer.launch_kwargs = kwargs
|
|
return _FakeBrowserHandle(outer._page)
|
|
|
|
return _Chromium()
|
|
|
|
def stop(self):
|
|
pass
|
|
|
|
|
|
class BrowserTest(unittest.TestCase):
|
|
def setUp(self):
|
|
self._tmp = tempfile.TemporaryDirectory()
|
|
self.page = _FakePage()
|
|
self.browser = Browser(cwd=self._tmp.name,
|
|
proxy="http://proxy:8888",
|
|
browser_factory=lambda: _FakePW(self.page))
|
|
|
|
def tearDown(self):
|
|
self._tmp.cleanup()
|
|
|
|
def test_navigate_uses_proxy_and_returns_title(self):
|
|
out = self.browser.navigate("https://example.com")
|
|
self.assertEqual(out["title"], "Example Domain")
|
|
self.assertTrue(out["ok"])
|
|
# Proxy is passed explicitly to launch, not left to env.
|
|
self.assertEqual(self.browser._pw.launch_kwargs["proxy"],
|
|
{"server": "http://proxy:8888"})
|
|
self.assertIn("--no-sandbox",
|
|
self.browser._pw.launch_kwargs["args"])
|
|
|
|
def test_snapshot_truncates_with_flag(self):
|
|
self.browser.navigate("https://example.com")
|
|
out = self.browser.snapshot()
|
|
self.assertTrue(out["truncated"])
|
|
self.assertEqual(len(out["text"]), 8000)
|
|
|
|
def test_click_fill_evaluate_passthrough(self):
|
|
self.browser.navigate("https://example.com")
|
|
self.browser.click("button.go")
|
|
self.browser.fill("input.q", "hello")
|
|
out = self.browser.evaluate("1 + 1")
|
|
self.assertEqual(out["result"], "42")
|
|
self.assertIn(("click", "button.go"), self.page.calls)
|
|
self.assertIn(("fill", "input.q", "hello"), self.page.calls)
|
|
|
|
def test_screenshot_validates_basename_and_writes(self):
|
|
self.browser.navigate("https://example.com")
|
|
out = self.browser.screenshot("shot.png")
|
|
self.assertEqual(out["filename"], "shot.png")
|
|
self.assertTrue(out["bytes"] > 0)
|
|
self.assertTrue(os.path.exists(out["path"]))
|
|
with self.assertRaises(BrowserError):
|
|
self.browser.screenshot("../evil.png")
|
|
|
|
def test_close_is_idempotent(self):
|
|
self.browser.navigate("https://example.com")
|
|
self.assertTrue(self.browser.close()["closed"])
|
|
self.assertTrue(self.browser.close()["closed"])
|
|
|
|
def test_click_at_and_bounds(self):
|
|
self.browser.navigate("https://example.com")
|
|
out = self.browser.click_at(100, 200)
|
|
self.assertTrue(out["ok"])
|
|
self.assertEqual(out["x"], 100)
|
|
self.assertIn(("mouse_click", 100, 200), self.page.calls)
|
|
with self.assertRaises(BrowserError):
|
|
self.browser.click_at(2000, 10)
|
|
with self.assertRaises(BrowserError):
|
|
self.browser.click_at("left", 10)
|
|
|
|
def test_remote_input_methods(self):
|
|
self.browser.navigate("https://example.com")
|
|
out = self.browser.mouse_move(10, 20)
|
|
self.assertTrue(out["ok"])
|
|
self.assertIn(("mouse_move", 10, 20), self.page.calls)
|
|
out = self.browser.key_press("Enter")
|
|
self.assertTrue(out["ok"])
|
|
self.assertIn(("key_press", "Enter"), self.page.calls)
|
|
out = self.browser.type_text("hi")
|
|
self.assertTrue(out["ok"])
|
|
self.assertIn(("key_type", "hi"), self.page.calls)
|
|
with self.assertRaises(BrowserError):
|
|
self.browser.mouse_move(9999, 10)
|
|
|
|
def test_snapshot_refs_and_click_ref(self):
|
|
self.browser.navigate("https://example.com")
|
|
out = self.browser.snapshot_refs()
|
|
self.assertEqual(out["count"], 2)
|
|
self.assertEqual(out["elements"][0]["ref"], "e0")
|
|
self.assertEqual(out["elements"][0]["name"], "GO")
|
|
clicked = self.browser.click_ref("e0")
|
|
self.assertTrue(clicked["ok"])
|
|
self.assertEqual(clicked["ref"], "e0")
|
|
self.assertIn(("locator", "button"), self.page.calls)
|
|
with self.assertRaises(BrowserError):
|
|
self.browser.click_ref("e99")
|
|
|
|
def test_screencast_attach_and_pump(self):
|
|
from playwright_remote import SimpleQueue
|
|
sink = SimpleQueue()
|
|
self.browser.set_frame_sink(sink)
|
|
self.browser.navigate("https://example.com")
|
|
self.assertIn(("Page.startScreencast",
|
|
{"format": "jpeg", "quality": 60,
|
|
"maxWidth": 1280, "maxHeight": 800,
|
|
"everyNthFrame": 2}),
|
|
[(m, p) for m, p in self.page.cdp.sent])
|
|
handler = self.page.cdp.handlers["Page.screencastFrame"]
|
|
handler({"sessionId": "s1", "data": "FRAMEDATA"})
|
|
frame = self.browser.pump_screencast()
|
|
self.assertEqual(frame, "FRAMEDATA")
|
|
self.assertIn(("Page.screencastFrameAck", {"sessionId": "s1"}),
|
|
[(m, p) for m, p in self.page.cdp.sent])
|
|
|
|
def test_live_shot_writes_fixed_file_and_status(self):
|
|
self.browser.navigate("https://example.com")
|
|
out = self.browser.live_shot()
|
|
self.assertEqual(out["image"], "browser-live.png")
|
|
self.assertTrue(out["bytes"] > 0)
|
|
self.assertTrue(os.path.exists(out["path"]))
|
|
with open(os.path.join(self._tmp.name,
|
|
"browser-live.json")) as fh:
|
|
status = json.load(fh)
|
|
self.assertEqual(status["image"], "browser-live.png")
|
|
self.assertEqual(status["url"], "https://example.com")
|
|
|
|
|
|
class FakeMCPBrowser:
|
|
def navigate(self, url):
|
|
return {"url": url, "title": "T", "ok": True}
|
|
|
|
def click_at(self, x, y):
|
|
return {"url": "https://x/", "title": "T", "ok": True,
|
|
"x": x, "y": y}
|
|
|
|
def live_shot(self, name=None):
|
|
return {"image": name or "browser-live.png",
|
|
"path": "/w/browser-live.png", "bytes": 456,
|
|
"url": "https://x/", "title": "T", "timestamp": 1}
|
|
|
|
def mouse_move(self, x, y):
|
|
return {"ok": True, "x": x, "y": y}
|
|
|
|
def key_press(self, key):
|
|
return {"ok": True, "key": key}
|
|
|
|
def type_text(self, text):
|
|
return {"ok": True}
|
|
|
|
def snapshot_refs(self):
|
|
return {"elements": [{"ref": "e0", "tag": "button",
|
|
"role": "", "name": "GO"}], "count": 1}
|
|
|
|
def click_ref(self, ref):
|
|
return {"ok": True, "ref": ref}
|
|
|
|
def snapshot(self):
|
|
return {"url": "https://x/", "title": "T", "text": "hi",
|
|
"truncated": False}
|
|
|
|
def screenshot(self, name=None):
|
|
return {"path": "/w/shot.png", "filename": name or "shot.png",
|
|
"bytes": 123}
|
|
|
|
|
|
class MCPTest(unittest.TestCase):
|
|
def test_tools_list_includes_all_browser_tools(self):
|
|
listed = handle_request({"jsonrpc": "2.0", "id": 1,
|
|
"method": "tools/list"})
|
|
names = [t["name"] for t in listed["result"]["tools"]]
|
|
for name in ("browser_navigate", "browser_snapshot",
|
|
"browser_click", "browser_fill",
|
|
"browser_evaluate", "browser_screenshot",
|
|
"browser_close", "browser_click_at",
|
|
"browser_live_shot"):
|
|
self.assertIn(name, names)
|
|
|
|
def test_tools_call_dispatches(self):
|
|
resp = handle_request(
|
|
{"jsonrpc": "2.0", "id": 2, "method": "tools/call",
|
|
"params": {"name": "browser_navigate",
|
|
"arguments": {"url": "https://example.com"}}},
|
|
browser=FakeMCPBrowser())
|
|
text = resp["result"]["content"][0]["text"]
|
|
self.assertIn("https://example.com", text)
|
|
self.assertFalse(resp["result"]["isError"])
|
|
|
|
def test_holder_user_blocks_agent_tools(self):
|
|
import playwright_mcp
|
|
fake = FakeMCPBrowser()
|
|
remote = playwright_mcp._get_bridge(fake)
|
|
remote.set_holder("user")
|
|
try:
|
|
resp = handle_request(
|
|
{"jsonrpc": "2.0", "id": 3, "method": "tools/call",
|
|
"params": {"name": "browser_navigate",
|
|
"arguments": {"url": "https://example.com"}}},
|
|
browser=fake)
|
|
self.assertTrue(resp["result"]["isError"])
|
|
self.assertIn("holds browser control",
|
|
resp["result"]["content"][0]["text"])
|
|
finally:
|
|
remote.set_holder("agent")
|
|
|
|
def test_new_tools_listed(self):
|
|
listed = handle_request({"jsonrpc": "2.0", "id": 4,
|
|
"method": "tools/list"})
|
|
names = [t["name"] for t in listed["result"]["tools"]]
|
|
for name in ("browser_snapshot_refs", "browser_click_ref",
|
|
"browser_mouse_move", "browser_key_press",
|
|
"browser_type_text"):
|
|
self.assertIn(name, names)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|