admin/codex-app-server
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
be41ad2c0a145fa2ef9557db63a6c6afd8d22f4c
codex-app-server/cloud/tests/test_playwright.py
T
host da7a9950fb feat: platform browser multi-tab, navigation allowlist, close cleanup, egress filter, allowlist worker wiring 
- Multi-tab: browser_new_page / browser_list_pages /
  browser_switch_page / browser_close_page in playwright_mcp.py;
  each new page gets a fresh Playwright context, so cookies and
  storage are isolated between pages. Single-page tools act on
  the current page; switching to a missing index is an error,
  not a crash; closing the last page is an error (use
  browser_close).
- Navigation allowlist: BROWSER_ALLOWED_HOSTS (comma list,
  leading-dot suffix match, case/port-insensitive) checked in
  _check_allowed_url before launching Chromium; unset (or empty)
  means no check, the old behaviour. file:// always allowed;
  localhost/127.0.0.1 have no special treatment.
- Close cleanup: browser_close closes all pages and best-effort
  removes cwd browser-live.png / browser-live.json; custom
  screenshot files are untouched. list_pages needs no launch.
- Egress filter: 10 programming-source Tinyproxy BASIC rules
  (python.org, nodejs.org, go.dev, pkg.go.dev, rust-lang.org,
  developer.mozilla.org, crates.io, stackoverflow.com,
  cve.mitre.org, nvd.nist.gov); already-present github.com /
  pypi.org / npmjs.org not duplicated; searx.diante.me stays
  exact-only, no diante.me wildcard.
- Allowlist worker wiring: _docker_task_context now passes a
  set, non-empty BROWSER_ALLOWED_HOSTS from the worker env into
  [mcp_servers.playwright.env] in the tenant config, so the MCP
  process actually sees it (container -e alone does not reach
  MCP subprocesses, 2026-10-30). Unset/empty writes nothing and
  never overwrites; the value never goes into logs; it is not
  added to the websearch env block or the container env.
  ensure_playwright_mcp_env now returns False without rewriting
  when the result would be identical (canonical join: one blank
  line before the block), so a second prep leaves an unchanged
  tenant config alone; other sections untouched.
- Skill + docs: browser-automation SKILL.md multi-tab/allowlist
  section; docs/platform-browser.md comparison doc (new file).
- Tests (mock/fake only, no network, no real Chromium, no real
  credentials): cloud.tests.test_playwright +
  cloud.tests.test_websearch_worker, 48 tests OK, including 6
  new wiring tests (switch off passes no variable, switch on
  reaches MCP env verbatim, empty never written, old-tenant
  merge idempotent + no-rewrite, ensure no-op when identical,
  allowlist enforces only after wiring).
- NOT done (see PR body): no persistent profile, no credential
  vault, no approval card, no CAPTCHA bypass, in-page link
  navigation does not pass the allowlist, no real-Chromium live
  verification, host tinyproxy filter not reloaded.
2026-11-01 08:34:13 +00:00

748 lines
27 KiB
Python
Raw Blame History

