Peyton Green

Posted on Sep 3

Testing Async Python Without Losing Your Mind

#python #testing #asyncio #pytest

You've written async Python. Your code works. Then you try to test it, and suddenly nothing makes sense.

RuntimeError: Event loop is closed
ScopeMismatch: You tried to access the function scoped fixture ... with a session scoped request object
asyncio.exceptions.CancelledError

Async testing in Python has a reputation for being hostile. It isn't — but it requires understanding three things that sync testing doesn't: event loop scope, fixture lifecycle in async context, and the specific patterns that silently break under pytest's default assumptions.

This guide covers the exact problems that break async tests and the exact solutions. No hand-waving.

The Setup

Two plugins. Pick one based on what you're building:

pytest-asyncio — the standard choice for asyncio-native projects:

pip install pytest-asyncio

anyio (via pytest-anyio) — if you want asyncio + trio compatibility or you're building a library:

pip install anyio[trio] pytest-anyio

We'll use pytest-asyncio throughout. The patterns apply to anyio with minor changes.

pytest.ini configuration — do this first:

[pytest]
asyncio_mode = auto

asyncio_mode = auto means every async test function is automatically treated as a coroutine to await. Without this, you need @pytest.mark.asyncio on every async test function. You will forget. Add it to pytest.ini.

Part 1: The Event Loop Problem

The most common async testing failure is event loop scope mismatch. Understanding it saves hours.

What the event loop is

Every async operation runs inside an event loop. In production, you have one event loop per process. In tests, pytest-asyncio creates event loops — and the question is: one per test, one per module, or one for the whole session?

Default: one per test function. This is the source of most "Event loop is closed" errors.

Why default scope breaks database fixtures

# This breaks with default (function-scoped) event loop
@pytest.fixture(scope="session")
async def db_engine():
    engine = create_async_engine("sqlite+aiosqlite:///test.db")
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    yield engine
    await engine.dispose()

The fixture is scope="session" — created once, shared across all tests. But pytest-asyncio's default event loop is scope="function" — new loop per test. When the second test runs, the session-scoped fixture's coroutines try to run on a closed event loop. Crash.

Fix: match your event loop scope to your most-broadly-scoped fixture:

# pytest.ini
[pytest]
asyncio_mode = auto

# conftest.py
import pytest

@pytest.fixture(scope="session")
def event_loop_policy():
    # pytest-asyncio 0.23+: use loop_scope on fixtures instead
    pass

For pytest-asyncio 0.23+, the preferred approach is loop_scope on the fixture:

# conftest.py
import pytest
import asyncio

@pytest.fixture(scope="session", loop_scope="session")
async def db_engine():
    engine = create_async_engine("sqlite+aiosqlite:///test.db")
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    yield engine
    await engine.dispose()

@pytest.fixture(loop_scope="session")  # matches db_engine scope
async def db_session(db_engine):
    async with AsyncSession(db_engine) as session:
        yield session
        await session.rollback()

Rule: If a fixture has loop_scope="session", every fixture it depends on needs loop_scope="session" too. The error message when you get this wrong is ScopeMismatch — the loop_scope tells you exactly which fixture is the problem.

Part 2: Testing FastAPI Async Endpoints

The correct client for async FastAPI testing is httpx.AsyncClient, not TestClient.

TestClient works by running the async app in a thread with its own event loop. It works for simple tests, but it can't share async fixtures — each TestClient call is isolated. You can't, for example, pre-populate an async database fixture and then call the endpoint that reads it. The contexts don't share.

The right pattern:

# pip install httpx

import pytest
from httpx import AsyncClient, ASGITransport
from myapp import app, get_db
from myapp.database import AsyncSession

@pytest.fixture(loop_scope="session")
async def db_engine():
    engine = create_async_engine("sqlite+aiosqlite:///./test.db")
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    yield engine
    await engine.dispose()

@pytest.fixture
async def db_session(db_engine):
    async with AsyncSession(db_engine) as session:
        yield session
        await session.rollback()

@pytest.fixture
async def client(db_session):
    # Override the app's dependency to use the test database session
    async def override_get_db():
        yield db_session

    app.dependency_overrides[get_db] = override_get_db

    async with AsyncClient(
        transport=ASGITransport(app=app),
        base_url="http://test"
    ) as ac:
        yield ac

    app.dependency_overrides.clear()

# Test — both the endpoint and its database state are in the same async context
async def test_create_user(client, db_session):
    response = await client.post("/users", json={"email": "test@example.com"})
    assert response.status_code == 201

    # Verify in database — same session, same transaction
    result = await db_session.execute(select(User).where(User.email == "test@example.com"))
    user = result.scalar_one()
    assert user.email == "test@example.com"

