Async Testing
When your FastAPI app uses async dependencies (async DB sessions, async HTTP clients), you need an async test client to test them properly. httpx.AsyncClient with ASGITransport lets you test against the live ASGI app without running a server.
Learning Focus
By the end of this lesson you can: write async def tests with pytest-asyncio, set up a test database session, override async dependencies, and run async test suites efficiently.
Installation
install-async-test.sh
pip install pytest-asyncio httpx aiosqlite
Basic Async Test
tests/test_async.py
import pytest
import httpx
from app.main import app
@pytest.mark.asyncio
async def test_health():
async with httpx.AsyncClient(
transport=httpx.ASGITransport(app=app),
base_url="http://test",
) as client:
resp = await client.get("/health")
assert resp.status_code == 200
assert resp.json() == {"status": "ok"}
Async Client Fixture
tests/conftest.py
import pytest
import pytest_asyncio
import httpx
from app.main import app
@pytest_asyncio.fixture
async def async_client():
async with httpx.AsyncClient(
transport=httpx.ASGITransport(app=app),
base_url="http://test",
) as client:
yield client
tests/test_items.py
import pytest
@pytest.mark.asyncio
async def test_create_item(async_client):
resp = await async_client.post("/items/", json={"name": "Widget", "price": 5.0})
assert resp.status_code == 201
assert resp.json()["name"] == "Widget"
Test Database Setup
Use an in-memory SQLite database for tests:
tests/conftest.py
import pytest
import pytest_asyncio
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession
from app.db.base import Base
from app.db.session import get_db
from app.main import app
TEST_DATABASE_URL = "sqlite+aiosqlite:///:memory:"
@pytest_asyncio.fixture(scope="session")
async def test_engine():
engine = create_async_engine(TEST_DATABASE_URL)
async with engine.begin() as conn:
await conn.run_sync(Base.metadata.create_all)
yield engine
await engine.dispose()
@pytest_asyncio.fixture
async def db_session(test_engine):
factory = async_sessionmaker(test_engine, expire_on_commit=False)
async with factory() as session:
yield session
await session.rollback() # Roll back after each test for isolation
@pytest_asyncio.fixture
async def async_client(db_session: AsyncSession):
async def override_get_db():
yield db_session
app.dependency_overrides[get_db] = override_get_db
async with httpx.AsyncClient(
transport=httpx.ASGITransport(app=app),
base_url="http://test",
) as client:
yield client
app.dependency_overrides.clear()
pytest.ini / pyproject.toml Configuration
pytest.ini
[pytest]
asyncio_mode = auto
Or in pyproject.toml:
pyproject.toml
[tool.pytest.ini_options]
asyncio_mode = "auto"
Full CRUD Test Suite
tests/test_users.py
import pytest
@pytest.mark.asyncio
async def test_create_user(async_client):
resp = await async_client.post("/users/", json={
"username": "alice",
"email": "alice@example.com",
"password": "Secur3Pass!",
})
assert resp.status_code == 201
data = resp.json()
assert data["username"] == "alice"
assert "password" not in data # Never exposed
@pytest.mark.asyncio
async def test_get_user_not_found(async_client):
resp = await async_client.get("/users/99999")
assert resp.status_code == 404
@pytest.mark.asyncio
async def test_duplicate_email(async_client):
payload = {"username": "bob", "email": "bob@example.com", "password": "Secur3Pass!"}
await async_client.post("/users/", json=payload)
resp = await async_client.post("/users/", json=payload)
assert resp.status_code == 409 # Conflict
Common Pitfalls
| Pitfall | Cause / Symptom | Fix |
|---|---|---|
ScopeMismatch error | Session-scoped fixture used by function-scoped fixture | Match fixture scopes or use scope="function" |
| Tests not isolated | Same DB session across tests without rollback | Roll back after each test |
asyncio_mode not set | pytest-asyncio doesn't detect async def tests | Set asyncio_mode = auto in pytest config |
aiosqlite not installed | SQLite async tests fail | pip install aiosqlite |
| App lifespan not triggered in tests | DB warm-up or init code not running | Use async with app.router.lifespan_context(app) or ASGITransport which respects lifespan |
Hands-On Practice
run-async-tests.sh
pip install pytest pytest-asyncio httpx aiosqlite
# Run all async tests
pytest tests/ -v --asyncio-mode=auto
# Run specific test file
pytest tests/test_users.py -v
# Show coverage
pytest tests/ --cov=app --cov-report=term-missing