FastAPI

FastAPI Async Endpoint Testing with pytest-asyncio

HelpMeTest

HelpMeTest

— 8 min read
FastAPI Async Endpoint Testing with pytest-asyncio

FastAPI is built on top of Starlette and runs on an ASGI server. Most of its interesting behavior — middleware, database calls, background tasks — happens asynchronously. Testing this code properly requires pytest-asyncio and a solid understanding of how async test infrastructure works. This guide goes beyond the basics to cover async fixtures, concurrent request testing, and async context managers in test setups.

Setting Up pytest-asyncio

Install the required packages:

pip install pytest-asyncio httpx anyio

Configure pytest.ini or pyproject.toml:

# pyproject.toml
[tool.pytest.ini_options]
asyncio_mode = "auto"

Or per-file using a marker:

# conftest.py
import pytest

# Apply auto mode to all tests in the project
pytest_plugins = ['pytest_asyncio']

The asyncio_mode = "auto" setting means all async def test functions are automatically treated as async tests — you don't need @pytest.mark.asyncio on every function.

anyio Mode vs asyncio Mode

pytest-asyncio supports two backends: asyncio (default) and anyio. FastAPI itself uses anyio under the hood, so using anyio mode can expose subtle bugs that asyncio mode masks.

# pyproject.toml — anyio backend
[tool.pytest.ini_options]
asyncio_mode = "auto"

[tool.anyio]
backends = ["asyncio", "trio"]  # test with both!
# tests/test_anyio_mode.py
import pytest
import anyio
from httpx import AsyncClient, ASGITransport
from app.main import app

# This test runs under both asyncio AND trio backends
@pytest.mark.anyio
async def test_endpoint_works_on_both_backends():
    async with AsyncClient(
        transport=ASGITransport(app=app),
        base_url="http://test"
    ) as client:
        response = await client.get("/health")
    assert response.status_code == 200

# Restrict to asyncio only when you need asyncio-specific behavior
@pytest.mark.asyncio
async def test_asyncio_specific():
    import asyncio
    loop = asyncio.get_event_loop()
    assert loop is not None

Running with pytest --anyio-backends=asyncio,trio will run anyio-marked tests twice — once per backend. This catches issues where your code accidentally depends on asyncio-specific behavior.

Async Fixtures

Async fixtures work just like sync fixtures but use async def. They can await setup and teardown code:

# tests/conftest.py
import pytest
import pytest_asyncio
from httpx import AsyncClient, ASGITransport
from app.main import app
from app.database import engine, Base

@pytest_asyncio.fixture(scope="session")
async def create_tables():
    """Create all tables once per test session."""
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    yield
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.drop_all)

@pytest_asyncio.fixture(scope="function")
async def async_client(create_tables):
    """Fresh async client per test function."""
    async with AsyncClient(
        transport=ASGITransport(app=app),
        base_url="http://testserver"
    ) as client:
        yield client

@pytest_asyncio.fixture
async def authenticated_async_client(async_client):
    """Client with auth token pre-injected."""
    # Get a real token through the auth endpoint
    response = await async_client.post("/auth/token", data={
        "username": "testuser",
        "password": "testpassword"
    })
    token = response.json()["access_token"]
    async_client.headers["Authorization"] = f"Bearer {token}"
    yield async_client

Notice @pytest_asyncio.fixture instead of @pytest.fixture — this is required for async fixtures. Using @pytest.fixture with an async def function will give you a coroutine object rather than the awaited result.

# tests/test_async_fixtures.py
async def test_authenticated_request(authenticated_async_client):
    response = await authenticated_async_client.get("/users/me")
    assert response.status_code == 200
    assert "email" in response.json()

async def test_unauthenticated_request_rejected(async_client):
    response = await async_client.get("/users/me")
    assert response.status_code == 401

Testing Async Middleware

Async middleware can modify requests and responses, add headers, log requests, or short-circuit processing. Testing it requires verifying both the middleware behavior and its effect on endpoints.