The key: app.dependency_overrides lets you inject the test database session into the FastAPI dependency graph. The client and the database fixture share the same async context.

What TestClient gets you vs. AsyncClient:

	`TestClient`	`AsyncClient`
Sync test functions	✅	❌ (needs async test)
Share async fixtures	❌	✅
Test websockets	Limited	✅
Startup/shutdown events	✅	✅
Works with sync routes	✅	✅

Use TestClient for simple smoke tests. Use AsyncClient when you need fixture sharing.

Part 3: Testing Async Background Tasks

Background tasks are the hardest async pattern to test because they run outside the request/response cycle.

FastAPI BackgroundTasks — the problem:

# endpoint
@app.post("/send-email")
async def send_email(background_tasks: BackgroundTasks, email: EmailRequest):
    background_tasks.add_task(send_email_async, email.to, email.subject, email.body)
    return {"status": "queued"}

If you call this endpoint in a test and immediately check the result, the background task may not have run yet. You're racing the event loop.

Fix 1: mock the task function

from unittest.mock import AsyncMock, patch

async def test_send_email_queues_task(client):
    with patch("myapp.routes.send_email_async", new_callable=AsyncMock) as mock_send:
        response = await client.post("/send-email", json={
            "to": "user@example.com",
            "subject": "Test",
            "body": "Hello"
        })

        assert response.status_code == 200
        mock_send.assert_called_once_with(
            "user@example.com", "Test", "Hello"
        )

This verifies the task was scheduled with the right arguments. You're not testing the task itself — you're testing that the endpoint correctly delegates.

Fix 2: test the task directly

async def test_send_email_task_sends_smtp(smtp_mock):
    # Test the task function in isolation, not through the endpoint
    await send_email_async("user@example.com", "Subject", "Body")
    smtp_mock.send_message.assert_called_once()

Fix 3: when you need end-to-end

If you need to verify the full flow (endpoint → task → side effect), you need to wait for the task:

import asyncio
from unittest.mock import patch, AsyncMock

async def test_email_sent_end_to_end(client):
    sent_emails = []

    async def capture_email(to, subject, body):
        sent_emails.append({"to": to, "subject": subject})

    with patch("myapp.routes.send_email_async", side_effect=capture_email):
        response = await client.post("/send-email", json={
            "to": "user@example.com", "subject": "Test", "body": "Hello"
        })
        # Give the event loop time to run the background task
        await asyncio.sleep(0)

        assert len(sent_emails) == 1
        assert sent_emails[0]["to"] == "user@example.com"

await asyncio.sleep(0) yields control to the event loop, allowing queued coroutines to run. This is enough for simple background tasks. For tasks that involve real I/O, you'll need either a mock or a proper wait condition.

Part 4: Async Fixtures — What's Different

Async fixtures work exactly like sync fixtures, with two differences:

1. Use yield in async fixtures:

@pytest.fixture
async def redis_client():
    client = await aioredis.from_url("redis://localhost")
    yield client
    await client.aclose()  # cleanup runs automatically after each test

2. Fixture scope interacts with event loop scope:

# BAD: session-scoped fixture on function-scoped loop (crashes on second test)
@pytest.fixture(scope="session")
async def redis_pool():
    pool = await aioredis.from_url("redis://localhost", max_connections=10)
    yield pool
    await pool.aclose()

# GOOD: match loop_scope to fixture scope
@pytest.fixture(scope="session", loop_scope="session")
async def redis_pool():
    pool = await aioredis.from_url("redis://localhost", max_connections=10)
    yield pool
    await pool.aclose()

Common async fixture patterns:

# conftest.py

# One database per test run
@pytest.fixture(scope="session", loop_scope="session")
async def db_engine():
    engine = create_async_engine(
        "sqlite+aiosqlite:///./test.db",
        echo=False,
    )
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    yield engine
    await engine.dispose()

# One session per test, rolled back after
@pytest.fixture
async def db_session(db_engine):
    async with AsyncSession(db_engine) as session:
        async with session.begin():
            yield session
            await session.rollback()

# Test data seeding
@pytest.fixture
async def sample_users(db_session):
    users = [
        User(email="alice@example.com", name="Alice"),
        User(email="bob@example.com", name="Bob"),
    ]
    db_session.add_all(users)
    await db_session.flush()  # assigns IDs without committing
    return users

Part 5: Testing Async Code That Calls External Services

