test: consolidate polling into a single shared helper and deflake tests

Replace the two integration polling helpers with a single poll_until_condition
in a shared tests/_utils.py, matching the helper in apify-client-python (with
backoff_factor covering the exponential-backoff use case). Use it across
integration, e2e, and unit tests instead of hand-rolled retry loops and fixed
sleeps, and fix the flaky webhook e2e test by keeping the client run alive
briefly after add_webhook so the run-succeeded event cannot outrun the webhook
registration.

Author: vdusek

tests/__init__.py

@@ -2,7 +2,6 @@
 
 import asyncio
 import inspect
-import logging
 import time
 from typing import TYPE_CHECKING, TypeVar, cast, overload
 
@@ -11,76 +10,27 @@
 if TYPE_CHECKING:
     from collections.abc import Awaitable, Callable
 
-logger = logging.getLogger(__name__)
-
 T = TypeVar('T')
 
 
 async def _maybe_await(value: Awaitable[T] | T) -> T:
     """Await `value` if it is awaitable, otherwise return it unchanged.
 
-    Lets `call_with_exp_backoff` and `poll_until_condition` accept both sync and async callables.
+    Lets `poll_until_condition` accept both sync and async callables.
     """
     if inspect.isawaitable(value):
         return await cast('Awaitable[T]', value)
     return cast('T', value)
 
 
-@overload
-async def call_with_exp_backoff(
-    fn: Callable[[], Awaitable[T]],
-    condition: Callable[[T], bool] = ...,
-    *,
-    max_retries: int = ...,
-    base_delay: float = ...,
-) -> T: ...
-@overload
-async def call_with_exp_backoff(
-    fn: Callable[[], T],
-    condition: Callable[[T], bool] = ...,
-    *,
-    max_retries: int = ...,
-    base_delay: float = ...,
-) -> T: ...
-async def call_with_exp_backoff(
-    fn: Callable[[], Awaitable[T] | T],
-    condition: Callable[[T], bool] = bool,
-    *,
-    max_retries: int = 5,
-    base_delay: float = 1.0,
-) -> T:
-    """Call `fn`, retrying with exponential backoff until `condition(result)` is True.
-
-    Calls `fn` and checks whether `condition` holds for its result. If it does not, `fn` is retried up to
-    `max_retries` times, sleeping `base_delay * 2 ** attempt` seconds before each retry. The last result is
-    returned regardless of whether the condition was ever satisfied, so the caller can run its own assertion.
-
-    This is useful for eventually-consistent APIs where a freshly added, reclaimed, or handled item may take a
-    moment to become visible (see https://github.com/apify/apify-sdk-python/issues/808). The default condition
-    checks for a truthy result. Pass `max_retries=0` to call `fn` exactly once without any retries.
-
-    Unlike `poll_until_condition`, the delay between attempts grows exponentially rather than staying constant.
-    """
-    result = await _maybe_await(fn())
-    for attempt in range(max_retries):
-        if condition(result):
-            return result
-        delay = base_delay * 2**attempt
-        logger.info(
-            'Condition not met for %r, retrying in %ss (attempt %d/%d).', result, delay, attempt + 1, max_retries
-        )
-        await asyncio.sleep(delay)
-        result = await _maybe_await(fn())
-    return result
-
-
 @overload
 async def poll_until_condition(
     fn: Callable[[], Awaitable[T]],
     condition: Callable[[T], bool] = ...,
     *,
     timeout: float = ...,
     poll_interval: float = ...,
+    backoff_factor: float = ...,
 ) -> T: ...
 @overload
 async def poll_until_condition(
@@ -89,31 +39,36 @@ async def poll_until_condition(
     *,
     timeout: float = ...,
     poll_interval: float = ...,
+    backoff_factor: float = ...,
 ) -> T: ...
 async def poll_until_condition(
     fn: Callable[[], Awaitable[T] | T],
     condition: Callable[[T], bool] = bool,
     *,
-    timeout: float = 60,
-    poll_interval: float = 5,
+    timeout: float = 5,
+    poll_interval: float = 1,
+    backoff_factor: float = 1,
 ) -> T:
     """Poll `fn` until `condition(result)` is True or the timeout expires.
 
     Polls `fn` at `poll_interval`-second intervals until `condition` is satisfied or `timeout` seconds have elapsed.
     Returns the last polled result regardless of whether the condition was met, so the caller can run its own
-    assertion. The default condition checks for a truthy result.
+    assertion. The default condition checks for a truthy result. Pass `timeout=0` to call `fn` exactly once.
 
-    Use this instead of a fixed `asyncio.sleep` when waiting for eventually-consistent state (e.g. request queue
-    stats) that may take a variable amount of time to propagate. Unlike `call_with_exp_backoff`, the interval
-    between polls stays constant.
+    Use this instead of a fixed `asyncio.sleep` when waiting for eventually-consistent state (e.g. a freshly
+    created resource appearing in a listing) that may take a variable amount of time to propagate. For highly
+    variable wait times (e.g. an Actor run container starting up), pass `backoff_factor` > 1 to multiply the
+    interval after each poll, covering a long timeout with few calls.
     """
     deadline = time.monotonic() + timeout
+    delay = poll_interval
     result = await _maybe_await(fn())
     while not condition(result):
         remaining = deadline - time.monotonic()
         if remaining <= 0:
             break
-        await asyncio.sleep(min(poll_interval, remaining))
+        await asyncio.sleep(min(delay, remaining))
+        delay *= backoff_factor
         result = await _maybe_await(fn())
     return result
 

tests/integration/_utils.py tests/_utils.pytests/integration/_utils.py renamed to tests/_utils.py

@@ -17,7 +17,7 @@
 from crawlee import service_locator
 
 import apify._actor
-from ._utils import generate_unique_resource_name
+from .._utils import generate_unique_resource_name
 from apify._models import ActorRun
 from apify.storage_clients._apify._alias_resolving import AliasResolver
 

tests/e2e/_utils.py

@@ -7,7 +7,7 @@
 from apify_shared.consts import ActorPermissionLevel
 from crawlee._utils.crypto import crypto_random_object_id
 
-from ._utils import generate_unique_resource_name
+from .._utils import generate_unique_resource_name, poll_until_condition
 from apify import Actor
 from apify._models import ActorRun
 
@@ -425,6 +425,8 @@ def do_POST(self) -> None:
             await Actor.set_value('WEBHOOK_BODY', webhook_body)
 
     async def main_client() -> None:
+        import asyncio
+
         from apify import Webhook, WebhookEventType
 
         async with Actor:
@@ -438,6 +440,12 @@ async def main_client() -> None:
                 )
             )
 