# app/middleware.py
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.requests import Request
from starlette.responses import Response
import time
import logging

logger = logging.getLogger(__name__)

class RequestTimingMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        start_time = time.perf_counter()
        response = await call_next(request)
        process_time = time.perf_counter() - start_time
        response.headers["X-Process-Time"] = str(round(process_time * 1000, 2))
        return response

class RateLimitMiddleware(BaseHTTPMiddleware):
    def __init__(self, app, max_requests: int = 100, window: int = 60):
        super().__init__(app)
        self.max_requests = max_requests
        self.window = window
        self._requests: dict = {}

    async def dispatch(self, request: Request, call_next):
        client_ip = request.client.host
        now = time.time()

        # Clean old entries
        self._requests = {
            k: v for k, v in self._requests.items()
            if now - v["first_seen"] < self.window
        }

        entry = self._requests.get(client_ip, {"count": 0, "first_seen": now})
        entry["count"] += 1
        self._requests[client_ip] = entry

        if entry["count"] > self.max_requests:
            return Response(
                content='{"detail": "Rate limit exceeded"}',
                status_code=429,
                media_type="application/json"
            )

        return await call_next(request)
# tests/test_middleware.py
import pytest
from httpx import AsyncClient, ASGITransport
from fastapi import FastAPI
from app.middleware import RequestTimingMiddleware, RateLimitMiddleware
from app.routers import api_router

@pytest.fixture
def app_with_timing():
    test_app = FastAPI()
    test_app.add_middleware(RequestTimingMiddleware)
    test_app.include_router(api_router)
    return test_app

@pytest.fixture
def app_with_rate_limit():
    test_app = FastAPI()
    test_app.add_middleware(RateLimitMiddleware, max_requests=3, window=60)
    test_app.include_router(api_router)
    return test_app

async def test_timing_middleware_adds_header(app_with_timing):
    async with AsyncClient(
        transport=ASGITransport(app=app_with_timing),
        base_url="http://test"
    ) as client:
        response = await client.get("/health")

    assert response.status_code == 200
    assert "x-process-time" in response.headers
    process_time = float(response.headers["x-process-time"])
    assert 0 <= process_time < 5000  # Less than 5 seconds in ms

async def test_rate_limit_middleware_blocks_after_threshold(app_with_rate_limit):
    async with AsyncClient(
        transport=ASGITransport(app=app_with_rate_limit),
        base_url="http://test"
    ) as client:
        # First 3 requests should succeed
        for i in range(3):
            response = await client.get("/health")
            assert response.status_code == 200, f"Request {i+1} should succeed"

        # 4th request should be rate limited
        response = await client.get("/health")
        assert response.status_code == 429
        assert "Rate limit exceeded" in response.json()["detail"]

Async Database Calls in Tests

Testing endpoints that make async database calls requires an async test database session. The key pattern is using a transaction per test that gets rolled back after:

# tests/conftest.py
import pytest
import pytest_asyncio
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker
from app.models import Base
from app.dependencies import get_db
from app.main import app

TEST_DATABASE_URL = "postgresql+asyncpg://postgres:password@localhost:5432/test_db"

@pytest_asyncio.fixture(scope="session")
async def engine():
    engine = create_async_engine(TEST_DATABASE_URL, echo=False)
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    yield engine
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.drop_all)
    await engine.dispose()

@pytest_asyncio.fixture
async def db_session(engine):
    """Rolls back all changes after each test."""
    connection = await engine.connect()
    transaction = await connection.begin()

    session_factory = sessionmaker(
        bind=connection,
        class_=AsyncSession,
        expire_on_commit=False,
    )
    session = session_factory()

    yield session

    await session.close()
    await transaction.rollback()
    await connection.close()

