refactor: keep networkidle timeout handling inline, mock-based tests

2026-09-24 14:20:08 +01:00 · 2026-08-08 00:37:09 +02:00
parent 7b904a5ffd
commit 490e2fad97
2 changed files with 33 additions and 67 deletions
@@ -7,7 +7,6 @@ from typing import Annotated

 from fastapi import APIRouter, Depends, HTTPException
 from fastapi.responses import RedirectResponse
-from playwright.async_api import Page
 from playwright.async_api import TimeoutError as PlaywrightTimeoutError
 from playwright_captcha import CaptchaType

@@ -52,23 +51,6 @@ async def health_check(sb: BrowserDep):
    return HealthcheckResponse(user_agent=health_check_request.solution.user_agent)


-async def _wait_for_networkidle(page: Page, timer: TimeoutTimer) -> None:
-    """
-    Wait for the network to go idle; a timeout is non-fatal.
-
-    Some sites keep background connections open (analytics beacons,
-    websockets, ...), so ``networkidle`` may never settle even though the page
-    is fully usable once ``domcontentloaded`` has fired. Log and continue
-    instead of failing the request.
-    """
-    try:
-        await page.wait_for_load_state("networkidle", timeout=timer.remaining() * 1000)
-    except PlaywrightTimeoutError:
-        logger.info(
-            "networkidle timed out after domcontentloaded; continuing with loaded page"
-        )
-
-
@router.post("/v1")
 async def read_item(request: LinkRequest, dep: BrowserDep) -> LinkResponse:
    """Handle POST requests."""
@@ -111,7 +93,18 @@ async def read_item(request: LinkRequest, dep: BrowserDep) -> LinkResponse:
            status = HTTPStatus.OK
            logger.debug("Challenge solved successfully.")
        else:
-            await _wait_for_networkidle(dep.page, timer)
+            # Best-effort: some sites keep background connections open
+            # (analytics beacons, websockets, ...), so ``networkidle`` may
+            # never settle. The page is fully usable once ``domcontentloaded``
+            # has fired, so log and continue instead of failing the request.
+            try:
+                await dep.page.wait_for_load_state(
+                    "networkidle", timeout=timer.remaining() * 1000
+                )
+            except PlaywrightTimeoutError:
+                logger.info(
+                    "networkidle timed out after domcontentloaded; continuing with loaded page"
+                )
    except (TimeoutError, PlaywrightTimeoutError) as e:
        logger.error("Timed out while loading the page or solving the challenge")
        raise HTTPException(
@@ -1,6 +1,6 @@
 from http import HTTPStatus
 from json import JSONDecodeError
-from types import SimpleNamespace
+from unittest.mock import AsyncMock, MagicMock

 import httpx
 import pytest
@@ -106,55 +106,28 @@ def test_max_timeout_normalization(payload: dict, expected: int):
    assert request.max_timeout == expected


-class FakeLoadStateError(PlaywrightTimeoutError):
-    """Raised by FakePage.wait_for_load_state for configured failing states."""
+def fake_dep(*, fail_states: set[str] | None = None) -> BrowserDepClass:
+    """Build a browser dependency triple backed by mocks."""
+    page = AsyncMock()
+    page.url = "https://example.test/login"
+    page.goto.return_value = MagicMock(
+        status=HTTPStatus.OK, headers={"content-type": "text/html"}
+    )
+    page.title.return_value = "Login"
+    page.evaluate.return_value = "UnitTestBrowser/1.0"
+    page.content.return_value = "<html><title>Login</title></html>"

-
-class FakePage:
-    """Playwright Page double; load-state waits fail for configured states."""
-
-    url = "https://example.test/login"
-
-    def __init__(self, *, fail_states: set[str] | None = None) -> None:
-        """Create a page whose waits fail for the given load states."""
-        self.fail_states = fail_states or set()
-
-    async def goto(self, _url: str, **_kwargs: object) -> SimpleNamespace:
-        """Return a successful navigation result."""
-        return SimpleNamespace(
-            status=HTTPStatus.OK, headers={"content-type": "text/html"}
-        )
-
-    async def wait_for_load_state(self, state: str, **_kwargs: object) -> None:
-        """Fail for configured states; otherwise do nothing."""
-        if state in self.fail_states:
+    def wait_for_load_state(state: str, **_kwargs: object) -> None:
+        """Fail the wait when asked for a configured state."""
+        if state in (fail_states or set()):
            message = "load state wait timed out"
-            raise FakeLoadStateError(message)
+            raise PlaywrightTimeoutError(message)

-    async def title(self) -> str:
-        """Return a title that is not a challenge title."""
-        return "Login"
+    page.wait_for_load_state.side_effect = wait_for_load_state

-    async def evaluate(self, _expression: str) -> str:
-        """Return the user agent the API reports."""
-        return "UnitTestBrowser/1.0"
-
-    async def content(self) -> str:
-        """Return the HTML body the API should return."""
-        return "<html><title>Login</title></html>"
-
-
-class FakeContext:
-    """Playwright BrowserContext double."""
-
-    async def cookies(self) -> list[object]:
-        """Return no cookies."""
-        return []
-
-
-def make_dep(page: FakePage) -> BrowserDepClass:
-    """Build the browser dependency triple around a fake page."""
-    return BrowserDepClass(page=page, solver=SimpleNamespace(), context=FakeContext())
+    context = AsyncMock()
+    context.cookies.return_value = []
+    return BrowserDepClass(page=page, solver=AsyncMock(), context=context)


@pytest.mark.asyncio
@@ -162,7 +135,7 @@ async def test_networkidle_timeout_after_domcontentloaded_returns_content():
    """Pages that never go idle after DOM load must still return their content."""
    response = await read_item(
        LinkRequest(url="https://example.test/login"),
-        make_dep(FakePage(fail_states={"networkidle"})),
+        fake_dep(fail_states={"networkidle"}),
    )

    assert response.status == "ok"
@@ -176,7 +149,7 @@ async def test_domcontentloaded_timeout_returns_408():
    with pytest.raises(HTTPException) as exc:
        await read_item(
            LinkRequest(url="https://example.test/login"),
-            make_dep(FakePage(fail_states={"domcontentloaded"})),
+            fake_dep(fail_states={"domcontentloaded"}),
        )

    assert exc.value.status_code == HTTPStatus.REQUEST_TIMEOUT