# 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](https://flakytest.dev/guides/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:

```python
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:

```python
# 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](https://flakytest.dev/guides/celery-eager-in-tests/).
- **Executors**: inject the executor and pass a synchronous one in tests.

```python
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
```

```python
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.

## Further reading

- [Python: threading.Event](https://docs.python.org/3/library/threading.html#event-objects)
- [Python: concurrent.futures](https://docs.python.org/3/library/concurrent.futures.html)
- [Python: asyncio.wait_for](https://docs.python.org/3/library/asyncio-task.html#asyncio.wait_for)
- [Celery: testing](https://docs.celeryq.dev/en/stable/userguide/testing.html)