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 - or use eventually_sync() for the same polling without an event loop. See Testing for usage.

eventually

eventually(
    *,
    timeout: float = 5.0,
    interval: float = 0.5,
    ignoring: type[Exception]
    | tuple[type[Exception], ...] = (),
    trace: bool = True,
) -> 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.

By default only a failing assertion is retried: any exception raised by val() itself propagates immediately. A probe that signals "not ready yet" by raising (a connection refused while a service boots, a record not yet visible) can be retried too by listing those exception types in ignoring.

Polling itself is always strict - retrying requires hard failures - but the final timeout failure honors the builder's mode: inside soft_assertions() it is collected instead of raised, and under assert_warn() it is logged.

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
ignoring	type[Exception] | tuple[type[Exception], ...]	an Exception subclass (or tuple of them) the polling loop retries instead of propagating (default: none)	()
trace	bool	record a PollTrace flight recorder attached to the timeout failure (default True); pass False to skip recording, for tight polling loops where per-poll snapshots of a heavy probed value are too costly	True

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)
)

Retry a probe that raises while the system under test is not ready yet:

await assert_that(get_order).eventually(timeout=10, ignoring=ConnectionError).has_status("PAID")

# or configure fluently on the returned builder
await assert_that(get_order).eventually().within(10).ignoring(ConnectionError).has_status("PAID")

Returns:

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

Raises:

Type	Description
TypeError	if val is not callable, or if ignoring contains anything that is not an Exception subclass

Source code in assertpy2/assertpy.py

