browser-cli/tests/test_mcp.py

"""Tests for the optional stateless MCP adapter."""
from __future__ import annotations

import base64
from types import SimpleNamespace

import pytest

pytest.importorskip("mcp")

from mcp import Client
from mcp.types import ImageContent

from browser_cli.mcp.serialization import structured
from browser_cli.mcp.naming import resolve_tool_prefix
from browser_cli.mcp.server import _screenshot_bytes, create_server, main
from browser_cli.models import Tab

class FakeClient:
  instances: list["FakeClient"] = []

  def __init__(self, browser=None, remote=None, key=None):
    self.target = (browser, remote, key)
    self.calls: list[tuple] = []
    self.tabs = SimpleNamespace(
      list=self.tabs_list,
      open=self.tabs_open,
      close=self.tabs_close,
      active=self.tabs_active,
      status=self.tabs_status,
      screenshot=self.tabs_screenshot,
    )
    self.nav = SimpleNamespace(to=self.navigate_to)
    self.page = SimpleNamespace(info=self.page_info)
    self.extract = SimpleNamespace(text=self.extract_text, markdown=self.extract_markdown)
    self.dom = SimpleNamespace(query=self.dom_query, click=self.dom_click, type=self.dom_type)
    self.instances.append(self)

  def tabs_list(self):
    return [{"id": 7, "title": "Example", "url": "https://example.com"}]

  def tabs_open(self, url, **kwargs):
    self.calls.append(("open", url, kwargs))
    return {"id": 8, "title": "Opened", "url": url}

  def tabs_close(self, tab_id):
    self.calls.append(("close", tab_id))
    return 1

  def tabs_active(self):
    self.calls.append(("active",))
    return SimpleNamespace(id=7)

  def tabs_status(self, tab_id):
    self.calls.append(("status", tab_id))
    return {"id": tab_id, "title": "Navigated", "url": "https://example.com/next"}

  def tabs_screenshot(self, tab_id, **kwargs):
    self.calls.append(("screenshot", tab_id, kwargs))
    return "data:image/png;base64," + base64.b64encode(b"png-data").decode()

  def navigate_to(self, tab_id, url):
    self.calls.append(("navigate", tab_id, url))

  def page_info(self):
    return {"title": "Example", "url": "https://example.com"}

  def extract_text(self):
    return "Page text"

  def extract_markdown(self, selector=None):
    return f"# Page {selector or ''}".rstrip()

  def dom_query(self, selector):
    return [{"tag": "button", "selector": selector}]

  def dom_click(self, selector):
    self.calls.append(("click", selector))

  def dom_type(self, selector, text):
    self.calls.append(("type", selector, text))

@pytest.fixture(autouse=True)
def clear_instances():
  FakeClient.instances.clear()

@pytest.fixture
def anyio_backend():
  return "asyncio"

@pytest.fixture
async def client():
  server = create_server(client_factory=FakeClient)
  async with Client(server, raise_exceptions=True) as connected:
    yield connected

@pytest.mark.anyio
async def test_each_tool_call_constructs_a_fresh_targeted_client(client, monkeypatch):
  monkeypatch.delenv("BROWSER_CLI_PROFILE", raising=False)
  monkeypatch.delenv("BROWSER_CLI_REMOTE", raising=False)
  monkeypatch.delenv("BROWSER_CLI_KEY", raising=False)
  first = await client.call_tool("browser_tabs_list", {
    "browser": "work",
    "remote": "browser-host.example:443",
    "key": "agent",
  })
  second = await client.call_tool("browser_page_info", {})

  assert first.structured_content == {
    "result": [{"id": 7, "title": "Example", "url": "https://example.com"}]
  }
  assert second.structured_content == {
    "title": "Example",
    "url": "https://example.com",
  }
  assert len(FakeClient.instances) == 2
  assert FakeClient.instances[0].target == ("work", "browser-host.example:443", "agent")
  assert FakeClient.instances[1].target == (None, None, None)

@pytest.mark.anyio
async def test_environment_pins_all_calls_to_one_browser(client, monkeypatch):
  monkeypatch.setenv("BROWSER_CLI_PROFILE", "testing")
  monkeypatch.delenv("BROWSER_CLI_REMOTE", raising=False)
  monkeypatch.delenv("BROWSER_CLI_KEY", raising=False)

  await client.call_tool("browser_tabs_list", {})

  assert FakeClient.instances[-1].target == ("testing", None, None)