"""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 = {}
self.fail_capture = False
def on(self, event, handler):
self.handlers[event] = handler
def send(self, method, params=None):
self.sent.append((method, params or {}))
if method == "Page.captureScreenshot":
if self.fail_capture:
raise RuntimeError("capture boom")
return {"data": "EAGERDATA"}
return None
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"])
for flag in ("--no-first-run", "--no-default-browser-check",
"--disable-default-apps", "--disable-extensions",
"--disable-background-networking", "--disable-sync",
"--disable-translate"):
self.assertIn(flag, 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": 50,
"maxWidth": 1280, "maxHeight": 800,
"everyNthFrame": 1}),
[(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_navigate_pushes_eager_frame_without_ack(self):
from playwright_remote import SimpleQueue
sink = SimpleQueue()
self.browser.set_frame_sink(sink)
result = self.browser.navigate("https://example.com")
self.assertTrue(result["ok"])
self.assertIn(("Page.captureScreenshot",
{"format": "jpeg", "quality": 50}),
[(m, p) for m, p in self.page.cdp.sent])
frame = self.browser.pump_screencast()
self.assertEqual(frame, "EAGERDATA")
acks = [(m, p) for m, p in self.page.cdp.sent
if m == "Page.screencastFrameAck"]
self.assertEqual(acks, [])
def test_navigate_survives_failed_eager_capture(self):
from playwright_remote import SimpleQueue
sink = SimpleQueue()
self.browser.set_frame_sink(sink)
self.page.cdp.fail_capture = True
try:
result = self.browser.navigate("https://example.com")
finally:
self.page.cdp.fail_capture = False
self.assertTrue(result["ok"])
self.assertEqual(self.browser.pump_screencast(), "")
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
class _FakeBrowserHandleMulti:
"""Distinct fake page per new_page (the old single-page fake
returns the same page every time, which cannot tell pages
apart)."""
def __init__(self):
self.pages = []
def new_page(self, viewport=None):
page = _FakePage()
page.url = "https://page%d.example.com/" % len(self.pages)
self.pages.append(page)
return page
def close(self):
pass
class _FakePWMulti:
def __init__(self):
self.handle = _FakeBrowserHandleMulti()
self.launch_kwargs = None
self.launch_count = 0
def start(self):
return self
@property
def chromium(self):
outer = self
class _Chromium:
def launch(self, **kwargs):
outer.launch_kwargs = kwargs
outer.launch_count += 1
return outer.handle
return _Chromium()
def stop(self):
pass
class _EnvGuard:
"""Save/restore BROWSER_ALLOWED_HOSTS around a test case."""
def setUp(self):
self._old_allowlist = os.environ.get("BROWSER_ALLOWED_HOSTS")
def tearDown(self):
if self._old_allowlist is None:
os.environ.pop("BROWSER_ALLOWED_HOSTS", None)
else:
os.environ["BROWSER_ALLOWED_HOSTS"] = self._old_allowlist
class MultiPageTest(_EnvGuard, unittest.TestCase):
def setUp(self):
_EnvGuard.setUp(self)
self._tmp = tempfile.TemporaryDirectory()
self.factory = _FakePWMulti()
self.browser = Browser(cwd=self._tmp.name,
proxy="http://proxy:8888",
browser_factory=lambda: self.factory)
def tearDown(self):
self._tmp.cleanup()
_EnvGuard.tearDown(self)
def test_new_list_switch_current_page_semantics(self):
self.browser.navigate("https://a.example.com")
self.assertEqual(self.browser.list_pages()["count"], 1)
out = self.browser.new_page()
self.assertEqual(out["index"], 1)
self.assertEqual(out["count"], 2)
self.browser.navigate("https://b.example.com")
self.browser.click("button.x")
page0, page1 = self.factory.handle.pages
self.assertIn(("goto", "https://b.example.com"), page1.calls)
self.assertNotIn(("goto", "https://b.example.com"), page0.calls)
self.assertIn(("click", "button.x"), page1.calls)
self.assertNotIn(("click", "button.x"), page0.calls)
switched = self.browser.switch_page(0)
self.assertEqual(switched["index"], 0)
self.browser.click("a.y")
self.assertIn(("click", "a.y"), page0.calls)
self.assertNotIn(("click", "a.y"), page1.calls)
listed = self.browser.list_pages()
self.assertEqual(listed["count"], 2)
self.assertEqual(listed["current"], 0)
self.assertEqual([p["url"] for p in listed["pages"]],
["https://a.example.com",
"https://b.example.com"])
self.assertEqual([p["current"] for p in listed["pages"]],
[True, False])
def test_switch_missing_page_errors_and_keeps_current(self):
self.browser.navigate("https://a.example.com")
self.browser.new_page()
for bad in (99, -1, "x"):
with self.assertRaises(BrowserError) as ctx:
self.browser.switch_page(bad)
if bad != "x":
self.assertIn("no such page", str(ctx.exception))
self.assertEqual(self.browser.list_pages()["current"], 1)
def test_close_current_switches_to_previous(self):
self.browser.navigate("https://a.example.com")
self.browser.new_page()
self.browser.new_page()
out = self.browser.close_page()
self.assertEqual(out["closed_index"], 2)
self.assertEqual(out["index"], 1)
self.assertEqual(out["count"], 2)
self.assertIs(self.browser._page,
self.factory.handle.pages[1])
self.assertEqual(self.browser.list_pages()["current"], 1)
def test_close_noncurrent_keeps_same_page(self):
self.browser.navigate("https://a.example.com")
self.browser.new_page()
self.browser.new_page()
self.browser.switch_page(2)
page2 = self.factory.handle.pages[2]
out = self.browser.close_page(0)
self.assertEqual(out["count"], 2)
self.assertEqual(out["index"], 1)
self.assertIs(self.browser._page, page2)
def test_close_last_page_errors_and_keeps_page(self):
self.browser.navigate("https://a.example.com")
with self.assertRaises(BrowserError) as ctx:
self.browser.close_page()
self.assertIn("last page", str(ctx.exception))
with self.assertRaises(BrowserError):
self.browser.close_page(0)
self.assertEqual(self.browser.list_pages()["count"], 1)
# Still usable after the refused close.
self.browser.navigate("https://b.example.com")
self.assertEqual(self.browser._current_url,
"https://b.example.com")
def test_close_removes_live_files_and_list_needs_no_launch(self):
self.browser.navigate("https://a.example.com")
self.browser.live_shot()
self.assertTrue(os.path.exists(os.path.join(
self._tmp.name, "browser-live.png")))
self.assertTrue(os.path.exists(os.path.join(
self._tmp.name, "browser-live.json")))
launches = self.factory.launch_count
self.assertTrue(self.browser.close()["closed"])
self.assertFalse(os.path.exists(os.path.join(
self._tmp.name, "browser-live.png")))
self.assertFalse(os.path.exists(os.path.join(
self._tmp.name, "browser-live.json")))
self.assertTrue(self.browser.close()["closed"])
self.assertEqual(self.browser.list_pages(),
{"pages": [], "current": None, "count": 0})
self.assertEqual(self.factory.launch_count, launches)
self.assertIsNone(self.browser._pw)
class AllowlistTest(_EnvGuard, unittest.TestCase):
def setUp(self):
_EnvGuard.setUp(self)
self._tmp = tempfile.TemporaryDirectory()
self.factory = _FakePWMulti()
self.browser = Browser(cwd=self._tmp.name,
proxy="http://proxy:8888",
browser_factory=lambda: self.factory)
def tearDown(self):
self._tmp.cleanup()
_EnvGuard.tearDown(self)
def test_unset_allowlist_behaves_as_before(self):
os.environ.pop("BROWSER_ALLOWED_HOSTS", None)
out = self.browser.navigate("https://anything.example")
self.assertTrue(out["ok"])
self.assertEqual(self.factory.launch_count, 1)
def test_exact_and_suffix_match(self):
os.environ["BROWSER_ALLOWED_HOSTS"] = "example.com, .python.org"
self.assertTrue(self.browser.navigate(
"https://example.com")["ok"])
self.assertTrue(self.browser.navigate(
"https://docs.python.org")["ok"])
self.assertTrue(self.browser.navigate(
"https://python.org")["ok"])
def test_port_and_case_insensitive(self):
os.environ["BROWSER_ALLOWED_HOSTS"] = "Example.COM"
out = self.browser.navigate("https://EXAMPLE.com:8443/path")
self.assertTrue(out["ok"])
def test_miss_blocked_before_launch(self):
os.environ["BROWSER_ALLOWED_HOSTS"] = "example.com"
with self.assertRaises(BrowserError) as ctx:
self.browser.navigate("https://evil.com")
self.assertIn("evil.com", str(ctx.exception))
self.assertEqual(self.factory.launch_count, 0)
self.assertIsNone(self.browser._pw)
self.assertIsNone(self.browser._page)
def test_localhost_blocked_unless_listed(self):
os.environ["BROWSER_ALLOWED_HOSTS"] = "example.com"
with self.assertRaises(BrowserError):
self.browser.navigate("http://localhost:8000/")
self.assertEqual(self.factory.launch_count, 0)
os.environ["BROWSER_ALLOWED_HOSTS"] = "localhost"
out = self.browser.navigate("http://localhost:8000/")
self.assertTrue(out["ok"])
self.assertEqual(self.factory.launch_count, 1)
def test_file_allowed_when_allowlist_set(self):
os.environ["BROWSER_ALLOWED_HOSTS"] = "example.com"
out = self.browser.navigate("file:///tmp/page.html")
self.assertTrue(out["ok"])
def test_non_http_blocked_when_allowlist_set(self):
os.environ["BROWSER_ALLOWED_HOSTS"] = "example.com"
with self.assertRaises(BrowserError):
self.browser.navigate("about:blank")
self.assertEqual(self.factory.launch_count, 0)
class MCPMultiPageTest(_EnvGuard, unittest.TestCase):
def setUp(self):
_EnvGuard.setUp(self)
os.environ.pop("BROWSER_ALLOWED_HOSTS", None)
self._tmp = tempfile.TemporaryDirectory()
self.factory = _FakePWMulti()
self.browser = Browser(cwd=self._tmp.name,
proxy="http://proxy:8888",
browser_factory=lambda: self.factory)
def tearDown(self):
self._tmp.cleanup()
_EnvGuard.tearDown(self)
def _call(self, name, args=None):
resp = handle_request(
{"jsonrpc": "2.0", "id": 9, "method": "tools/call",
"params": {"name": name, "arguments": args or {}}},
browser=self.browser)
return resp
def _call_result(self, name, args=None):
resp = self._call(name, args)
self.assertFalse(resp["result"]["isError"],
resp["result"]["content"][0]["text"])
return json.loads(resp["result"]["content"][0]["text"])
def test_tools_list_includes_new_page_tools(self):
listed = handle_request({"jsonrpc": "2.0", "id": 9,
"method": "tools/list"})
names = [t["name"] for t in listed["result"]["tools"]]
for name in ("browser_new_page", "browser_list_pages",
"browser_switch_page", "browser_close_page"):
self.assertIn(name, names)
def test_mcp_dispatch_new_list_switch_close(self):
self._call_result("browser_navigate",
{"url": "https://a.example.com"})
self.assertEqual(self._call_result("browser_new_page")["count"],
2)
listed = self._call_result("browser_list_pages")
self.assertEqual(listed["count"], 2)
self.assertEqual(listed["current"], 1)
self.assertEqual(self._call_result("browser_switch_page",
{"index": 0})["index"], 0)
closed = self._call_result("browser_close_page")
self.assertEqual(closed["count"], 1)
self.assertEqual(closed["index"], 0)
def test_mcp_switch_missing_and_close_last_are_errors_not_crash(self):
self._call_result("browser_navigate",
{"url": "https://a.example.com"})
resp = self._call("browser_switch_page", {"index": 99})
self.assertTrue(resp["result"]["isError"])
self.assertIn("no such page",
resp["result"]["content"][0]["text"])
resp = self._call("browser_close_page")
self.assertTrue(resp["result"]["isError"])
self.assertIn("last page",
resp["result"]["content"][0]["text"])
if __name__ == "__main__":
unittest.main()
Reference in new issue View Git Blame Copy Permalink