The pattern for mocking external async calls:

from unittest.mock import AsyncMock, patch

# AsyncMock is the async version of MagicMock
async def test_openai_call(client):
    mock_response = AsyncMock()
    mock_response.choices = [AsyncMock(message=AsyncMock(content="Hello"))]

    with patch("myapp.services.openai_client.chat.completions.create",
               return_value=mock_response) as mock_create:
        response = await client.post("/summarize", json={"text": "Long text..."})

        assert response.status_code == 200
        mock_create.assert_called_once()

When to use AsyncMock vs MagicMock:

# Use AsyncMock when mocking an async function
async_function_mock = AsyncMock(return_value="result")
await async_function_mock()  # works

# Use MagicMock for sync code in async context
sync_function_mock = MagicMock(return_value="result")
sync_function_mock()  # works, no await

# Gotcha: MagicMock on an async function returns a coroutine object, not the value
async_fn_wrong = MagicMock(return_value="result")
result = await async_fn_wrong()  # TypeError: object MagicMock can't be used in 'await'

Rule: if the function you're mocking has async def, use AsyncMock.

Part 6: The Full conftest.py

A complete conftest.py for a FastAPI + async SQLAlchemy + Redis application:

# conftest.py
import asyncio
import pytest
import pytest_asyncio
from httpx import AsyncClient, ASGITransport
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker
import aioredis

from myapp import app
from myapp.database import get_db, get_redis, Base

# Session-scoped: created once for all tests
@pytest.fixture(scope="session", loop_scope="session")
async def db_engine():
    engine = create_async_engine("sqlite+aiosqlite:///./test.db")
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    yield engine
    await engine.dispose()

@pytest.fixture(scope="session", loop_scope="session")
async def redis_pool():
    pool = await aioredis.from_url("redis://localhost:6379/15")  # test DB
    yield pool
    await pool.flushdb()  # clean up test data
    await pool.aclose()

# Function-scoped: fresh state per test
@pytest.fixture
async def db_session(db_engine):
    async_session = sessionmaker(db_engine, class_=AsyncSession, expire_on_commit=False)
    async with async_session() as session:
        async with session.begin():
            yield session
            await session.rollback()

@pytest.fixture
async def redis_client(redis_pool):
    yield redis_pool
    await redis_pool.flushdb()  # clean test keys after each test

@pytest.fixture
async def client(db_session, redis_client):
    async def override_db():
        yield db_session

    async def override_redis():
        return redis_client

    app.dependency_overrides[get_db] = override_db
    app.dependency_overrides[get_redis] = override_redis

    async with AsyncClient(
        transport=ASGITransport(app=app),
        base_url="http://test"
    ) as ac:
        yield ac

    app.dependency_overrides.clear()

The Quick Reference

"Event loop is closed" → Your fixture scope doesn't match the loop scope. Add loop_scope="session" to session-scoped async fixtures.

"ScopeMismatch" → A narrow-scoped fixture depends on a wide-scoped one. Check the fixture chain and align scopes.

Tests pass individually but fail in batch → Shared state between tests. Add rollback in db_session fixture, flushdb in redis fixture.

Can't use async fixture in sync test → Wrap with asyncio.run() or make the test async.

AsyncMock returns wrong type → Check you're using AsyncMock (not MagicMock) for async functions.

What You Can Test Without Any of This

Not every async test needs a complex setup. For pure async logic with no I/O:

async def test_retry_logic():
    call_count = 0

    async def flaky_service():
        nonlocal call_count
        call_count += 1
        if call_count < 3:
            raise ConnectionError("timeout")
        return "success"

    result = await retry_with_backoff(flaky_service, max_retries=3)
    assert result == "success"
    assert call_count == 3

Pure async logic — no fixtures, no mocks, no event loop configuration. Most of the complexity in this guide is about testing async code that touches shared resources (databases, caches, external services). If your async function is pure, just await it.

The Products Behind This Guide

The async testing patterns here — the retry wrapper, the session fixtures, the dependency override pattern — are adapted from the Python Automation Cookbook ($39): 25 production-ready Python scripts with full test suites. Every script has been debugged against real production workloads.

The conftest.py above is a simplified version of the one included with the cookbook. The full version adds:

Parallel test support (pytest-xdist with async isolation)
Test data factories (factory_boy async adapters)
Performance assertion helpers (verify endpoint latency under load)

→ Python Automation Cookbook on Gumroad

If this was useful, the pytest fixtures guide and pytest plugins guide cover the sync side of the same patterns.

DEV Community