422 lines
14 KiB
Python
422 lines
14 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)
|
|
|
|
|
|
class LineReaderTest(unittest.TestCase):
|
|
"""Pipelined stdin lines must all arrive (no select starvation).
|
|
|
|
Regression test for the live failure where the MCP answered
|
|
initialize and then hung forever: the client wrote
|
|
notifications/initialized + tools/call back-to-back, one
|
|
readline() buffered both, and select() never saw the second
|
|
line again because the fd was drained.
|
|
"""
|
|
|
|
def test_pipelined_lines_all_delivered(self):
|
|
import time
|
|
from playwright_mcp import _LineReader
|
|
rfd, wfd = os.pipe()
|
|
try:
|
|
reader = _LineReader(rfd)
|
|
os.write(wfd, b"init\n")
|
|
lines, drained = reader.poll()
|
|
self.assertEqual(lines, ["init"])
|
|
self.assertFalse(drained)
|
|
# Second batch arrives as one write, like a real client
|
|
# sending notification + next request back-to-back.
|
|
os.write(wfd, b"notif\nnavigate\n")
|
|
# Simulate the drain-loop delay: both lines sit in the
|
|
# pipe before the next poll.
|
|
time.sleep(0.2)
|
|
got = []
|
|
deadline = time.time() + 2
|
|
while len(got) < 2 and time.time() < deadline:
|
|
lines, drained = reader.poll()
|
|
got.extend(lines)
|
|
if not lines:
|
|
time.sleep(0.02)
|
|
self.assertEqual(got, ["notif", "navigate"])
|
|
finally:
|
|
os.close(wfd)
|
|
os.close(rfd)
|
|
|
|
def test_partial_line_flushed_at_eof(self):
|
|
from playwright_mcp import _LineReader
|
|
rfd, wfd = os.pipe()
|
|
try:
|
|
reader = _LineReader(rfd)
|
|
os.write(wfd, b"partial-no-newline")
|
|
os.close(wfd)
|
|
lines, drained = reader.poll()
|
|
self.assertEqual(lines, [])
|
|
self.assertFalse(drained)
|
|
lines, drained = reader.poll()
|
|
self.assertEqual(lines, ["partial-no-newline"])
|
|
self.assertTrue(drained)
|
|
finally:
|
|
try:
|
|
os.close(rfd)
|
|
except OSError:
|
|
pass
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|