Tests blocked on threads, queues and the event loop

Tests that wait in Thread.join, Event.wait, Queue.get or select() burn time doing nothing. Lower timeouts in tests, signal completion, and run background work synchronously.

A test that starts background work has to wait for it. When the wait ends because the work signalled that it's done, that's fine. When it ends because a timeout ran out, the test spends the whole timeout doing nothing, every run.

This is the less obvious cousin of time.sleep in tests: there's no sleep() anywhere, but the test still sits blocked in Thread.join(), Event.wait(), Queue.get() or the event loop's select() call until a timer fires.

Where tests block

The usual culprits:

Worker threads that only notice shutdown on a timeout. A loop like while running: item = queue.get(timeout=1) takes up to a second to exit after running = False, and the fixture's thread.join() waits for it.
Asserting that something doesn't happen. assert not event.wait(timeout=2) or queue.get(timeout=5) inside pytest.raises(queue.Empty) always costs the full timeout.
Production timeouts in tests. A client with a 30 second read timeout, a flush interval of 10 seconds, a health check that retries for a minute: in tests they turn one failure, or one missing signal, into a long wait.
Thread and process pools. executor.shutdown(wait=True) and future.result() wait for work you may not need to run concurrently at all.
The event loop idling. In async tests, await asyncio.wait_for(queue.get(), timeout=5) waiting for a message that never arrives, or a server task that's only cancelled on a timeout, leaves the loop blocked in select().
Live servers and consumers that are started per test and stopped by waiting for a timeout.

Signal completion instead of waiting for a timeout

Give background workers an explicit way to stop. A sentinel on the queue wakes the worker immediately:

import queue
import threading

STOP = object()


def worker(jobs: queue.Queue) -> None:
    while (job := jobs.get()) is not STOP:
        job.run()


def test_worker_processes_jobs():
    jobs = queue.Queue()
    thread = threading.Thread(target=worker, args=(jobs,))
    thread.start()
    jobs.put(make_job())
    jobs.put(STOP)
    thread.join(timeout=5)  # returns as soon as the worker sees STOP
    assert not thread.is_alive()

The same goes for a stop flag: use a threading.Event and have the worker wait on stop.wait(interval) instead of time.sleep(interval), so setting the event wakes it at once. For async code, cancel the task (task.cancel(), then await it) instead of waiting for it to time out.

When you wait for a result, wait for that result, not for a fixed time. future.result(timeout=5), done.wait(timeout=5) and thread.join(timeout=5) all return as soon as the work finishes; a generous timeout only matters when the test is about to fail.

Lower the timeouts in test settings

Timeouts that make sense in production rarely make sense in tests. Make them configurable and set them low in test settings:

# settings/test.py
PAYMENT_CLIENT_TIMEOUT = 0.5
EVENT_FLUSH_INTERVAL = 0.01
HEALTH_CHECK_RETRIES = 1

For tests that check that something doesn't happen, the timeout is the whole cost of the test. Keep it short (a few tens of milliseconds is often enough when the code under test runs in the same process), or restructure the test to check the decision directly instead of waiting for a non-event.

Run background work synchronously

Most tests don't care that work runs in the background; they care about its result. Run it inline:

Celery: CELERY_TASK_ALWAYS_EAGER = True in test settings runs tasks in-process when they're queued. See run Celery tasks eagerly in tests.
Executors: inject the executor and pass a synchronous one in tests.

from concurrent.futures import Executor, Future


class SynchronousExecutor(Executor):
    def submit(self, fn, /, *args, **kwargs):
        future = Future()
        try:
            future.set_result(fn(*args, **kwargs))
        except Exception as exc:
            future.set_exception(exc)
        return future

def test_report_is_generated():
    service = ReportService(executor=SynchronousExecutor())
    future = service.generate(report_id=1)
    assert future.result().status == "done"

map() and shutdown() come from the Executor base class, so the code under test doesn't notice the difference.

Django on_commit hooks that start background jobs: in a TestCase, self.captureOnCommitCallbacks(execute=True) runs them when the block exits, without a real transaction commit.

Keep the tests correct

A short timeout makes a test fail faster when the code is broken, but too short and it fails on a slow CI machine. Use short timeouts where the test expects nothing to happen, and generous ones (that rarely expire) where it waits for a result.
Synchronous execution hides concurrency bugs (races, deadlocks, ordering). Keep a few tests that run the real threads, pool or worker.
A worker or server that is shut down by a sentinel or cancellation must still be shut down: leaked threads keep running into the next test and make results depend on test order.

How flakytest finds it

flakytest puts samples blocked in Event.wait, Thread.join, lock acquisition, Queue.get/put/join, select() (the selectors module, which is also where an idle asyncio event loop waits) and concurrent.futures in the waiting category. When waiting takes more than a second and 3% of test time, it suggests Tests are blocked waiting on threads, queues or the event loop, with the tests and the lines of your code that wait the longest. Not all of that time is avoidable, so flakytest estimates the saving as half of it. Run flakytest report --test <test id> to see the call tree of one test and what it's waiting on.

Related guides

Tests are sleeping: replace time.sleep: Fixed time.sleep and asyncio.sleep calls make pytest suites slow and flaky. Poll for the condition, patch the sleep in retry and backoff code, and move the clock with time-machine.
Run Celery tasks eagerly in tests: Tests that call task.delay() publish to a broker, retry against a missing one or wait on a worker. Run tasks eagerly with CELERY_TASK_ALWAYS_EAGER, or mock delay.
One event loop per async test is slow: pytest-asyncio gives every async test its own event loop by default. Share one loop with asyncio_default_test_loop_scope and loop_scope="session" so async fixtures can be shared too.

Updated October 18, 2026 · Suggest a change