@pytest_asyncio.fixture
async def client(db_session):
    """Client using the test 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://testserver"
    ) as c:
        yield c

    app.dependency_overrides.clear()
# tests/test_database_endpoints.py
from app.models import User

async def test_create_and_retrieve_user(client, db_session):
    # Create via API
    response = await client.post("/users", json={
        "email": "test@example.com",
        "name": "Test User"
    })
    assert response.status_code == 201
    user_id = response.json()["id"]

    # Verify in database directly
    result = await db_session.get(User, user_id)
    assert result is not None
    assert result.email == "test@example.com"

    # Retrieve via API
    response = await client.get(f"/users/{user_id}")
    assert response.status_code == 200
    assert response.json()["email"] == "test@example.com"

async def test_database_changes_rolled_back_between_tests(client, db_session):
    # This test runs with a clean database — previous test's data was rolled back
    from sqlalchemy import select
    from app.models import User
    result = await db_session.execute(select(User))
    users = result.scalars().all()
    assert len(users) == 0

Concurrent Request Testing

One of the most valuable things async testing enables is testing concurrent requests to verify your application handles parallelism correctly.

# tests/test_concurrent.py
import asyncio
import pytest
from httpx import AsyncClient, ASGITransport
from app.main import app

@pytest_asyncio.fixture
async def client():
    async with AsyncClient(
        transport=ASGITransport(app=app),
        base_url="http://testserver"
    ) as c:
        yield c

async def test_concurrent_requests_all_succeed(client):
    """Verify the app handles 50 concurrent requests without errors."""
    tasks = [
        client.get("/api/products")
        for _ in range(50)
    ]
    responses = await asyncio.gather(*tasks)

    assert all(r.status_code == 200 for r in responses)

async def test_concurrent_writes_maintain_consistency(client):
    """Concurrent counter increments should all register."""
    # Reset counter
    await client.post("/counter/reset")

    # Fire 20 concurrent increments
    tasks = [client.post("/counter/increment") for _ in range(20)]
    responses = await asyncio.gather(*tasks)

    assert all(r.status_code == 200 for r in responses)

    # Check final value
    response = await client.get("/counter")
    assert response.json()["value"] == 20

async def test_concurrent_requests_with_different_users(client):
    """Multiple users hitting the API simultaneously."""
    async def user_flow(user_id: int):
        # Each user creates a resource and retrieves it
        create_resp = await client.post("/items", json={
            "name": f"item-{user_id}",
            "owner_id": user_id
        })
        assert create_resp.status_code == 201
        item_id = create_resp.json()["id"]

        get_resp = await client.get(f"/items/{item_id}")
        assert get_resp.status_code == 200
        assert get_resp.json()["owner_id"] == user_id
        return get_resp.json()

    results = await asyncio.gather(*[user_flow(i) for i in range(10)])
    assert len(results) == 10
    # Each user got their own item back
    owner_ids = {r["owner_id"] for r in results}
    assert owner_ids == set(range(10))

async def test_request_timeout_handling(client):
    """Endpoints that take too long should be cancelled properly."""
    with pytest.raises(Exception):  # TimeoutException or similar
        await asyncio.wait_for(
            client.get("/slow-endpoint"),
            timeout=0.01  # 10ms timeout — will fail for slow endpoints
        )

async def test_concurrent_requests_dont_share_state(client):
    """Verify no request-level state leaks between concurrent requests."""
    async def make_request(request_id: int):
        response = await client.post("/echo-context", json={"request_id": request_id})
        return response.json()["echoed_id"]

    results = await asyncio.gather(*[make_request(i) for i in range(30)])

    # Each request should get back its own ID, not another request's
    for i, result in enumerate(results):
        assert result == i

Async Context Managers in Test Setup

Complex test setups often require async context managers — for example, spinning up a mock server, establishing a WebSocket connection, or managing async resources.

# tests/helpers.py
from contextlib import asynccontextmanager
import asyncio
from aiohttp import web
import aiohttp

@asynccontextmanager
async def mock_external_service(responses: dict):
    """Start a real HTTP server for testing external service calls."""
    async def handler(request):
        path = request.path
        if path in responses:
            data = responses[path]
            return web.json_response(data)
        return web.Response(status=404)

    app = web.Application()
    app.router.add_get("/{path_info:.*}", handler)

    runner = web.AppRunner(app)
    await runner.setup()
    site = web.TCPSite(runner, "localhost", 0)
    await site.start()

    port = site._server.sockets[0].getsockname()[1]
    base_url = f"http://localhost:{port}"

    yield base_url

    await runner.cleanup()
# tests/test_external_calls.py
import pytest
import pytest_asyncio
from httpx import AsyncClient, ASGITransport
from app.main import app
from app.dependencies import get_external_service_url
from tests.helpers import mock_external_service

async def test_endpoint_calls_external_service():
    external_data = {
        "/api/users/42": {"id": 42, "name": "External User", "verified": True}
    }

    async with mock_external_service(external_data) as mock_url:
        # Override the external service URL dependency
        app.dependency_overrides[get_external_service_url] = lambda: mock_url

        async with AsyncClient(
            transport=ASGITransport(app=app),
            base_url="http://testserver"
        ) as client:
            response = await client.get("/proxy/users/42")

        app.dependency_overrides.clear()

    assert response.status_code == 200
    assert response.json()["name"] == "External User"

async def test_websocket_endpoint():
    from httpx_ws import aconnect_ws

    async with AsyncClient(
        transport=ASGITransport(app=app),
        base_url="http://testserver"
    ) as client:
        async with aconnect_ws("/ws/chat/room1", client) as ws:
            await ws.send_text('{"type": "message", "content": "Hello"}')
            data = await ws.receive_text()
            import json
            msg = json.loads(data)
            assert msg["type"] == "echo"
            assert msg["content"] == "Hello"

AsyncExitStack for Multiple Async Context Managers

When your test setup involves multiple async resources, AsyncExitStack prevents nested context manager hell:

# tests/test_complex_setup.py
from contextlib import AsyncExitStack
import pytest_asyncio
from httpx import AsyncClient, ASGITransport
from app.main import app

@pytest_asyncio.fixture
async def complex_client(mock_redis, mock_s3):
    async with AsyncExitStack() as stack:
        # Each resource is properly cleaned up even if a later one fails
        redis = await stack.enter_async_context(mock_redis.connect())
        s3 = await stack.enter_async_context(mock_s3.session())
        client = await stack.enter_async_context(
            AsyncClient(transport=ASGITransport(app=app), base_url="http://test")
        )

        app.state.redis = redis
        app.state.s3 = s3

        yield client

async def test_file_upload_stores_in_s3_and_caches_url(complex_client):
    with open("tests/fixtures/sample.pdf", "rb") as f:
        response = await complex_client.post(
            "/documents/upload",
            files={"file": ("sample.pdf", f, "application/pdf")}
        )

    assert response.status_code == 201
    doc_id = response.json()["id"]
    assert response.json()["url"].startswith("https://")

    # Verify subsequent access is served from cache
    response = await complex_client.get(f"/documents/{doc_id}")
    assert response.headers.get("x-cache") == "HIT"

Event Loop Configuration

For tests that create and consume async queues or streams, you may need to configure the event loop scope:

# conftest.py
import pytest

@pytest.fixture(scope="session")
def event_loop_policy():
    """Use uvloop for session-scoped tests if available."""
    try:
        import uvloop
        return uvloop.EventLoopPolicy()
    except ImportError:
        import asyncio
        return asyncio.DefaultEventLoopPolicy()

The most common pitfall with pytest-asyncio is mixing sync and async fixtures — sync fixtures can't await things, and if you accidentally use @pytest.fixture on an async def, you'll get confusing errors. Stick to @pytest_asyncio.fixture for anything that uses await, and you'll avoid the majority of async testing headaches.

Read more

Start now free