Pytest Helpers

Machinery also offers some helpers for working with asyncio in pytest. It is also recommended to use alt-pytest-asyncio to write asyncio tests in pytest, but pytest-asyncio should also work fine.

Future Dominoes

protocol machinery._test_helpers._future_dominos.Domino

Used to represent the futures provided by FutureDominos.

A type alias for this exists at machinery.test_helpers.Domino

Classes that implement this protocol must have the following methods / attributes:

__await__() → Generator[None]: Wait for the domino to be knocked over.

add_done_callback(cb: Callable[[FutureStatus[None]], None]) → FutureCallback[None]: Add a callback for when the domino is knocked over

cancel() → None: Set the domino as cancelled (give it an exception of asyncio.CancelledError)

cancelled() → bool: Return true if the domino was cancelled

done() → bool: Return true if the domino has been knocked over

exception() → BaseException | None: Return true if the domino was given an exception

result() → None: Return the result from the domino

set_exception(exc: BaseException) → None: Set an exception on the domino

set_result(value: None) → None: Knock over a domino with no exception

protocol machinery._test_helpers._future_dominos.FutureDominos

An object that represents a “domino” set of futures that only complete as each previous domino is retrieved and awaited

A type alias for this exists at machinery.test_helpers.FutureDominos

Classes that implement this protocol must have the following methods / attributes:

__getitem__(num: int) → Domino: Get the domino indexed by num

begin() → None: Used by the test to indicate that the dominos should begin

property finished: Event: This is set for us when all the dominos have been knocked over

property started: Event: This is set for us when the dominos have started

machinery.test_helpers.future_dominos(*, expected: int, loop: AbstractEventLoop | None = None, log: Logger | None = None, name: str = '') → AsyncGenerator[FutureDominos]

A helper to start a domino of futures.

For example:

from collections.abc import AsyncGenerator

from machinery import test_helpers as thp

async def run() -> None:
    async with thp.future_dominos(loop=loop, expected=8) as futs:
        called: list[object] = []

        async def one() -> None:
            await futs[1]
            called.append("first")
            await futs[2]
            called.append("second")
            await futs[5]
            called.append("fifth")
            await futs[7]
            called.append("seventh")

        async def two() -> AsyncGenerator[tuple[str, int]]:
            await futs[3]
            called.append("third")

            start = 4
            while start <= 6:
                await futs[start]
                called.append(("gen", start))
                yield ("genresult", start)
                start += 2

        async def three() -> None:
            await futs[8]
            called.append("final")

        loop = ...
        loop.create_task(three())
        loop.create_task(one())

        async def run_two() -> None:
            async for r in two():
                called.append(r)

        loop.create_task(run_two())
        futs.begin()
        await futs.finished.wait()

        assert called == [
            "first",
            "second",
            "third",
            ("gen", 4),
            ("genresult", 4),
            "fifth",
            ("gen", 6),
            ("genresult", 6),
            "seventh",
            "final",
        ]

Mocked call later

protocol machinery._test_helpers._mocked_call_later.Cancellable

An object that can be cancelled.

A type alias for this exists at machinery.test_helpers.Cancellable

Classes that implement this protocol must have the following methods / attributes:

cancel() → None

protocol machinery._test_helpers._mocked_call_later.MockedCallLater

The interface returned by thp.mocked_call_later

A type alias for this exists at machinery.test_helpers.MockedCallLater

Classes that implement this protocol must have the following methods / attributes:

async add(amount: float) → None: Process the loop this amount of time, taking care to call any callbacks that would be fired if that amount of real time passed, in the order that they would.

property called_times: list[float]: A list of times that represent the time callbacks were called.

machinery.test_helpers.mocked_call_later(*, ctx: CTX | None = None, precision: float = 0.1, start_time: float = 0, name: str = '') → AsyncGenerator[MockedCallLater]

This gets us the ability to wait large periods of time whilst not passing very much clock time.

This works by mocking loop.call_later so that time in the loop isn’t in line with time in real life.

Usage is:

from machinery import test_helpers as thp
from machinery import helpers as hp
import time

ctx: hp.CTX = ...


async with thp.mocked_call_later(ctx=ctx) as m:
    assert time.time() == 0

    event = asyncio.Event()
    ctx.loop.call_later(3, event.set())
    await event.wait()

    assert time.time() == 3 # but in reality effectively no time has passed