+            # Keep the run alive for a moment after registering the webhook. Without this, the run finishes
+            # just milliseconds later and the platform may process the run-succeeded event before the freshly
+            # added ad-hoc webhook has propagated, in which case the webhook never fires and the server Actor
+            # waits until it times out.
+            await asyncio.sleep(5)
+
     server_actor, client_actor = await asyncio.gather(
         make_actor(label='add-webhook-server', main_func=main_server),
         make_actor(label='add-webhook-client', main_func=main_client),
@@ -446,10 +454,15 @@ async def main_client() -> None:
     server_actor_run = await server_actor.start()
     server_actor_container_url = server_actor_run['containerUrl']
 
-    server_actor_initialized = await server_actor.last_run().key_value_store().get_record('INITIALIZED')
-    while not server_actor_initialized:
-        server_actor_initialized = await server_actor.last_run().key_value_store().get_record('INITIALIZED')
-        await asyncio.sleep(1)
+    # Wait for the server Actor's container to start up and bind its HTTP server. The startup time is highly
+    # variable (image pull, container creation), so poll with a growing interval instead of a fixed sleep.
+    server_actor_initialized = await poll_until_condition(
+        lambda: server_actor.last_run().key_value_store().get_record('INITIALIZED'),
+        timeout=300,
+        poll_interval=1,
+        backoff_factor=1.5,
+    )
+    assert server_actor_initialized is not None, 'The server Actor did not initialize in time.'
 
     ac_run_result = await run_actor(
         client_actor,