Python Testing & Debugging Beyond unittestAdvanced Python testing playbooks
Async & Concurrency

Sharing an Event Loop Across a Test Module

Creating a Postgres pool, a Redis client and an HTTP server for every test in a module costs a second each and, at function loop scope, is unavoidable — the loop dies with the test, so anything holding sockets on it must die too. Widening the loop to the module changes the arithmetic: the pool is built once, each test takes a connection from it, and the module's runtime drops from a minute to a few seconds.

Prerequisites

  • pytest-asyncio >= 0.24, which carries loop_scope on both pytest.mark.asyncio and pytest_asyncio.fixture.
  • pytest >= 8.0, Python 3.10+.
  • The scope rules from pytest-asyncio in depth, particularly that a test may use fixtures on its own loop or a wider one, never a narrower one.

Solution

Declare the loop scope once at module level, and give the expensive fixture the same scope on both axes.

Python
import asyncio

import asyncpg
import pytest
import pytest_asyncio

# Every test in this module runs on one loop that lives for the module.
pytestmark = pytest.mark.asyncio(loop_scope="module")


@pytest_asyncio.fixture(scope="module", loop_scope="module")
async def pool():
    # Built once. Because loop_scope matches scope, the pool's sockets are
    # registered with a loop that is still alive when `close()` runs.
    pool = await asyncpg.create_pool(
        dsn="postgresql://test@localhost/test", min_size=2, max_size=10
    )
    try:
        yield pool
    finally:
        await pool.close()


@pytest_asyncio.fixture(loop_scope="module")
async def conn(pool):
    # Function-scoped by default: a fresh connection and transaction per test,
    # on the module's loop, so nothing crosses a loop boundary.
    async with pool.acquire() as connection:
        transaction = connection.transaction()
        await transaction.start()
        try:
            yield connection
        finally:
            await transaction.rollback()


async def test_insert_is_visible_within_the_transaction(conn):
    await conn.execute("INSERT INTO widget (name) VALUES ($1)", "a")
    assert await conn.fetchval("SELECT count(*) FROM widget") == 1


async def test_previous_insert_was_rolled_back(conn):
    # Isolation comes from the transaction, not from a new loop.
    assert await conn.fetchval("SELECT count(*) FROM widget") == 0
One module loop carrying a pool across three tests A module-scoped loop is created once and a connection pool is built on it. Three tests then run in sequence, each acquiring a connection and opening a transaction that is rolled back at the end of the test. The loop and the pool are closed once after the last test. Built once, isolated per test module loop — created once, closed after the last test pool built one handshake test 1 acquire · begin · rollback test 2 acquire · begin · rollback test 3 same pool Function loops would rebuild the pool three times — three handshakes, three teardowns, and the same isolation the transaction was already providing.
Isolation and lifetime are separate concerns. Widening the loop does not weaken isolation as long as the per-test transaction is doing that job.

Why this works

loop_scope tells pytest-asyncio which of its loop instances to run an item on. Loops are keyed by scope and by the node that owns them, so a module scope produces one loop per test module and hands it to every test and fixture declaring that scope. The pool is created inside that loop, registers its sockets with that loop's selector, and is closed while the loop is still running — which is exactly the invariant a function-scoped loop cannot provide for a module-scoped object.

The per-test connection fixture is function-scoped but declares the same loop_scope, so it executes on the module loop while still running once per test. That combination — narrow scope, wide loop_scope — is the one worth remembering, because it is what gives per-test state on shared infrastructure.

Edge cases and failure modes

  • A test on a narrower loop requesting a wider fixture. Legal. The reverse is not: a module-loop fixture cannot be used by a session-loop test, and pytest-asyncio reports it as a fixture requested from a different loop.
  • pytestmark inside a class. Assigning pytestmark in a class body applies to that class only; tests outside it in the same file silently keep function loops. Put it at module level unless the split is deliberate.
  • Leaked tasks carried forward. A test that starts a task and does not await it leaves that task pending on a loop the next test will use. With function loops the leak dies with the loop; with a shared one it does not.
  • Module-scoped fixtures that are not async. A synchronous fixture has no loop affinity at all and needs no loop_scope; adding one is harmless but signals a misunderstanding worth correcting.
  • Ordering dependence. Tests sharing a loop and a pool can accidentally depend on each other through anything not rolled back — a Redis key, an in-memory cache, a background consumer. Run the module with -p randomly occasionally to catch it.

Guarding a shared loop against leaks

The one real cost of a shared loop is that a leak survives the test that caused it. An autouse fixture makes the leak fail its own test instead of a later one.

Python
import asyncio

import pytest


@pytest.fixture(autouse=True)
async def no_pending_tasks():
    yield
    pending = [t for t in asyncio.all_tasks() if t is not asyncio.current_task()]
    if pending:
        for task in pending:
            task.cancel()                      # do not poison the next test
        await asyncio.gather(*pending, return_exceptions=True)
        raise AssertionError(f"test left {len(pending)} pending task(s): {pending}")

