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 other exception raised by val() itself propagates immediately. An AssertionError from val() reaches the same place a failing assertion does and is retried the same way, which the timeout message names. A probe that signals "not ready yet" by raising something else (a connection refused while a service boots, a record not yet visible) can be retried too by listing those 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). Infinity, or a number past the float range, retries until the assertion passes	5.0
interval	float	seconds between retries (default 0.5), finite	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, if timeout or interval is not a real number or is a bool, or if ignoring contains anything that is not an Exception subclass
ValueError	if timeout or interval is negative or NaN, or interval is infinite

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 other exception raised by ``val()`` itself
    propagates immediately.  An ``AssertionError`` from ``val()`` reaches the same place a failing
    assertion does and is retried the same way, which the timeout message names.  A probe that
    signals "not ready yet" by raising something else (a connection refused while a service boots,
    a record not yet visible) can be retried too by listing those 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``). Infinity, or a number past the
            float range, retries until the assertion passes
        interval: seconds between retries (default ``0.5``), finite
        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, if ``timeout`` or ``interval`` is not a real number or is
            a bool, or if ``ignoring`` contains anything that is not an ``Exception`` subclass
        ValueError: if ``timeout`` or ``interval`` is negative or NaN, or ``interval`` is infinite
    """
    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). Infinity, or a number past the float range, retries until the assertion passes	5.0
interval	float	seconds between retries (default 0.5), finite	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, if timeout or interval is not a real number or is a bool, or if ignoring contains anything that is not an Exception subclass
ValueError	if timeout or interval is negative or NaN, or interval is infinite

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``). Infinity, or a number past the
            float range, retries until the assertion passes
        interval: seconds between retries (default ``0.5``), finite
        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, if ``timeout`` or ``interval`` is not a real number or is
            a bool, or if ``ignoring`` contains anything that is not an ``Exception`` subclass
        ValueError: if ``timeout`` or ``interval`` is negative or NaN, or ``interval`` is infinite
    """
    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,
    )

AsyncAssertionBuilder

AsyncAssertionBuilder(
    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, ...] = (),
)

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

Do not instantiate directly; use eventually() instead.

A chain records the calls made on it and runs them together when awaited, so the wait covers every link rather than only the first. The four coroutine methods below make it a coroutine to asyncio.run() and Task, which take nothing else below 3.15.

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,
    steps: tuple[_Step, ...] = (),
):
    self._func = func
    self._builder_func = builder_func
    self._description = description
    self._timeout = _duration(timeout, "timeout", endless=True)
    self._interval = _duration(interval, "interval", endless=False)
    self._ignoring = ignoring
    self._kind = kind
    self._logger = logger
    self._trace = trace
    self._steps = steps  # replayed as a whole on every poll, see `_replay`
    self._coro: Any = None
    self._awaited = False
    self._superseded = False

within

within(timeout: float) -> Self

Override the timeout (in seconds).

Infinity, or a number past the float range, retries until the assertion passes.

Raises:

Type	Description
TypeError	if timeout is not a real number, or is a bool
ValueError	if timeout is negative or NaN

Source code in assertpy2/async_assertions.py

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

    Infinity, or a number past the float range, retries until the assertion passes.

    Raises:
        TypeError: if ``timeout`` is not a real number, or is a bool
        ValueError: if ``timeout`` is negative or NaN
    """
    self._timeout = _duration(timeout, "timeout", endless=True)
    return self

every

every(interval: float) -> Self

Override the polling interval (in seconds).

Raises:

Type	Description
TypeError	if interval is not a real number, or is a bool
ValueError	if interval is negative, NaN or infinite

Source code in assertpy2/async_assertions.py

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

    Raises:
        TypeError: if ``interval`` is not a real number, or is a bool
        ValueError: if ``interval`` is negative, NaN or infinite
    """
    self._interval = _duration(interval, "interval", endless=False)
    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

SyncAssertionBuilder

SyncAssertionBuilder(
    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,
)

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 = _duration(timeout, "timeout", endless=True)
    self._interval = _duration(interval, "interval", endless=False)
    self._ignoring = ignoring
    self._kind = kind
    self._logger = logger
    self._trace = trace
    self._steps = steps  # replayed as a whole on every poll, see `_replay`
    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).

Infinity, or a number past the float range, retries until the assertion passes.

Raises:

Type	Description
TypeError	if timeout is not a real number, or is a bool
ValueError	if timeout is negative or NaN

Source code in assertpy2/async_assertions.py

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

    Infinity, or a number past the float range, retries until the assertion passes.

    Raises:
        TypeError: if ``timeout`` is not a real number, or is a bool
        ValueError: if ``timeout`` is negative or NaN
    """
    self._timeout = _duration(timeout, "timeout", endless=True)
    return self

every

every(interval: float) -> Self

Override the polling interval (in seconds).

Raises:

Type	Description
TypeError	if interval is not a real number, or is a bool
ValueError	if interval is negative, NaN or infinite

Source code in assertpy2/async_assertions.py

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

    Raises:
        TypeError: if ``interval`` is not a real number, or is a bool
        ValueError: if ``interval`` is negative, NaN or infinite
    """
    self._interval = _duration(interval, "interval", endless=False)
    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