def test_tool_prefix_defaults_to_browser_and_is_configurable():
  assert resolve_tool_prefix({}) == "browser_"
  assert resolve_tool_prefix({"BROWSER_CLI_MCP_TOOL_PREFIX": ""}) == ""
  assert resolve_tool_prefix({"BROWSER_CLI_MCP_TOOL_PREFIX": "web"}) == "web_"
  assert resolve_tool_prefix({"BROWSER_CLI_MCP_TOOL_PREFIX": "web_"}) == "web_"
  with pytest.raises(ValueError):
    resolve_tool_prefix({"BROWSER_CLI_MCP_TOOL_PREFIX": "9-bad prefix"})

@pytest.mark.anyio
async def test_hosts_that_namespace_tools_can_drop_the_builtin_prefix():
  server = create_server(client_factory=FakeClient, tool_prefix="")
  async with Client(server, raise_exceptions=True) as connected:
    names = {tool.name for tool in (await connected.list_tools()).tools}

  assert "tabs_list" in names
  assert not any(name.startswith("browser_") for name in names)

@pytest.mark.anyio
async def test_explicit_target_overrides_environment_pin(client, monkeypatch):
  monkeypatch.setenv("BROWSER_CLI_PROFILE", "testing")
  monkeypatch.delenv("BROWSER_CLI_REMOTE", raising=False)
  monkeypatch.delenv("BROWSER_CLI_KEY", raising=False)

  await client.call_tool("browser_tabs_list", {"browser": "main"})

  assert FakeClient.instances[-1].target == ("main", None, None)

@pytest.mark.anyio
async def test_mutating_tools_use_sdk_and_return_fresh_state(client):
  opened = await client.call_tool("browser_tabs_open", {
    "url": "https://example.com",
    "wait": True,
    "focus": True,
  })
  navigated = await client.call_tool("browser_navigate", {
    "tab_id": 8,
    "url": "https://example.com/next",
  })
  clicked = await client.call_tool("browser_dom_click", {"selector": "#submit"})
  typed = await client.call_tool("browser_dom_type", {"selector": "#name", "text": "Daniel"})

  assert opened.structured_content == {
    "id": 8, "title": "Opened", "url": "https://example.com"
  }
  assert navigated.structured_content["id"] == 8
  assert clicked.structured_content["url"] == "https://example.com"
  assert typed.structured_content == {"typed": True}
  assert FakeClient.instances[0].calls == [
    ("open", "https://example.com", {
      "wait": True, "timeout": 30.0, "background": False, "focus": True
    })
  ]
  assert FakeClient.instances[1].calls == [
    ("navigate", 8, "https://example.com/next"), ("status", 8)
  ]

@pytest.mark.anyio
async def test_tab_tools_default_to_the_active_tab(client):
  navigated = await client.call_tool("browser_navigate", {"url": "https://example.com/next"})
  closed = await client.call_tool("browser_tabs_close", {})

  assert navigated.structured_content["id"] == 7
  assert closed.structured_content == {"closed": 1, "tab_id": 7}
  assert FakeClient.instances[0].calls == [
    ("active",), ("navigate", 7, "https://example.com/next"), ("status", 7)
  ]
  assert FakeClient.instances[1].calls == [("active",), ("close", 7)]

@pytest.mark.anyio
async def test_screenshot_returns_image_content(client):
  result = await client.call_tool("browser_screenshot", {"tab_id": 7, "format": "png"})

  assert result.structured_content is None
  assert len(result.content) == 1
  assert isinstance(result.content[0], ImageContent)
  assert result.content[0].data == base64.b64encode(b"png-data").decode()
  assert result.content[0].mime_type == "image/png"

def test_screenshot_decoder_rejects_non_data_url():
  with pytest.raises(ValueError, match="invalid screenshot"):
    _screenshot_bytes("not-an-image")

def test_sdk_dataclass_serialization_does_not_traverse_bound_client():
  tab = Tab(id=7, window_id=1, active=True, title="Example")
  tab._browser = SimpleNamespace(secret="must not be serialized")

  result = structured(tab)

  assert result["id"] == 7
  assert result["window_id"] == 1
  assert "_browser" not in result

def test_http_server_refuses_non_local_bind(monkeypatch):
  monkeypatch.setattr("browser_cli.mcp.server.create_server", lambda: SimpleNamespace(run=lambda **kwargs: None))

  with pytest.raises(SystemExit, match="Refusing to expose"):
    main(["--transport", "streamable-http", "--host", "0.0.0.0"])

def test_http_server_enables_stateless_json_transport(monkeypatch):
  calls = []
  monkeypatch.setattr("browser_cli.mcp.server.create_server", lambda: SimpleNamespace(run=lambda **kwargs: calls.append(kwargs)))

  main(["--transport", "streamable-http", "--port", "9000", "--path", "/browser"])

  assert calls == [{
    "transport": "streamable-http",
    "host": "127.0.0.1",
    "port": 9000,
    "streamable_http_path": "/browser",
    "stateless_http": True,
    "json_response": True,
  }]