Make Async Tests Repeatable with Handshakes and Fake Clocks

The previous article replaced sleeps with events. This one adds four tools. A handshake holds a task at a checkpoint. A fake clock lets you test code that reads the time. A task snapshot finds the tasks a test left running. A barrier starts several threads at the same moment.

Hold a task at a checkpoint with two events

You want to test what a reader sees while a writer is halfway through. One event tells the reader that the writer has reached the halfway point. A second event holds the writer there until the reader has looked. Two events used this way are a handshake:

import asyncio


def test_consumer_reads_between_writes() -> None:
    async def scenario() -> list[str]:
        half_written = asyncio.Event()
        read_complete = asyncio.Event()
        observed: list[str] = []

        async def producer() -> None:
            observed.append("wrote-half")
            half_written.set()
            await read_complete.wait()
            observed.append("wrote-rest")

        async def consumer() -> None:
            await half_written.wait()
            observed.append("read")
            read_complete.set()

        async with asyncio.timeout(5):
            async with asyncio.TaskGroup() as group:
                group.create_task(producer())
                group.create_task(consumer())
        return observed

    assert asyncio.run(scenario()) == ["wrote-half", "read", "wrote-rest"]

The first event puts the read after the first write. The second keeps the last write until after the read. With only the first event, the writer could finish before the reader looks. As in the previous article, the asyncio.timeout(5) only stops a broken handshake from hanging the suite.

This test checks the handshake itself. In your own tests, put the checkpoints inside the code under test, for example through a hook that the test passes in. Then assert on that code's state.

Test time-based code with a fake clock

A cache entry expires after a minute, and your test should finish in milliseconds. Pass the clock in as an argument, and give the test a clock it can set:

from collections.abc import Callable
from dataclasses import dataclass


@dataclass
class FakeClock:
    now: float = 0.0

    def __call__(self) -> float:
        return self.now


class Deadline:
    def __init__(self, seconds: float, *, clock: Callable[[], float]) -> None:
        self.clock = clock
        self.expires_at = clock() + seconds

    def expired(self) -> bool:
        return self.clock() >= self.expires_at


def test_expiry_boundary() -> None:
    clock = FakeClock()
    deadline = Deadline(10.0, clock=clock)
    clock.now = 9.0
    assert not deadline.expired()
    clock.now = 10.0
    assert deadline.expired()

In production, pass time.monotonic, a clock that never goes backward. Move the fake clock only forward too, so the test matches what production sees.

The fake clock changes only what clock() returns. It doesn't move asyncio's timers, so a task in asyncio.sleep(60) still sleeps a real minute. Use the fake clock for code that reads the time, and an event for code that waits.

Find the tasks a test left running

A handler that starts a task and forgets it can pass every test while the task runs on in the background. To catch it, compare asyncio.all_tasks() before and after the call:

import asyncio


async def handler() -> None:
    asyncio.create_task(asyncio.sleep(60), name="refresh")  # bug: nobody awaits it


async def main() -> None:
    before = asyncio.all_tasks()
    await handler()
    leftover = asyncio.all_tasks() - before
    print(sorted(task.get_name() for task in leftover))  # ['refresh']


asyncio.run(main())

all_tasks() returns the event loop's unfinished tasks, including tasks that have not started yet. Three things affect what the difference shows:

• A task that has already finished is missing, even if nobody read its exception.
• A task that another fixture started on purpose shows up too, so decide which tasks your check is responsible for.
• Take the second snapshot while the loop is still running. asyncio.run cancels the leftover tasks as it returns, so afterwards there is nothing to see.

Once you have the names, cancel the leftover tasks and await them, so the next test starts clean. The exercise builds this check.

Line threads up with a barrier

A stress test often runs one function on several threads at once. threading.Barrier(n) makes each thread wait until all n have arrived, then lets them all go. That makes overlap more likely. After that, the scheduler still decides which thread runs first, as the previous article said, so a passing run proves nothing. Keep the lock in the code either way.

Going further

Debug mode, from the article on debugging async services, adds the line that created a coroutine to the "never awaited" warning. Python prints that warning when it deletes the coroutine, which can be after the test that caused it has finished. With debug mode on, look at the creation line the warning shows, not at the test that was running when it appeared.