Cancelling before raising matters: without it the assertion fails and the stray tasks continue into the next test, producing a second, confusing failure. Cancelling first means exactly one test goes red and the module continues cleanly.

The same idea applies to other shared state on the loop. A module that starts a background consumer should assert the queue is empty at the end of each test; one that uses a lock should assert it is not held. Each check is one line and turns a future ordering mystery into an immediate, attributable failure.

How a leaked task behaves under function versus shared loops Under function-scoped loops, a task leaked by the first test dies when that test's loop is closed and the next test is unaffected. Under a shared module loop the leaked task survives into the following tests, where it can consume the pool or mutate state, producing a failure attributed to the wrong test. A shared loop carries leaks forward function loops test 1 leaks a task → its loop closes → the task is gone later tests unaffected; the leak is invisible and also untested shared module loop test 1 leaks a task → the loop lives on → the task keeps running test 3 fails because test 1's task drained the pool — wrong test blamed
The autouse guard converts the lower row into the upper row's convenience without losing the information: the leak is reported, against the test that caused it.

Choosing between module and session scope

Module scope is the safer default and session scope the faster one, and the choice follows from what the resource costs and how far the blast radius should reach.

A session loop amortises setup across the entire run, which is right for something genuinely global: a database pool, a containerised service, a message broker connection. Its risk is that every test in the suite shares one failure domain, so a leak anywhere can affect anything.

A module loop confines both the benefit and the risk to one file. For a module of thirty tests against one service that is usually the better trade — the setup is paid once per file rather than once per run, which is nearly the same saving, and a leak can only affect the twenty-nine tests the author is already looking at.

Python
# conftest.py — session scope for the truly global resource
import pytest_asyncio


@pytest_asyncio.fixture(scope="session", loop_scope="session")
async def broker():
    client = await connect_broker()
    try:
        yield client
    finally:
        await client.close()
Python
# tests/test_orders.py — module loop for everything else
import pytest

pytestmark = pytest.mark.asyncio(loop_scope="module")

Mixing the two needs one rule respected: a test on the module loop cannot use the session-loop broker fixture. Either the test moves to session scope, or the fixture provides something loop-independent — a DSN, a factory, a synchronous handle — that the module-loop test uses to build its own client. The second option is usually cleaner, and it is the same separation recommended in database fixtures and transactional tests: expensive discovery at session scope, cheap per-scope construction on top.

Confirming the arrangement

One temporary test proves the whole configuration, and it is worth running once whenever the scopes change.

Python
import asyncio

import pytest

pytestmark = pytest.mark.asyncio(loop_scope="module")

_seen: list[int] = []


async def test_records_the_loop_a():
    _seen.append(id(asyncio.get_running_loop()))


async def test_records_the_loop_b(pool):
    _seen.append(id(asyncio.get_running_loop()))
    # The pool must be usable, which it is only if it lives on this same loop.
    assert await pool.fetchval("SELECT 1") == 1
    assert len(set(_seen)) == 1, f"tests ran on different loops: {_seen}"
Reading the loop-identity check Two outcomes. Matching loop identifiers across the two tests confirm the module scope applied and the pool is reachable. Differing identifiers mean the marker did not apply, and the usual causes are a class-level pytestmark or a fixture left at the default loop scope. What the identifiers tell you identical ids one loop for the module pool query succeeds configuration is correct delete the temporary test different ids pytestmark inside a class fixture loop_scope left default or the marker was overridden per test check the module level first
Two identifiers and one query. Anything that would break the arrangement in a hundred-test module shows up here in under a second.

Delete the temporary test once it passes; keeping it permanently asserts an implementation detail rather than behaviour, and the autouse leak guard above is the check worth keeping for the long term.

If the identifiers differ, work outward from the narrowest declaration. A loop_scope argument on an individual test's marker overrides the module-level pytestmark for that test alone, which is easy to leave behind after debugging. Next check that the fixture chain agrees: a fixture requesting another fixture that was left at the default scope pulls the whole chain back onto a function loop, and the error surfaces at the first test that uses the pool rather than at the fixture that caused it.

Frequently Asked Questions

Does a shared loop mean tests share state? It means they share whatever was created on that loop. A connection pool, a running server and any lock or queue built inside it persist across the tests in that scope. Per-test isolation then has to come from something else — a transaction that rolls back, a fresh connection from the pool, an explicit reset in a fixture.

Can one module use a session loop while another uses function loops? Yes. loop_scope is declared per test or per module, so a module with pytestmark set to a session loop coexists with modules that say nothing and get function loops. The only rule is that a test may not request a fixture bound to a narrower loop than its own.

How do I prove two tests really shared a loop? Assert on id(asyncio.get_running_loop()) in both. Matching identifiers confirm the shared loop; differing ones mean the loop_scope did not apply, usually because the marker was placed on the class rather than the module or the fixture's loop_scope was left at its default.

← Back to pytest-asyncio in Depth