File size: 4,938 Bytes
f440f03 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 | """Tests for Maris browser automation endpoints."""
from __future__ import annotations
import base64
import pytest
from fastapi import HTTPException
from maris_core.browser.automation import (
BrowserAutomationUnavailableError,
BrowserExtractRequest,
BrowserNavigateRequest,
BrowserScreenshotRequest,
BrowserSessionRequest,
BrowserSessionStartRequest,
_browser_sessions,
browser_capabilities,
close_browser_session,
extract_browser_text,
navigate_browser,
screenshot_browser,
start_browser_session,
)
class _FakeBrowserSession:
def __init__(self) -> None:
self.headless = True
self.viewport = {"width": 1280, "height": 720}
self.url = "about:blank"
self.title = "Blank"
self.closed = False
async def snapshot(self) -> dict[str, str]:
return {"url": self.url, "title": self.title}
async def navigate(self, url: str, *, wait_until: str, timeout_ms: int) -> dict[str, str]:
assert wait_until == "load"
assert timeout_ms == 3210
self.url = url
self.title = "Target page"
return await self.snapshot()
async def click(
self, selector: str, *, timeout_ms: int, wait_until_after: str | None = None
) -> dict[str, str]:
raise AssertionError(f"Unexpected click on {selector} with {timeout_ms} {wait_until_after}")
async def fill(
self, selector: str, value: str, *, timeout_ms: int, submit: bool
) -> dict[str, str]:
raise AssertionError(f"Unexpected fill on {selector}={value} with {timeout_ms} {submit}")
async def extract_text(self, selector: str | None, *, timeout_ms: int, max_length: int) -> str:
assert selector == "#main"
assert timeout_ms == 4321
return "Sveiks no browser workflow"[:max_length]
async def screenshot_png(self, *, full_page: bool) -> bytes:
assert full_page is True
return b"png-bytes"
async def close(self) -> None:
self.closed = True
@pytest.fixture(autouse=True)
def clear_browser_sessions() -> None:
_browser_sessions.clear()
@pytest.mark.asyncio
async def test_browser_capabilities_are_exposed() -> None:
response = await browser_capabilities()
assert response.provider == "playwright"
assert "navigate" in response.supported_actions
assert response.max_sessions >= 1
@pytest.mark.asyncio
async def test_browser_session_lifecycle_supports_navigation_extract_and_screenshot(
monkeypatch,
) -> None:
fake_session = _FakeBrowserSession()
async def _fake_create_browser_session(
*, headless: bool, viewport: dict[str, int]
) -> _FakeBrowserSession:
assert headless is True
assert viewport == {"width": 1280, "height": 720}
return fake_session
monkeypatch.setattr(
"maris_core.browser.automation._create_browser_session",
_fake_create_browser_session,
)
started = await start_browser_session(
BrowserSessionStartRequest(headless=True, viewport_width=1280, viewport_height=720)
)
session_id = started.session_id
navigated = await navigate_browser(
BrowserNavigateRequest(
session_id=session_id,
url="https://example.com",
wait_until="load",
timeout_ms=3210,
)
)
extracted = await extract_browser_text(
BrowserExtractRequest(
session_id=session_id,
selector="#main",
timeout_ms=4321,
max_length=128,
)
)
screenshot = await screenshot_browser(
BrowserScreenshotRequest(session_id=session_id, full_page=True)
)
closed = await close_browser_session(BrowserSessionRequest(session_id=session_id))
assert started.active is True
assert navigated.url == "https://example.com"
assert extracted.text == "Sveiks no browser workflow"
assert base64.b64decode(screenshot.image_base64) == b"png-bytes"
assert closed.active is False
assert fake_session.closed is True
@pytest.mark.asyncio
async def test_browser_start_returns_503_when_playwright_is_unavailable(monkeypatch) -> None:
async def _unavailable(**_: object) -> None:
raise BrowserAutomationUnavailableError("Playwright nav pieejams")
monkeypatch.setattr("maris_core.browser.automation._create_browser_session", _unavailable)
with pytest.raises(HTTPException) as exc_info:
await start_browser_session(BrowserSessionStartRequest())
assert exc_info.value.status_code == 503
assert "Playwright" in exc_info.value.detail
@pytest.mark.parametrize(
"url",
[
"file:///etc/passwd",
"javascript:alert(1)",
"ftp://example.com",
"chrome://settings",
],
)
def test_browser_request_rejects_unsafe_schemes(url: str) -> None:
with pytest.raises(ValueError):
BrowserNavigateRequest(session_id="session", url=url)
|