The most common concurrency test is also one of the weakest: start background work, sleep for an amount that feels safe, then assert. On a fast machine the sleep wastes time. On a loaded CI runner it is too short. When it fails, the test says only that the scheduler did not honor a guess.

Deterministic concurrency tests do not control the scheduler. They control protocol states. A worker signals "I reached the read." The test releases "you may write." A coroutine exposes "resource acquired." The test requests cancellation and awaits cleanup. Every wait has a diagnostic upper bound so a defect fails instead of hanging the suite.

This tutorial tests mechanisms with standard-library primitives. The patterns fit pytest directly; pytest-specific examples use ordinary synchronous test functions and asyncio.run, so no async plugin is required.

Version note. All blocks were run on GIL-enabled CPython 3.14.7. Thread scheduling, wake-up order, and timing are not Python guarantees. Synchronization primitive semantics are documented APIs. Task-group and cancellation behavior is Python 3.14 asyncio behavior. Free-threaded builds can expose additional interleavings and should be a separate CI dimension, not inferred from this run.

Experiment 1: replace sleep with a handshake

from threading import Event, Thread

started = Event()
release = Event()
finished = Event()

def worker():
    started.set()
    assert release.wait(timeout=1)
    finished.set()

thread = Thread(target=worker)
thread.start()
assert started.wait(timeout=1)
assert not finished.is_set()
release.set()
thread.join(timeout=1)

assert not thread.is_alive()
assert finished.is_set()

The test proves a state transition: before release, work is not finished; after release, it is. No duration is part of the product claim. One-second timeouts are circuit breakers for broken tests, not synchronization instructions.

Always assert the result of Event.wait() and verify join() actually ended the thread. A timed-out wait that is ignored lets the test continue in an unknown state. A non-daemon thread left alive can hang interpreter shutdown; a daemon thread can be abandoned mid-cleanup.

Python guarantee. Event.set() wakes waiters and future wait() calls return until clear(). Python does not guarantee which awakened thread runs first.

Experiment 2: a barrier creates a known race window

from threading import Barrier, Lock, Thread

gate = Barrier(3)
lock = Lock()
counter = 0

def increment():
    global counter
    gate.wait(timeout=1)
    with lock:
        observed = counter
        counter = observed + 1

threads = [Thread(target=increment) for _ in range(2)]
for thread in threads:
    thread.start()

gate.wait(timeout=1)
for thread in threads:
    thread.join(timeout=1)

assert all(not thread.is_alive() for thread in threads)
assert counter == 2

The barrier releases both workers and the test when all three arrive. The lock then protects the read-modify-write invariant. This test establishes concurrency at a meaningful boundary but does not claim both Python statements execute simultaneously.

Removing the lock and expecting a lost update is not a reliable regression test. A race bug means an invalid outcome is permitted, not that every scheduler must produce it. Instead, inject synchronization between read and write in a deliberately instrumented implementation or test the lock-bearing public abstraction.

Barrier can become broken if a participant times out or aborts. Cleanup should tolerate BrokenBarrierError, and fixture teardown must still join every thread.

Experiment 3: capture thread exceptions explicitly

from concurrent.futures import ThreadPoolExecutor

def fail():
    raise ValueError('worker failed')

with ThreadPoolExecutor(max_workers=1) as pool:
    future = pool.submit(fail)
    try:
        future.result(timeout=1)
    except ValueError as error:
        assert str(error) == 'worker failed'
    else:
        raise AssertionError('exception was not propagated')

An exception in a bare thread does not automatically fail the test function. Pytest may warn about unhandled thread exceptions, depending on its plugin behavior, but explicit propagation is clearer. Futures retain either a result or exception and make bounded observation straightforward.

If the unit must create a Thread directly, provide a wrapper that captures BaseException into a queue and re-raises it in the test owner. Do not rely only on log text. Also restore every executor with a context manager so worker lifetime belongs to the test.

Calling future.cancel() cannot stop a function already running. Test cooperative thread cancellation with a stop event owned by the worker protocol.

Experiment 4: async events locate the cancellation point

import asyncio

async def worker(acquired, released):
    acquired.set()
    try:
        await asyncio.Event().wait()
    finally:
        released.set()

