- playwright_browser.Browser:单页 Chromium,navigate / snapshot(正文 8000 字截断标记)/ click / fill / evaluate / screenshot(工作区内)/ close;启动参数 --no-sandbox + --disable-dev-shm-usage(runsc + 1003 用户实测必需), 代理显式传给 launch(Chromium 在 Linux 下不稳定地读 env 代理),浏览器路径 PLAYWRIGHT_BROWSERS_PATH=/opt/ms-playwright。 - playwright_mcp:stdio MCP,7 个 browser_* 工具,同一进程 单 Browser 实例;测试用 fake browser / fake factory。 - worker:--playwright-mcp flag,_tenant_dirs 播种两个脚本 + [mcp_servers.playwright],_prepare_tenant 写 env(代理 + 浏览器路径,MCP 子进程不继承 Codex 进程 env,镜像 ENV 在此必须重写,缺了会报 Executable doesn't exist)。 - 技能 browser-automation:JS 页、翻页、表单、截图时用, 静态页优先 web_fetch / web_search;截图须 PIL 读回验证。 - 镜像(层构建,非本 PR):FROM latest + fonts-noto-cjk + playwright + chromium(npmmirror)+ install-deps,3.77GB; runsc + 1003 + docker-codex 条件下 JS 执行、example.com 标题、截图均已实测通过。
187 lines
5.7 KiB
Python
187 lines
5.7 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
|
|
|
|
|
|
class _FakePage:
|
|
def __init__(self):
|
|
self.url = "https://example.com/"
|
|
self.calls = []
|
|
self.shot_path = None
|
|
|
|
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))
|
|
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"])
|
|
|
|
|
|
class FakeMCPBrowser:
|
|
def navigate(self, url):
|
|
return {"url": url, "title": "T", "ok": True}
|
|
|
|
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"):
|
|
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"])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|