test: Deflake throttled-request deferral tests with a stopped throttler clock (#2275)

`test_static_handler_raised_throttled_error_is_deferred` failed on macOS
([run](https://github.com/apify/crawlee-python/actions/runs/37274209355/job/111647535443)).
The test put the domain into a 50 ms backoff and raised
`RequestThrottledError`. The crawler defers the request only if
`is_throttled()` still holds when it checks, and on a loaded runner the
backoff ran out first. The request then went down the ordinary-error
path and fell back to the browser.

A new helper, `open_throttler_held_until_reclaim()` in
`tests/unit/utils.py`, opens the throttler on a stopped `Mock` clock
with a 1-minute backoff. The domain is always still held back when the
crawler checks. When the crawler reclaims the deferred request, the
clock moves past the backoff, so the retry is dispatched right away.

The same race also failed
`test_handler_raised_throttled_error_is_deferred` in the basic crawler
suite
([run](https://github.com/apify/crawlee-python/actions/runs/37325111977/job/111813759733),
on #2274).

Six deferral tests in the basic, HTTP, Playwright and adaptive crawler
suites now use the helper. Their assertions are unchanged, and all of
them fail when `_is_held_back` is forced to `False`. The stall tests
still use the 50 ms backoff, since they need real time to pass through
`max_domain_stall`.

Closes: #2271

*✍️ Drafted by Claude Code*

Author: vdusek

tests/unit/crawlers/_adaptive_playwright/test_adaptive_playwright_crawler.py

@@ -32,11 +32,11 @@
 from crawlee.crawlers._adaptive_playwright._adaptive_playwright_crawling_context import AdaptiveContextError
 from crawlee.errors import RequestThrottledError
 from crawlee.http_clients import HttpClient, HttpCrawlingResult
-from crawlee.request_loaders import ThrottlingRequestManager
 from crawlee.sessions import SessionPool
 from crawlee.statistics import Statistics
 from crawlee.storage_clients import SqlStorageClient
 from crawlee.storages import KeyValueStore, RequestQueue
+from tests.unit.utils import open_throttler_held_until_reclaim
 
 if TYPE_CHECKING:
     from collections.abc import AsyncGenerator, Iterator
@@ -941,16 +941,10 @@ async def request_handler(_context: AdaptivePlaywrightCrawlingContext) -> None:
         mocked_handler.assert_called()
 
 
-async def test_throttled_429_is_deferred() -> None:
+async def test_throttled_429_is_deferred(monkeypatch: pytest.MonkeyPatch) -> None:
     """A throttled 429 from the browser sub crawler is retried later without spending a retry."""
     url = 'https://throttled.placeholder.com/page'
-    throttler = ThrottlingRequestManager(
-        await RequestQueue.open(),
-        domains=['throttled.placeholder.com'],
-        request_manager_opener=RequestQueue.open,
-        base_delay=timedelta(milliseconds=50),
-        max_delay=timedelta(milliseconds=100),
-    )
+    throttler = await open_throttler_held_until_reclaim(monkeypatch)
     failed_request_handler = AsyncMock()
     crawler = AdaptivePlaywrightCrawler.with_beautifulsoup_static_parser(
         rendering_type_predictor=_SimpleRenderingTypePredictor(
@@ -980,15 +974,9 @@ async def handler(context: AdaptivePlaywrightCrawlingContext) -> None:
     assert handled[0].request.retry_count == 0
 
 
-async def test_throttled_static_429_is_deferred() -> None:
+async def test_throttled_static_429_is_deferred(monkeypatch: pytest.MonkeyPatch) -> None:
     """A throttled 429 from the static sub crawler defers the request without falling back to the browser."""
-    throttler = ThrottlingRequestManager(
-        await RequestQueue.open(),
-        domains=['throttled.placeholder.com'],
-        request_manager_opener=RequestQueue.open,
-        base_delay=timedelta(milliseconds=50),
-        max_delay=timedelta(milliseconds=100),
-    )
+    throttler = await open_throttler_held_until_reclaim(monkeypatch)
     http_client = AsyncMock(spec=HttpClient)
     http_client.crawl.side_effect = [
         HttpCrawlingResult(http_response=Mock(status_code=status, headers={}, read=AsyncMock(return_value=b'')))
@@ -1025,16 +1013,10 @@ async def handler(context: AdaptivePlaywrightCrawlingContext) -> None:
     assert handled[0].request.retry_count == 0
 
 
-async def test_static_handler_raised_throttled_error_is_deferred() -> None:
+async def test_static_handler_raised_throttled_error_is_deferred(monkeypatch: pytest.MonkeyPatch) -> None:
     """A `RequestThrottledError` from the static handler defers the request without falling back to the browser."""
     url = 'https://throttled.placeholder.com/page'
-    throttler = ThrottlingRequestManager(
-        await RequestQueue.open(),
-        domains=['throttled.placeholder.com'],
-        request_manager_opener=RequestQueue.open,
-        base_delay=timedelta(milliseconds=50),
-        max_delay=timedelta(milliseconds=100),
-    )
+    throttler = await open_throttler_held_until_reclaim(monkeypatch)
     http_client = AsyncMock(spec=HttpClient)
     http_client.crawl.return_value = HttpCrawlingResult(
         http_response=Mock(status_code=200, headers={}, read=AsyncMock(return_value=b''))

tests/unit/crawlers/_basic/test_basic_crawler.py

@@ -40,7 +40,7 @@
 from crawlee.statistics import FinalStatistics, StatisticsState
 from crawlee.storage_clients import FileSystemStorageClient, MemoryStorageClient
 from crawlee.storages import Dataset, KeyValueStore, RequestQueue
-from tests.unit.utils import poll_until_condition
+from tests.unit.utils import open_throttler_held_until_reclaim, poll_until_condition
 
 if TYPE_CHECKING:
     from collections.abc import Callable, Sequence
@@ -2745,9 +2745,9 @@ async def open_throttler() -> ThrottlingRequestManager[RequestQueue]:
     )
 
 
-async def test_handler_raised_throttled_error_is_deferred() -> None:
+async def test_handler_raised_throttled_error_is_deferred(monkeypatch: pytest.MonkeyPatch) -> None:
     """A `RequestThrottledError` from the handler costs neither a retry nor session reputation."""
-    throttler = await open_throttler()
+    throttler = await open_throttler_held_until_reclaim(monkeypatch)
     failed_request_handler = AsyncMock()
     crawler = BasicCrawler(
         request_manager=throttler,
@@ -2960,9 +2960,11 @@ async def handler(context: BasicCrawlingContext) -> None:
         pytest.param(['error', 'error', '429', 'ok'], [0, 0, 1], id='after-retries'),
     ],
 )
-async def test_deferral_not_counted_as_retry(outcomes: list[str], expected_histogram: list[int]) -> None:
+async def test_deferral_not_counted_as_retry(
+    outcomes: list[str], expected_histogram: list[int], monkeypatch: pytest.MonkeyPatch
+) -> None:
     """A deferral doesn't show up in the retry histogram, while earlier retries still do."""
-    throttler = await open_throttler()
+    throttler = await open_throttler_held_until_reclaim(monkeypatch)
     crawler = BasicCrawler(request_manager=throttler, max_request_retries=3)
 
     @crawler.router.default_handler

tests/unit/crawlers/_http/test_http_crawler.py

@@ -18,6 +18,7 @@
 from crawlee.statistics import Statistics
 from crawlee.storages import RequestQueue
 from tests.unit.server_endpoints import HELLO_WORLD
+from tests.unit.utils import open_throttler_held_until_reclaim
 
 if TYPE_CHECKING:
     from collections.abc import Awaitable, Callable
@@ -705,15 +706,9 @@ def crawl_result(status_code: int) -> HttpCrawlingResult:
         pytest.param({'ignore_http_error_status_codes': {429}}, id='ignored_429'),
     ],
 )
-async def test_throttled_429_is_deferred(crawler_kwargs: dict[str, Any]) -> None:
+async def test_throttled_429_is_deferred(crawler_kwargs: dict[str, Any], monkeypatch: pytest.MonkeyPatch) -> None:
     """A throttled 429 is retried later without spending a retry, a session rotation or session reputation."""
-    throttler = ThrottlingRequestManager(
-        await RequestQueue.open(),
-        domains=['throttled.placeholder.com'],
-        request_manager_opener=RequestQueue.open,
-        base_delay=timedelta(milliseconds=50),
-        max_delay=timedelta(milliseconds=100),
-    )
+    throttler = await open_throttler_held_until_reclaim(monkeypatch)
     http_client = AsyncMock(spec=HttpClient)
     http_client.crawl.side_effect = [crawl_result(429), crawl_result(200)]
     failed_request_handler = AsyncMock()

tests/unit/crawlers/_playwright/test_playwright_crawler.py

@@ -38,12 +38,12 @@
 from crawlee.fingerprint_suite._header_generator import fingerprint_browser_type_from_playwright_browser_type
 from crawlee.http_clients import ImpitHttpClient
 from crawlee.proxy_configuration import ProxyConfiguration
-from crawlee.request_loaders import ThrottlingRequestManager
 from crawlee.sessions import Session, SessionPool
 from crawlee.statistics import Statistics
 from crawlee.statistics._error_snapshotter import ErrorSnapshotter
 from crawlee.storages import RequestQueue
 from tests.unit.server_endpoints import GENERIC_RESPONSE, HELLO_WORLD
+from tests.unit.utils import open_throttler_held_until_reclaim
 
 if TYPE_CHECKING:
     from pathlib import Path
@@ -1362,16 +1362,10 @@ async def failed_handler(context: BasicCrawlingContext | PlaywrightCrawlingConte
     assert set(failed_handler_calls) <= {HELLO_WORLD.decode()}
 
 
-async def test_throttled_429_is_deferred() -> None:
+async def test_throttled_429_is_deferred(monkeypatch: pytest.MonkeyPatch) -> None:
     """A throttled 429 is retried later without spending a retry, a session rotation or session reputation."""
     url = 'https://throttled.placeholder.com/page'
-    throttler = ThrottlingRequestManager(
-        await RequestQueue.open(),
-        domains=['throttled.placeholder.com'],
-        request_manager_opener=RequestQueue.open,
-        base_delay=timedelta(milliseconds=50),
-        max_delay=timedelta(milliseconds=100),
-    )
+    throttler = await open_throttler_held_until_reclaim(monkeypatch)
     failed_request_handler = AsyncMock()
     crawler = PlaywrightCrawler(
         request_manager=throttler,

tests/unit/utils.py

@@ -5,18 +5,24 @@
 import sys
 import time
 from contextlib import asynccontextmanager
+from datetime import datetime, timedelta, timezone
 from typing import TYPE_CHECKING, Any, TypeVar, cast, overload
-from unittest.mock import AsyncMock, MagicMock
+from unittest.mock import AsyncMock, MagicMock, Mock
 
 import pytest
 
 from crawlee.http_clients._base import HttpClient, HttpResponse
+from crawlee.request_loaders import ThrottlingRequestManager, _throttling_request_manager
+from crawlee.storages import RequestQueue
 
 if TYPE_CHECKING:
     from collections.abc import AsyncIterator, Awaitable, Callable
 
     from yarl import URL
 
+    from crawlee import Request
+    from crawlee.storage_clients.models import ProcessedRequest
+
 T = TypeVar('T')
 
 run_alone_on_mac = pytest.mark.run_alone if sys.platform == 'darwin' else lambda x: x
@@ -60,6 +66,35 @@ async def read_stream() -> AsyncIterator[bytes]:
     return client, attempts
 
 
+async def open_throttler_held_until_reclaim(
+    monkeypatch: pytest.MonkeyPatch,
+) -> ThrottlingRequestManager[RequestQueue]:
+    """Open a throttler for `throttled.placeholder.com` whose backoff lasts until the crawler gives a request back.
+
+    The throttler runs on a stopped clock, so a backoff can't run out before the crawler checks whether the domain is
+    held back. Each reclaimed request moves the clock past the backoff, so a deferred request is dispatched right away.
+    """
+    backoff = timedelta(minutes=1)
+    clock = Mock()
+    clock.now.return_value = datetime.now(timezone.utc)
+    monkeypatch.setattr(_throttling_request_manager, 'datetime', clock)
+    throttler = ThrottlingRequestManager(
+        await RequestQueue.open(),
+        domains=['throttled.placeholder.com'],
+        request_manager_opener=RequestQueue.open,
+        base_delay=backoff,
+        max_delay=backoff,
+    )
+    reclaim_request = throttler.reclaim_request
+
+    async def reclaim_after_backoff(request: Request, *, forefront: bool = False) -> ProcessedRequest | None:
+        clock.now.return_value += backoff
+        return await reclaim_request(request, forefront=forefront)
+
+    monkeypatch.setattr(throttler, 'reclaim_request', reclaim_after_backoff)
+    return throttler
+
+
 async def maybe_await(value: Awaitable[T] | T) -> T:
     """Await `value` if it is awaitable, otherwise return it unchanged.