async def scenario():
    acquired = asyncio.Event()
    released = asyncio.Event()
    task = asyncio.create_task(worker(acquired, released))

    await asyncio.wait_for(acquired.wait(), timeout=1)
    task.cancel()
    try:
        await asyncio.wait_for(task, timeout=1)
    except asyncio.CancelledError:
        pass

    assert task.cancelled()
    assert released.is_set()

asyncio.run(scenario())

The test cancels only after acquisition. This is a contract assertion: cancellation after that point must release the resource and propagate. await asyncio.sleep(0) would only suggest that the worker probably started; the explicit event proves it reached the state relevant to the test.

Use an event for each interruption boundary that changes expected cleanup: before acquisition, after acquisition, after external submission, and during commit. You need not expose test-only events in every production signature. Inject a collaborator whose awaited methods are controlled fakes, or place hooks behind an internal adapter.

Bound the event wait and task completion separately. The resulting traceback then identifies whether startup or cancellation cleanup stalled.

Experiment 5: a controlled awaitable drives ordering

import asyncio

class ControlledGateway:
    def __init__(self):
        self.called = asyncio.Event()
        self.release = asyncio.Event()

    async def send(self, payload):
        self.called.set()
        await self.release.wait()
        return {'accepted': payload}

async def submit(gateway, payload):
    return await gateway.send(payload)

async def scenario():
    gateway = ControlledGateway()
    task = asyncio.create_task(submit(gateway, 'job-7'))
    await asyncio.wait_for(gateway.called.wait(), timeout=1)
    assert not task.done()
    gateway.release.set()
    result = await asyncio.wait_for(task, timeout=1)
    assert result == {'accepted': 'job-7'}

asyncio.run(scenario())

The fake models the async protocol rather than returning instantly. An AsyncMock with an immediate return value can miss ordering, cancellation, and backpressure bugs because it never suspends. A small controlled fake gives the test named lifecycle points.

Keep such fakes faithful and narrow. If the real gateway can reject, cancel, or produce partial effects, add only the states required by the contract under test. A giant fake server can become a second implementation with its own bugs.

Experiment 6: test timeout translation without wall-clock precision

import asyncio

async def never():
    await asyncio.Event().wait()

async def operation():
    async with asyncio.timeout(0.01):
        await never()

async def scenario():
    try:
        await operation()
    except TimeoutError:
        pass
    else:
        raise AssertionError('operation must time out')

asyncio.run(scenario())

This uses real loop time but asserts no narrow elapsed range. The test establishes error translation for a never-completing dependency. Ten milliseconds is still a delay and can accumulate across a large suite; where timeout duration is configurable, inject a very small test value.

Do not mock the event loop's time() casually. Asyncio schedules heaps of timers against one monotonic clock; jumping it incorrectly can violate loop assumptions. For extensive virtual-time testing, use a library explicitly designed for that event-loop integration and pin its version. Stdlib-first tests should prefer controlled completion plus a small number of integration tests for actual deadlines.

Experiment 7: simultaneous failures need a gate

import asyncio

async def failing(error, ready, release):
    ready.set()
    await release.wait()
    raise error

async def scenario():
    ready = [asyncio.Event(), asyncio.Event()]
    release = asyncio.Event()
    try:
        async with asyncio.TaskGroup() as group:
            group.create_task(failing(ValueError('left'), ready[0], release))
            group.create_task(failing(TypeError('right'), ready[1], release))
            await asyncio.wait_for(
                asyncio.gather(*(event.wait() for event in ready)),
                timeout=1,
            )
            release.set()
    except* Exception as errors:
        assert {type(error) for error in errors.exceptions} == {ValueError, TypeError}

asyncio.run(scenario())

Without the release gate, the first failing child may trigger sibling cancellation before the sibling raises. That would correctly test fail-fast behavior but not multiple-failure grouping. The gate creates the precondition for the desired case.

Concurrency tests must state which interleaving they establish. Never assert an order merely observed on one CPython build. Use sets when order is irrelevant and sequences only when the protocol enforces order.

For nested exception groups, assert meaningful tree structure and leaf types, not complete formatted traceback text. Rendering can change between Python releases and test runners.

Experiment 8: queues expose backpressure deterministically

import asyncio

async def scenario():
    queue = asyncio.Queue(maxsize=1)
    await queue.put('first')
    blocked_put = asyncio.create_task(queue.put('second'))
    await asyncio.sleep(0)
    assert not blocked_put.done()

    assert await queue.get() == 'first'
    await asyncio.wait_for(blocked_put, timeout=1)
    assert await queue.get() == 'second'