def eventually(
    self,
    *,
    timeout: float = 5.0,
    interval: float = 0.5,
    ignoring: type[Exception] | tuple[type[Exception], ...] = (),
    trace: bool = True,
) -> 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.

    By default only a failing assertion is retried: any exception raised by ``val()`` itself
    propagates immediately.  A probe that signals "not ready yet" by raising (a connection refused
    while a service boots, a record not yet visible) can be retried too by listing those exception
    types in ``ignoring``.

    Polling itself is always strict - retrying *requires* hard failures - but the final timeout
    failure honors the builder's mode: inside
    [`soft_assertions()`][assertpy2.assertpy.soft_assertions] it is collected instead of raised,
    and under [`assert_warn()`][assertpy2.assertpy.assert_warn] it is logged.

    Args:
        timeout: maximum seconds to keep retrying (default ``5.0``)
        interval: seconds between retries (default ``0.5``)
        ignoring: an ``Exception`` subclass (or tuple of them) the polling loop retries instead of
            propagating (default: none)
        trace: record a [`PollTrace`][assertpy2.errors.PollTrace] flight recorder attached to the
            timeout failure (default ``True``); pass ``False`` to skip recording, for tight
            polling loops where per-poll snapshots of a heavy probed value are too costly

    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)
            )

        Retry a probe that raises while the system under test is not ready yet:

            await assert_that(get_order).eventually(timeout=10, ignoring=ConnectionError).has_status("PAID")

            # or configure fluently on the returned builder
            await assert_that(get_order).eventually().within(10).ignoring(ConnectionError).has_status("PAID")

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

    Raises:
        TypeError: if ``val`` is not callable, or if ``ignoring`` contains anything that is not an
            ``Exception`` subclass
    """
    if not callable(self.val):
        refuse(self.val, "callable, since eventually() polls it")
    return AsyncAssertionBuilder(
        self.val,
        builder_func=_builder,
        description=self.description,
        timeout=timeout,
        interval=interval,
        ignoring=_normalize_ignoring(ignoring),
        kind=self.kind,
        logger=self.logger,
        trace=trace,
    )

eventually_sync

eventually_sync(
    *,
    timeout: float = 5.0,
    interval: float = 0.5,
    ignoring: type[Exception]
    | tuple[type[Exception], ...] = (),
    trace: bool = True,
) -> SyncAssertionBuilder

Switch to blocking polling mode for eventual-consistency assertions, without asyncio.

The synchronous sibling of eventually(): the current val must be a sync callable, and the returned SyncAssertionBuilder exposes assertion methods that block the calling thread (via time.sleep) while polling val() until the assertion passes or timeout expires - no event loop and no await needed. A probe that returns an awaitable raises TypeError; poll async probes with eventually().

Retry, failure-mode, and diagnostics semantics are identical to eventually(): only a failing assertion (or an exception type listed in ignoring) is retried, the final timeout failure honors the builder's soft/warn mode, and it carries the same PollTrace flight recorder.

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
ignoring	type[Exception] | tuple[type[Exception], ...]	an Exception subclass (or tuple of them) the polling loop retries instead of propagating (default: none)	()
trace	bool	record a PollTrace flight recorder attached to the timeout failure (default True); pass False to skip recording, for tight polling loops where per-poll snapshots of a heavy probed value are too costly	True

Examples:

Usage:

from assertpy2 import assert_that

counter = {"n": 0}

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

assert_that(get_count).eventually_sync(timeout=2, interval=0.1).is_equal_to(3)

Retry a probe that raises while the system under test is not ready yet:

assert_that(get_order).eventually_sync(timeout=10, ignoring=ConnectionError).has_status("PAID")

# or configure fluently on the returned builder
assert_that(get_order).eventually_sync().within(10).ignoring(ConnectionError).has_status("PAID")

Returns:

Name	Type	Description
SyncAssertionBuilder	SyncAssertionBuilder	a blocking builder whose assertion methods poll on call

Raises:

Type	Description
TypeError	if val is not callable, or if ignoring contains anything that is not an Exception subclass

Source code in assertpy2/assertpy.py

def eventually_sync(
    self,
    *,
    timeout: float = 5.0,
    interval: float = 0.5,
    ignoring: type[Exception] | tuple[type[Exception], ...] = (),
    trace: bool = True,
) -> SyncAssertionBuilder:
    """Switch to blocking polling mode for eventual-consistency assertions, without asyncio.

    The synchronous sibling of [`eventually()`][assertpy2.assertpy.AssertionBuilder.eventually]:
    the current ``val`` must be a sync callable, and the returned
    `SyncAssertionBuilder` exposes assertion methods
    that block the calling thread (via ``time.sleep``) while polling ``val()`` until the
    assertion passes or ``timeout`` expires - no event loop and no ``await`` needed.  A probe
    that returns an awaitable raises ``TypeError``; poll async probes with ``eventually()``.

    Retry, failure-mode, and diagnostics semantics are identical to ``eventually()``: only a
    failing assertion (or an exception type listed in ``ignoring``) is retried, the final
    timeout failure honors the builder's soft/warn mode, and it carries the same
    [`PollTrace`][assertpy2.errors.PollTrace] flight recorder.

    Args:
        timeout: maximum seconds to keep retrying (default ``5.0``)
        interval: seconds between retries (default ``0.5``)
        ignoring: an ``Exception`` subclass (or tuple of them) the polling loop retries instead of
            propagating (default: none)
        trace: record a [`PollTrace`][assertpy2.errors.PollTrace] flight recorder attached to the
            timeout failure (default ``True``); pass ``False`` to skip recording, for tight
            polling loops where per-poll snapshots of a heavy probed value are too costly

    Examples:
        Usage:

            from assertpy2 import assert_that

            counter = {"n": 0}

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

            assert_that(get_count).eventually_sync(timeout=2, interval=0.1).is_equal_to(3)

        Retry a probe that raises while the system under test is not ready yet:

            assert_that(get_order).eventually_sync(timeout=10, ignoring=ConnectionError).has_status("PAID")

            # or configure fluently on the returned builder
            assert_that(get_order).eventually_sync().within(10).ignoring(ConnectionError).has_status("PAID")

    Returns:
        SyncAssertionBuilder: a blocking builder whose assertion methods poll on call

    Raises:
        TypeError: if ``val`` is not callable, or if ``ignoring`` contains anything that is not an
            ``Exception`` subclass
    """
    if not callable(self.val):
        refuse(self.val, "callable, since eventually_sync() polls it")
    return SyncAssertionBuilder(
        self.val,
        builder_func=_builder,
        description=self.description,
        timeout=timeout,
        interval=interval,
        ignoring=_normalize_ignoring(ignoring),
        kind=self.kind,
        logger=self.logger,
        trace=trace,
    )

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
ignoring	tuple[type[Exception], ...]	exception types the polling loop retries instead of propagating	()
kind	str | None	the failure mode of the final timeout failure (None/"soft"/"warn"); polling itself always retries on hard failures	None
logger	object	the logger for "warn" mode	None
trace	bool	record a PollTrace of the polling timeline (default True); False skips the flight recorder entirely	True

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,
    ignoring: tuple[type[Exception], ...] = (),
    kind: str | None = None,
    logger: object = None,
    trace: bool = True,
):
    self._func = func
    self._builder_func = builder_func
    self._description = description
    self._timeout = timeout
    self._interval = interval
    self._ignoring = ignoring
    self._kind = kind
    self._logger = logger
    self._trace = trace

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

ignoring

ignoring(*exceptions: type[Exception]) -> Self

Replace the exception types the polling loop retries instead of propagating.

Examples:

Usage:

await assert_that(get_order).eventually().within(10).ignoring(ConnectionError).has_status("PAID")

Raises:

Type	Description
TypeError	if any argument is not an Exception subclass

Source code in assertpy2/async_assertions.py

def ignoring(self, *exceptions: type[Exception]) -> Self:
    """Replace the exception types the polling loop retries instead of propagating.

    Examples:
        Usage:

            await assert_that(get_order).eventually().within(10).ignoring(ConnectionError).has_status("PAID")

    Raises:
        TypeError: if any argument is not an ``Exception`` subclass
    """
    self._ignoring = _normalize_ignoring(exceptions)
    return self

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

Do not instantiate directly; use eventually_sync() instead.

Parameters:

Name	Type	Description	Default
func	Callable	a sync callable that produces the value to test (an async probe raises TypeError)	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
ignoring	tuple[type[Exception], ...]	exception types the polling loop retries instead of propagating	()
kind	str | None	the failure mode of the final timeout failure (None/"soft"/"warn"); polling itself always retries on hard failures	None
logger	object	the logger for "warn" mode	None
trace	bool	record a PollTrace of the polling timeline (default True); False skips the flight recorder entirely	True

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,
    ignoring: tuple[type[Exception], ...] = (),
    kind: str | None = None,
    logger: object = None,
    trace: bool = True,
    steps: tuple[_Step, ...] = (),
    last: Any = None,
):
    self._func = func
    self._builder_func = builder_func
    self._description = description
    self._timeout = timeout
    self._interval = interval
    self._ignoring = ignoring
    self._kind = kind
    self._logger = logger
    self._trace = trace
    # every call made on this chain, replayed against a fresh builder on each poll.  Handing back
    # the builder of the last poll instead used to end the polling silently: `.is_instance_of(int)`
    # passed once and returned an ordinary builder, so `.is_equal_to(4)` after it ran against that
    # single snapshot and failed without ever waiting
    self._steps = steps
    self._last = last

val property

val: object

The value the last passing poll saw.

Declared on the class rather than left to __getattr__, which answers every other name with a polling call: reading .val off a chain would otherwise poll once and hand back a function.

Before anything has passed there is no such value, and __getattr__ says so. A raise here would not: Python falls back to __getattr__ whenever an attribute lookup ends in AttributeError, property included, so the message would have been swallowed and answered with a polling call all the same.

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

ignoring

ignoring(*exceptions: type[Exception]) -> Self

Replace the exception types the polling loop retries instead of propagating.

Examples:

Usage:

assert_that(get_order).eventually_sync().within(10).ignoring(ConnectionError).has_status("PAID")

Raises:

Type	Description
TypeError	if any argument is not an Exception subclass

Source code in assertpy2/async_assertions.py

def ignoring(self, *exceptions: type[Exception]) -> Self:
    """Replace the exception types the polling loop retries instead of propagating.

    Examples:
        Usage:

            assert_that(get_order).eventually_sync().within(10).ignoring(ConnectionError).has_status("PAID")

    Raises:
        TypeError: if any argument is not an ``Exception`` subclass
    """
    self._ignoring = _normalize_ignoring(exceptions)
    return self