The default way to test concurrent code looks like this:
def test_counter_is_thread_safe() -> None:
counter = Counter()
threads = [threading.Thread(target=counter.increment) for _ in range(100)]
for t in threads:
t.start()
for t in threads:
t.join()
assert counter.value == 100
Run it against a counter with no lock at all and it passes. Almost always. Thread 1 finishes its increment long before thread 2 is scheduled, because starting a thread is orders of magnitude more expensive than incrementing an integer, so the threads never overlap and the race never happens.
Now put it in CI on a loaded runner and it fails once a fortnight. Somebody adds a retry. The bug ships.
A concurrency test is only worth having if it fails reliably when the code is wrong, and passes reliably when it is right. Both halves are hard, and both are achievable with a small set of techniques that have nothing to do with running the test a thousand times.
Force the interleaving with a Barrier
The problem above is that the threads never coexist. threading.Barrier fixes that directly: it blocks each arriving thread until the n-th arrives, then releases all of them at once.
def force_race(n: int, fn: Callable[[], None]) -> None:
barrier = threading.Barrier(n)
def worker() -> None:
barrier.wait() # everybody waits here
fn() # everybody proceeds from here, together
threads = [threading.Thread(target=worker) for _ in range(n)]
for t in threads:
t.start()
for t in threads:
t.join()
The expensive part — thread creation — now happens before the barrier. The cheap, racy part happens after it, with every thread already scheduled and runnable. An unsynchronised self._n = self._n + 1 under this harness fails on the first run rather than the four-hundredth.
A Barrier is also the tool for testing that something is correctly synchronised, which is the direction people forget. Run n threads through a properly locked counter behind a barrier and assert the total is exactly n: now the test is deterministic in both directions, and it stays deterministic on a machine with 128 cores.
💡A colleague suggests testing the same counter by looping the whole test 1,000 times, on the grounds that the race will show up eventually. Why is that not equivalent? click to reveal
Because it converts a deterministic failure into a probabilistic one, and probabilistic failures are the thing you are trying to eliminate.
Three concrete problems. It is still not reliable — the probability per iteration is unknown, hardware-dependent, and can be effectively zero on a single-core container, so “1,000 iterations” is a number with no relationship to the confidence it provides. It is slow, and slow tests get marked as nightly, and nightly tests get muted. And when it does fail, it fails on iteration 743 with no additional information, so you are back to reading the code and guessing.
The barrier version fails on iteration 1, every time, on every machine. That difference is not a matter of degree. One of these is a test; the other is sampling.
The same reasoning applies to time.sleep(0.1) used to “let the other thread get there”. It is the same bet with worse odds, and it also makes the suite slower in exact proportion to how careful you were.
Replace real time with a controllable clock
Anything that measures elapsed time — a rate limiter, a retry backoff, a cache TTL, a circuit breaker — cannot be tested with real time. A test that waits five seconds to prove a token bucket refills is a test that takes five seconds and still cannot cover the sixty-second window.
Take the clock as a parameter:
class Clock(Protocol):
def monotonic(self) -> float: ...
class FakeClock:
def __init__(self) -> None:
self.now = 0.0
def monotonic(self) -> float:
return self.now
Now the test sets clock.now = 5.0 and five seconds have passed, instantly and exactly. You can also do things real time will not let you: jump backwards, to check the code survives a clock adjustment; jump forward by a year; land exactly on a boundary.
For asyncio, the equivalent is a loop whose time source you control, or — more commonly — restructuring so that the thing being awaited is an asyncio.Event you set from the test rather than a sleep you wait out.
Pin the interleaving with an Event
When the assertion is about ordering rather than about a total, asyncio.Event (or threading.Event) is the tool. Instead of hoping the consumer runs before the producer finishes, make it so:
async def test_consumer_sees_partial_state() -> None:
released = asyncio.Event()
observed: list[str] = []
async def producer() -> None:
observed.append("wrote-half")
released.set() # hand over at a known point
await asyncio.sleep(0)
observed.append("wrote-rest")
async def consumer() -> None:
await released.wait() # cannot proceed before that point
observed.append("read")
async with asyncio.TaskGroup() as tg:
tg.create_task(consumer())
tg.create_task(producer())
assert observed == ["wrote-half", "read", "wrote-rest"]The assertion is on a recorded event log, not on a duration. That is the general move for every concurrency test in this course: make the code produce a deterministic artefact — an ordering, a count, a merged structure — and assert on that. Durations are not artefacts; they are measurements of the machine you happened to run on.
Detect leaked tasks
The most common asyncio defect in production code is a task nobody awaits:
asyncio.create_task(send_metrics(payload)) # fire and forgetNobody holds the result, so nobody sees its exceptions; the loop may cancel it mid-write at shutdown; and it can be garbage-collected before it finishes, since the loop holds only a weak reference. It never fails a test, because the test has already returned.
You can catch it mechanically:
async def run_and_report(body: Callable[[], Awaitable[T]]) -> tuple[T, list[str]]:
before = asyncio.all_tasks()
value = await body()
leaked = [
t for t in asyncio.all_tasks()
if t not in before and t is not asyncio.current_task()
]
names = sorted(t.get_name() for t in leaked)
for t in leaked:
t.cancel()
await asyncio.gather(*leaked, return_exceptions=True)
return value, names
Two details do the work. asyncio.all_tasks() returns only pending tasks, so anything the body awaited to completion has already dropped out and only genuine leaks remain. And the check happens immediately after the body returns, inside the loop — asyncio.run cancels stragglers on its way out, so a check after the loop closes finds nothing, every time.
Wrap it in an autouse fixture and the whole suite gains leak detection for free. asyncio.Runner (3.11+) is the right driver: same lifecycle as asyncio.run, but as a context manager, so a test can run several coroutines on one loop and decide when it closes.
💡Which of these belong in pyproject.toml on the day you write your first async test, and what does each one actually catch?
click to reveal
Two, and they catch different halves of the same class of bug.
filterwarnings = ["error::RuntimeWarning"] — or the broader ["error"]. The specific thing this catches is RuntimeWarning: coroutine 'foo' was never awaited, which is what Python emits when you write do_thing() instead of await do_thing(). As a warning it is invisible in a CI log; as an error it fails the test that contains it, immediately, with the coroutine’s name. A missing await is one of the two or three most common asyncio mistakes and this converts it from a silent no-op into a failure.
asyncio debug mode, via PYTHONASYNCIODEBUG=1 or asyncio.Runner(debug=True). This turns on slow-callback warnings — a coroutine blocking the loop for longer than a threshold, which is how a synchronous requests.get hidden three layers down announces itself — and “Task was destroyed but it is pending”, which is the leak above. It is noise in production and exactly the signal you want in CI.
What does not belong is a global timeout plugin used as a substitute for either. A test that hangs and gets killed at 60 seconds tells you nothing about why; the debug-mode warning tells you which callback blocked and for how long. Timeouts are a safety net for the pipeline, not a diagnostic.
The rule
Never assert on a duration. Never sleep to let something happen. Every concurrency test should be able to answer “what forces the interleaving I am testing?” with a specific object — a barrier, an event, a fake clock — and should assert on something you could write down on paper.
If the answer is “the scheduler, probably”, you have written a flaky test and you have not yet found out.