From 490e2fad97a5c70228deb9bdfc98d8f384a00263 Mon Sep 17 00:00:00 2001 From: ThePhaseless Date: Sat, 8 Aug 2026 00:37:09 +0200 Subject: [PATCH] refactor: keep networkidle timeout handling inline, mock-based tests --- src/endpoints.py | 31 ++++++++------------- tests/main_test.py | 69 ++++++++++++++-------------------------------- 2 files changed, 33 insertions(+), 67 deletions(-) diff --git a/src/endpoints.py b/src/endpoints.py index c7d0afd..49a9b29 100644 --- a/src/endpoints.py +++ b/src/endpoints.py @@ -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( diff --git a/tests/main_test.py b/tests/main_test.py index ac9c74b..df512db 100644 --- a/tests/main_test.py +++ b/tests/main_test.py @@ -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 = "Login" - -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 "Login" - - -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