asyncio.run(scenario())

Here sleep(0) is not an elapsed guess; it yields one event-loop turn so the newly created task can attempt put. The decisive assertion comes from queue capacity. For even stronger orchestration, wrap the queue or producer with an explicit "attempted put" event.

Test overload as a first-class behavior: whether producers block, reject, drop, or coalesce. Also test cancellation while a producer is waiting. A bounded queue that works only on the happy path can leak unfinished-task counts or lose shutdown sentinels.

Experiment 9: polling needs a deadline and diagnostics

from time import monotonic, sleep

state = {'ready': False}
attempts = 0
deadline = monotonic() + 0.2

while monotonic() < deadline:
    attempts += 1
    if attempts == 3:
        state['ready'] = True
    if state['ready']:
        break
    sleep(0.001)
else:
    raise AssertionError(f'state never became ready: {state!r}')

assert state['ready']

Sometimes the boundary offers no notification: an external service updates eventually, a subprocess writes a file, or a legacy API exposes only status. Polling is then honest. Use a monotonic deadline, a modest interval, and a failure message containing last observed state.

Polling differs from sleeping once. It finishes as soon as the condition holds and keeps the timeout as a failure bound rather than an assumed completion time. It remains an integration technique, not a substitute for adding notifications to code you control.

A pytest pattern for owned cleanup

In pytest, put each resource lifetime in a fixture, but keep synchronization in the test. A thread fixture should yield its control object and join in finally; if join times out, fail with thread state. An async test managed by a plugin should cancel and await tasks it creates. If a fixture creates background tasks, that fixture owns their teardown.

Do not make every timeout enormous to fix CI flakes. Large bounds turn deadlocks into slow failures. First replace guessed sleeps with handshakes; then choose a timeout generous enough for scheduler noise but short enough to preserve suite feedback. Mark genuinely slow external integration separately.

Useful stress runs supplement deterministic cases:

  • Repeat a focused test to broaden incidental schedules.
  • Run under CPU and I/O load to reveal hidden assumptions.
  • Test a free-threaded CPython build when claiming thread correctness there.
  • Randomize test order to expose leaked global state.
  • Use platform CI for event-loop and process-start differences.

Stress can discover a race but should not be the only regression test. Once found, identify the missing synchronization and build a controlled interleaving that reproduces the violated invariant.

Keep diagnostics when the bound expires

A timeout error without state is only slightly better than a hang. Include task names, thread names, queue sizes, barrier participants, and the last protocol transition in failure output. For asyncio, inspect owned tasks with task.get_stack() during diagnostics; for threads, faulthandler.dump_traceback() can show every thread's current frame. These are debugging aids whose formatting is CPython-specific, not stable assertions.

Centralize bounded-wait helpers so they always assert completion and attach context. Keep the helper thin: hiding all events behind a generic retry utility can make the protocol less visible. The test should still read as a sequence of states and permissions.

Finally, ensure the failure path performs cleanup. Put release signals and joins in finally, cancel and gather owned tasks after an assertion fails, and shut executors down. Otherwise the first useful assertion can create secondary warnings, leaked workers, or a hung suite that obscures the original defect.

Exercises

  1. Rewrite a test containing sleep(1) with two events: reached state and permission to continue.
  2. Remove the lock from Experiment 2, instrument reads and writes with barriers, and deterministically expose a lost update.
  3. Cancel Experiment 5 while the gateway is blocked. Assert caller result, fake state, and task cleanup.
  4. Add a queue shutdown protocol to Experiment 8 and test cancellation of a blocked producer.
  5. Wrap a bare thread so exceptions and tracebacks are re-raised in the owning test and every failure path joins the thread.

Keep this model

Time is a poor proxy for state. Synchronization primitives let tests establish the exact preconditions that matter: started, acquired, blocked, released, failed, or cleaned. Timeouts remain necessary as bounds, but they should fail stalled tests rather than drive ordinary progress.

Own every task, thread, executor, and queue. Capture background exceptions. Await cancellation cleanup. Assert only order guaranteed by the protocol. Deterministic tests do not eliminate concurrency; they turn scheduler luck into explicit, reviewable coordination.

Primary sources