codexpi/tests/test_browser_service.py

from __future__ import annotations

import base64
from pathlib import Path

import pytest

from codexpi.browser_service import BrowserError, BrowserService, screenshot_attachments
from codexpi.database import Database, utcnow


class FakeBrowser(BrowserService):
    def __init__(self, db: Database, root: Path):
        super().__init__(db, "http://browser.invalid", root)
        self.url = "about:blank"
        self.session_payload = None
        self.interactive_stopped = False

    def _ensure_interactive_chromium(self):
        return None

    def _stop_interactive_chromium(self):
        self.interactive_stopped = True

    def _request(self, method, path, payload=None, timeout=60):
        if path == "/status":
            return {"ready": True}
        if method == "POST" and path == "/session":
            self.session_payload = payload
            return {"sessionId": "session-1"}
        if path.endswith("/url") and method == "POST":
            self.url = payload["url"]
            return None
        if path.endswith("/url"):
            return self.url
        if path.endswith("/screenshot"):
            return base64.b64encode(b"\x89PNG\r\n\x1a\nimage").decode()
        return None


class RecoveringBrowser(FakeBrowser):
    def __init__(self, db: Database, root: Path):
        super().__init__(db, root)
        self.first_session = True
        self.restarted = False

    def _request(self, method, path, payload=None, timeout=60):
        if method == "POST" and path == "/session" and self.first_session:
            self.first_session = False
            raise BrowserError("session not created: DevToolsActivePort file doesn't exist")
        return super()._request(method, path, payload, timeout)

    def _restart_browser_container(self):
        self.restarted = True


class UnavailableBrowser(RecoveringBrowser):
    def _request(self, method, path, payload=None, timeout=60):
        if method == "POST" and path == "/session" and self.first_session:
            self.first_session = False
            raise BrowserError("Le service du navigateur intégré est indisponible")
        return FakeBrowser._request(self, method, path, payload, timeout)


def browser(tmp_path: Path) -> tuple[FakeBrowser, Database, str]:
    db = Database(tmp_path / "db.sqlite")
    db.initialize()
    project_id = "4c4c95da-7404-41db-aa9f-31d6bf13c7fc"
    db.execute("INSERT INTO projects(id,name,path,created_at) VALUES(?,?,?,?)", (project_id, "Demo", str(tmp_path), utcnow()))
    return FakeBrowser(db, tmp_path / "screenshots"), db, project_id


def add_project(db: Database, root: Path, name: str = "Second") -> str:
    project_id = "d57866c2-b8e4-4fc3-a897-d67d61a8ba33"
    db.execute("INSERT INTO projects(id,name,path,created_at) VALUES(?,?,?,?)", (project_id, name, str(root / name), utcnow()))
    return project_id


def test_url_validation_rejects_privileged_schemes():
    assert BrowserService.validate_url("example.com") == "https://example.com"
    with pytest.raises(BrowserError):
        BrowserService.validate_url("file:///etc/passwd")


def test_session_navigation_and_screenshot_are_persisted(tmp_path):
    service, db, project_id = browser(tmp_path)
    assert service.navigate(project_id, "example.com")["url"] == "https://example.com"
    shot = service.screenshot(project_id, "Résultat", "queue-1")
    assert Path(shot["path"]).is_file()
    assert screenshot_attachments(str(db.path), "queue-1")[0]["url"].endswith(shot["id"])
    state = service.status(project_id)
    assert state["running"] is True
    assert state["current_url"] == "https://example.com"


def test_webdriver_attaches_to_interactive_chromium(tmp_path):
    service, _, project_id = browser(tmp_path)

    service.ensure(project_id)

    options = service.session_payload["capabilities"]["alwaysMatch"]["goog:chromeOptions"]
    assert options == {"debuggerAddress": "127.0.0.1:9222"}


def test_handoff_blocks_codex_navigation(tmp_path):
    service, _, project_id = browser(tmp_path)
    service.ensure(project_id)
    service.set_control(project_id, "user", "Connectez-vous")
    with pytest.raises(BrowserError):
        service.navigate(project_id, "https://example.com")
    assert service.navigate(project_id, "https://example.com", require_codex=False)["url"] == "https://example.com"


def test_browser_session_and_handoff_are_shared_between_projects(tmp_path):
    service, db, first_project = browser(tmp_path)
    second_project = add_project(db, tmp_path)

    assert service.navigate(first_project, "https://example.com")["url"] == "https://example.com"
    assert service.status(second_project)["running"] is True
    assert service.status(second_project)["current_url"] == "https://example.com"

    service.set_control(first_project, "user", "Action sensible")

    assert service.status(second_project)["control_mode"] == "user"
    assert service.status(second_project)["handoff_message"] == "Action sensible"
    with pytest.raises(BrowserError):
        service.navigate(second_project, "https://openai.com")


def test_reset_closes_the_shared_interactive_browser(tmp_path):
    service, _, project_id = browser(tmp_path)
    service.ensure(project_id)

    service.reset(project_id)

    assert service.interactive_stopped is True


def test_stale_chromium_profile_is_recovered_once(tmp_path):
    _, db, project_id = browser(tmp_path)
    service = RecoveringBrowser(db, tmp_path / "recovery-screenshots")

    assert service.ensure(project_id) == "session-1"
    assert service.restarted is True


def test_unavailable_selenium_is_recovered_once(tmp_path):
    _, db, project_id = browser(tmp_path)
    service = UnavailableBrowser(db, tmp_path / "unavailable-screenshots")

    assert service.ensure(project_id) == "session-1"
    assert service.restarted is True