remotivelabs.topology.testing.hamcrest.await_at_most

Hamcrest-specific retry utilities for asynchronous testing.

class AwaitAtMost(typing.Protocol):

Protocol for await until functionality.

AwaitAtMost(*args, **kwargs)
async def until( self, func: Callable[[], Any], matcher: hamcrest.core.matcher.Matcher[typing.Any]) -> None:
def await_at_most( seconds: float) -> AwaitAtMost:

Creates a chainable assertion helper that retries until a hamcrest matcher is satisfied or timeout.

Provides a fluent syntax for asynchronous polling assertions. Retries calling a function until its result matches the hamcrest matcher or the timeout is reached.

Arguments:
  • seconds: The maximum duration to wait before giving up, in seconds.
Example:
from hamcrest import equal_to, greater_than

async def test_polling():
    counter = {"value": 0}

    async def increment():
        counter["value"] += 1
        return counter["value"]

    # Wait up to 5 seconds for counter to reach 10
    await await_at_most(5).until(lambda: counter["value"], equal_to(10))

    # Works with async functions too
    await await_at_most(2).until(increment, greater_than(5))
Raises:
  • TimeoutError: If the matcher condition is not satisfied within the timeout.