Async & eventual assertions

Poll a callable until an assertion passes or the timeout expires. Start with eventually() on a callable value, chain the assertion you expect to eventually hold, and await the result. See Testing for usage.

eventually

eventually(
    *, timeout: float = 5.0, interval: float = 0.5
) -> AsyncAssertionBuilder

Switch to async polling mode for eventual-consistency assertions.

The current val must be a callable (sync or async). Returns an AsyncAssertionBuilder whose assertion methods are coroutines that poll val() until the assertion passes or timeout expires.

Parameters:

Name	Type	Description	Default
timeout	float	maximum seconds to keep retrying (default 5.0)	5.0
interval	float	seconds between retries (default 0.5)	0.5

Examples:

Usage:

import asyncio
from assertpy2 import assert_that

counter = {"n": 0}

def get_count():
    counter["n"] += 1
    return counter["n"]

asyncio.run(
    assert_that(get_count).eventually(timeout=2).is_equal_to(3)
)

Returns:

Name	Type	Description
AsyncAssertionBuilder	AsyncAssertionBuilder	an async builder whose assertion methods are awaitable

Raises:

Type	Description
TypeError	if val is not callable

Source code in assertpy2/assertpy.py

def eventually(self, *, timeout: float = 5.0, interval: float = 0.5) -> AsyncAssertionBuilder:
    """Switch to async polling mode for eventual-consistency assertions.

    The current ``val`` must be a callable (sync or async).  Returns an
    `AsyncAssertionBuilder` whose assertion
    methods are coroutines that poll ``val()`` until the assertion passes or
    ``timeout`` expires.

    Args:
        timeout: maximum seconds to keep retrying (default ``5.0``)
        interval: seconds between retries (default ``0.5``)

    Examples:
        Usage:

            import asyncio
            from assertpy2 import assert_that

            counter = {"n": 0}

            def get_count():
                counter["n"] += 1
                return counter["n"]

            asyncio.run(
                assert_that(get_count).eventually(timeout=2).is_equal_to(3)
            )

    Returns:
        AsyncAssertionBuilder: an async builder whose assertion methods are awaitable

    Raises:
        TypeError: if ``val`` is not callable
    """
    if not callable(self.val):
        raise TypeError("val must be callable when using eventually()")
    return AsyncAssertionBuilder(
        self.val,
        builder_func=_builder,
        description=self.description,
        timeout=timeout,
        interval=interval,
    )

Async assertion builder that polls a callable until an assertion passes or timeout expires.

Do not instantiate directly; use eventually() instead.

Parameters:

Name	Type	Description	Default
func	Callable	a sync or async callable that produces the value to test	required
builder_func	Callable	factory function to create assertion builders (receives val, description)	required
description	str	optional error description forwarded to the builder	''
timeout	float	maximum seconds to keep retrying	5.0
interval	float	seconds between retries	0.5

Source code in assertpy2/async_assertions.py

def __init__(
    self,
    func: Callable,
    *,
    builder_func: Callable,
    description: str = "",
    timeout: float = 5.0,
    interval: float = 0.5,
):
    self._func = func
    self._builder_func = builder_func
    self._description = description
    self._timeout = timeout
    self._interval = interval

within

within(timeout: float) -> Self

Override the timeout (in seconds).

Source code in assertpy2/async_assertions.py

def within(self, timeout: float) -> Self:
    """Override the timeout (in seconds)."""
    self._timeout = timeout
    return self

every

every(interval: float) -> Self

Override the polling interval (in seconds).

Source code in assertpy2/async_assertions.py

def every(self, interval: float) -> Self:
    """Override the polling interval (in seconds)."""
    self._interval = interval
    return self