core-python/tests/test_browser.py

"""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)