gyxx-flow/tests/test_scrapling_adapter.py

from __future__ import annotations

import json
from pathlib import Path
from typing import Any

import pytest

import gyxx_flow.adapters.scrapling as scrapling_adapter
from gyxx_flow.adapters.scrapling import (
    BrowserTimeoutError,
    ScraplingBrowser,
    is_browser_timeout_error,
)


class _FakeContext:
    def __init__(self, *, browser: _FakeBrowser | None = None) -> None:
        self.browser = browser
        self.close_count = 0
        self.added_cookies: list[list[dict[str, Any]]] = []
        self.restored_states: list[dict[str, Any]] = []

    def close(self) -> None:
        self.close_count += 1

    def add_cookies(self, cookies: list[dict[str, Any]]) -> None:
        self.added_cookies.append(cookies)

    def set_storage_state(self, state: dict[str, Any]) -> None:
        self.restored_states.append(state)

    def cookies(self) -> list[dict[str, Any]]:
        return [{"name": "saved", "value": "cookie"}]

    def storage_state(self) -> dict[str, Any]:
        return {"cookies": self.cookies(), "origins": []}


class _ClosedOnExportContext(_FakeContext):
    def cookies(self) -> list[dict[str, Any]]:
        raise RuntimeError(
            "BrowserContext.cookies: Target page, context or browser has been closed"
        )


class _FakeBrowser:
    def __init__(self, contexts: list[_FakeContext] | None = None) -> None:
        self.contexts = contexts or []
        self.close_count = 0

    def close(self) -> None:
        self.close_count += 1


class _FakeSession:
    def __init__(
        self,
        *,
        context: _FakeContext,
        browser: _FakeBrowser | None = None,
        page: object | None = None,
        response: object | None = None,
    ) -> None:
        self.context = context
        self.browser = browser
        self.page = page or object()
        self.response = response or object()
        self.start_count = 0
        self.close_count = 0
        self.fetch_calls: list[tuple[str, dict[str, Any]]] = []

    def start(self) -> None:
        self.start_count += 1

    def close(self) -> None:
        self.close_count += 1
        if self.context is not None:
            self.context.close()
            self.context = None
        if self.browser is not None:
            self.browser.close()
            self.browser = None

    def fetch(self, url: str, **options: Any) -> object:
        self.fetch_calls.append((url, options))
        options["page_action"](self.page)
        return self.response


def _install_session_factory(
    monkeypatch: pytest.MonkeyPatch,
    name: str,
    session: _FakeSession,
) -> list[dict[str, Any]]:
    constructor_calls: list[dict[str, Any]] = []

    def factory(**options: Any) -> _FakeSession:
        constructor_calls.append(options)
        return session

    monkeypatch.setattr(scrapling_adapter, name, factory)
    return constructor_calls


