Write Concurrency Tests with Events, Not Sleeps

You want to test that cancelling a worker gives its connection back to the pool. A first try looks like this:

task = asyncio.create_task(worker(held))
await asyncio.sleep(0.1)  # hope the worker has its connection by now
task.cancel()

The sleep is a guess. The worker first waits for the pool, and on a busy CI machine that can take longer than 0.1 seconds. The test then cancels a worker that is still waiting for the pool. It passes without ever running the release code.

Instead, have the worker set an event when it holds the connection, and wait for that event:

import asyncio


async def worker(held: list[str], acquired: asyncio.Event) -> None:
    await asyncio.sleep(0.01)  # stands in for waiting on the pool
    held.append("conn")
    try:
        acquired.set()
        await asyncio.Event().wait()  # keep the connection until cancelled
    finally:
        held.remove("conn")


async def scenario() -> None:
    held: list[str] = []
    acquired = asyncio.Event()
    task = asyncio.create_task(worker(held, acquired))
    async with asyncio.timeout(5):  # fail instead of hanging
        await acquired.wait()
        task.cancel()
        await asyncio.wait([task])
    assert task.cancelled()
    assert held == []


def test_cancel_releases_connection() -> None:
    asyncio.run(scenario())

Now the test cancels the worker only after it holds the connection, however slow the machine is.

Assert on results, not on elapsed time

Assert on what the code's users can see: the final count, the resources given back, the order of recorded events. Elapsed time depends on how busy the machine is, so don't assert on it. For deadline logic, use a fake clock, as the concurrency-testing article after this one shows. Keep speed checks in separate benchmarks.

The asyncio.timeout(5) in the example only guards against a hang. If the coordination breaks, the test fails after 5 seconds instead of hanging the whole suite.

Set up the order you want to test

Put an event at each step whose order matters. Use asyncio.Event for tasks and threading.Event for threads. Three common shortcuts leave the order to chance.

• A threading.Barrier lines threads up at the start. After it opens, the scheduler still decides which thread runs each line, so the race you want to test happens only by chance.
• Running a racy loop thousands of times may expose a bug, but a pass can be luck.
• await asyncio.sleep(0) gives other tasks one turn. How many turns a task needs to reach a given line can change, for example with eager tasks.

Record safely, and always clean up

When several threads record what they saw, guard the record with a lock, or use a queue.Queue. Stop every worker in a finally block, even when an assertion fails, so threads from a failed test stop before the next test starts.