Async Testing — AsyncClient and WebSocket Testing

FastAPI’s TestClient is synchronous — it wraps the ASGI app in a synchronous interface using anyio, which works well for most tests. But for testing async def route handlers that use async SQLAlchemy, async dependencies, or WebSocket endpoints, httpx.AsyncClient and pytest-asyncio provide a true async test environment. WebSocket testing requires a different approach — either the starlette test client’s with client.websocket_connect() context manager or the httpx-ws library for async WebSocket tests.

Async Tests with AsyncClient #

# tests/conftest.py — async fixtures
import pytest
import pytest_asyncio
from httpx import AsyncClient, ASGITransport
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine, async_sessionmaker
from app.main import app
from app.dependencies import get_async_db, get_current_user
from app.database import Base

# Async test engine
ASYNC_TEST_URL = "sqlite+aiosqlite:///:memory:"
async_test_engine = create_async_engine(ASYNC_TEST_URL)
AsyncTestSession  = async_sessionmaker(async_test_engine, expire_on_commit=False)

@pytest_asyncio.fixture(scope="session", autouse=True)
async def async_create_tables():
    async with async_test_engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    yield
    async with async_test_engine.begin() as conn:
        await conn.run_sync(Base.metadata.drop_all)

@pytest_asyncio.fixture
async def async_db():
    async with AsyncTestSession() as session:
        yield session
        await session.rollback()

@pytest_asyncio.fixture
async def async_client(async_db, user):
    async def override_db():
        yield async_db

    def override_user():
        return user

    app.dependency_overrides[get_async_db]     = override_db
    app.dependency_overrides[get_current_user] = override_user

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

    app.dependency_overrides.clear()

# ── Async test functions ───────────────────────────────────────────────────────
@pytest.mark.asyncio
async def test_async_list_posts(async_client):
    response = await async_client.get("/posts/")
    assert response.status_code == 200
    assert "items" in response.json()

@pytest.mark.asyncio
async def test_async_create_post(async_client):
    response = await async_client.post("/posts/", json={
        "title": "Async Test Post",
        "body":  "Body content for async test.",
        "slug":  "async-test-post",
    })
    assert response.status_code == 201

WebSocket Tests #

from fastapi.testclient import TestClient
from starlette.testclient import WebSocketTestSession
from app.main import app
import json

def test_websocket_echo():
    """Test basic WebSocket echo endpoint."""
    client = TestClient(app)

    with client.websocket_connect("/ws/echo") as ws:
        ws.send_text("hello")
        data = ws.receive_text()
        assert data == "Echo: hello"

def test_websocket_json_exchange():
    client = TestClient(app)
    with client.websocket_connect("/ws/json") as ws:
        ws.send_json({"type": "ping", "payload": 42})
        response = ws.receive_json()
        assert response["status"] == "ok"
        assert response["received"]["type"] == "ping"

def test_websocket_auth_rejects_invalid_token():
    client = TestClient(app)
    # close() with code 1008 expected for invalid token
    with client.websocket_connect("/ws/notifications?token=invalid_token") as ws:
        # Depending on implementation: either raises or returns close event
        # Best pattern: server closes immediately after receiving invalid token
        try:
            ws.receive_json()   # may raise if connection was closed
        except Exception:
            pass   # connection was closed — expected behaviour

def test_websocket_auth_accepts_valid_token(client, user):
    from app.auth.tokens import create_access_token
    token = create_access_token(user.id, user.role)

    with client.websocket_connect(f"/ws/notifications?token={token}") as ws:
        ws.send_json({"type": "ping"})
        response = ws.receive_json()
        assert response["type"] == "pong"

Testing SSE Endpoints #

import pytest
from fastapi.testclient import TestClient
from app.main import app

def test_sse_heartbeat_streams_events():
    """Verify SSE endpoint produces valid event stream data."""
    client = TestClient(app)
    with client.stream("GET", "/events/heartbeat") as response:
        assert response.status_code == 200
        assert "text/event-stream" in response.headers["content-type"]

        # Read first event
        lines = []
        for line in response.iter_lines():
            lines.append(line)
            if line == "":   # blank line = event boundary
                break

        # Verify event format
        data_line = next((l for l in lines if l.startswith("data:")), None)
        assert data_line is not None
        import json
        event_data = json.loads(data_line.removeprefix("data: "))
        assert "time" in event_data

Common Mistakes #

Mistake 1 — Using @pytest.fixture on an async fixture #

❌ Wrong — returns coroutine, not the value:

@pytest.fixture   # should be @pytest_asyncio.fixture
async def async_client():
    async with AsyncClient(...) as c:
        yield c   # not run — returns coroutine object to test!

✅ Correct:

@pytest_asyncio.fixture
async def async_client(): ...

Mistake 2 — Not awaiting async client calls #

❌ Wrong — response is a coroutine, not a Response object:

async def test_posts(async_client):
    response = async_client.get("/posts/")   # missing await!
    assert response.status_code == 200       # AttributeError: coroutine has no status_code

✅ Correct:

response = await async_client.get("/posts/")   # ✓

Quick Reference #