def test_dynamic_session_supports_persistent_profile_and_fetch_result(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    browser = _FakeBrowser()
    context = _FakeContext(browser=browser)
    page = object()
    response = object()
    session = _FakeSession(context=context, page=page, response=response)
    calls = _install_session_factory(monkeypatch, "DynamicSession", session)

    with ScraplingBrowser(user_data_dir="profile-a") as client:
        actual_response, result = client.run_fetch_action(
            "https://example.test",
            lambda actual_page: (actual_page, "done"),
            wait=0,
        )

        assert client.session is session
        assert client.browser is browser
        assert client.context is context
        assert client.page is page
        assert actual_response is response
        assert result == (page, "done")

    assert calls == [{"user_data_dir": "profile-a", "retries": 1}]
    assert session.start_count == 1
    assert session.close_count == 1
    assert context.close_count == 1
    assert session.fetch_calls[0][0] == "https://example.test"
    assert session.fetch_calls[0][1]["wait"] == 0


def test_stealthy_session_is_selected(monkeypatch: pytest.MonkeyPatch) -> None:
    context = _FakeContext()
    session = _FakeSession(context=context)
    calls = _install_session_factory(monkeypatch, "StealthySession", session)

    browser = ScraplingBrowser(stealthy=True, user_data_dir="profile-b").start()
    browser.close()

    assert calls == [{"user_data_dir": "profile-b", "retries": 1}]
    assert session.start_count == 1


def test_run_fetch_action_propagates_scrapling_swallowed_error(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    context = _FakeContext()
    session = _FakeSession(context=context)
    _install_session_factory(monkeypatch, "DynamicSession", session)
    expected = ValueError("action failed")

    with ScraplingBrowser() as client:
        with pytest.raises(ValueError) as captured:
            client.run_fetch_action(
                "https://example.test",
                lambda _page: (_ for _ in ()).throw(expected),
            )

    assert captured.value is expected


def test_context_manager_preserves_action_error_when_state_export_target_closed(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    context = _ClosedOnExportContext()
    session = _FakeSession(context=context)
    _install_session_factory(monkeypatch, "DynamicSession", session)
    expected = ValueError("action failed")

    with pytest.raises(ValueError) as captured:
        with ScraplingBrowser(cookie_file=tmp_path / "cookies.json"):
            raise expected

    assert captured.value is expected
    assert session.close_count == 1


def test_explicit_close_ignores_target_closed_state_export(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    context = _ClosedOnExportContext()
    session = _FakeSession(context=context)
    _install_session_factory(monkeypatch, "DynamicSession", session)

    client = ScraplingBrowser(cookie_file=tmp_path / "cookies.json").start()
    client.close()

    assert session.close_count == 1


def test_cdp_reuses_only_existing_context_and_preserves_browser(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    existing_context = _FakeContext()
    created_context = _FakeContext()
    remote_browser = _FakeBrowser([existing_context, created_context])
    session = _FakeSession(context=created_context, browser=remote_browser)
    _install_session_factory(monkeypatch, "DynamicSession", session)

    client = ScraplingBrowser(
        cdp_url="http://127.0.0.1:9222",
        reuse_existing_cdp_context=True,
    ).start()

    assert created_context.close_count == 1
    assert client.browser is remote_browser
    assert client.context is existing_context
    assert session.context is existing_context

    client.close()

    assert session.close_count == 1
    assert existing_context.close_count == 0
    assert remote_browser.close_count == 0


@pytest.mark.parametrize("existing_context_count", [0, 2])
def test_cdp_reuse_requires_exactly_one_existing_context(
    monkeypatch: pytest.MonkeyPatch,
    existing_context_count: int,
) -> None:
    existing_contexts = [_FakeContext() for _ in range(existing_context_count)]
    created_context = _FakeContext()
    remote_browser = _FakeBrowser([*existing_contexts, created_context])
    session = _FakeSession(context=created_context, browser=remote_browser)
    _install_session_factory(monkeypatch, "DynamicSession", session)

    with pytest.raises(RuntimeError, match="exactly one existing context"):
        ScraplingBrowser(
            cdp_url="http://127.0.0.1:9222",
            reuse_existing_cdp_context=True,
        ).start()

    assert created_context.close_count == 1
    assert remote_browser.close_count == 0
    assert all(context.close_count == 0 for context in existing_contexts)


def test_browser_timeout_detection_uses_engine_type_metadata() -> None:
    playwright_timeout = type(
        "TimeoutError",
        (Exception,),
        {"__module__": "playwright._impl._errors"},
    )
    patchright_timeout = type(
        "TimeoutError",
        (Exception,),
        {"__module__": "patchright._impl._errors"},
    )
    unrelated_timeout = type(
        "TimeoutError",
        (Exception,),
        {"__module__": "unrelated.browser"},
    )

    assert is_browser_timeout_error(BrowserTimeoutError())
    assert is_browser_timeout_error(playwright_timeout())
    assert is_browser_timeout_error(patchright_timeout())
    assert not is_browser_timeout_error(TimeoutError())
    assert not is_browser_timeout_error(unrelated_timeout())


def test_cdp_reuse_requires_cdp_url() -> None:
    with pytest.raises(ValueError, match="requires cdp_url"):
        ScraplingBrowser(reuse_existing_cdp_context=True)


def test_bound_cookie_and_storage_state_are_restored_and_persisted(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    cookie_file = tmp_path / "cookies.json"
    storage_file = tmp_path / "storage.json"
    cookie_file.write_text(
        json.dumps([{"name": "loaded", "value": "cookie"}]),
        encoding="utf-8",
    )
    storage_file.write_text(
        json.dumps({"cookies": [], "origins": [{"origin": "https://example.test"}]}),
        encoding="utf-8",
    )
    context = _FakeContext()
    session = _FakeSession(context=context)
    _install_session_factory(monkeypatch, "DynamicSession", session)

    with ScraplingBrowser(
        cookie_file=cookie_file,
        storage_state_file=storage_file,
    ):
        assert context.added_cookies == [
            [{"name": "loaded", "value": "cookie"}]
        ]
        assert context.restored_states == [
            {"cookies": [], "origins": [{"origin": "https://example.test"}]}
        ]

    assert json.loads(cookie_file.read_text(encoding="utf-8")) == [
        {"name": "saved", "value": "cookie"}
    ]
    assert json.loads(storage_file.read_text(encoding="utf-8")) == {
        "cookies": [{"name": "saved", "value": "cookie"}],
        "origins": [],
    }


def test_borrowed_cdp_context_does_not_import_or_export_bound_state(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    cookie_file = tmp_path / "cookies.json"
    cookie_file.write_text("[]", encoding="utf-8")
    existing_context = _FakeContext()
    created_context = _FakeContext()
    browser = _FakeBrowser([existing_context, created_context])
    session = _FakeSession(context=created_context, browser=browser)
    _install_session_factory(monkeypatch, "DynamicSession", session)

    with ScraplingBrowser(
        cdp_url="http://127.0.0.1:9222",
        reuse_existing_cdp_context=True,
        cookie_file=cookie_file,
    ):
        pass

    assert cookie_file.read_text(encoding="utf-8") == "[]"
    assert existing_context.added_cookies == []