Pytest¶
Testing helpers for FastAPI applications with async client, database sessions, and parallel worker support.
Installation¶
Overview¶
The pytest module provides utilities for setting up async test clients, managing test database sessions, and supporting parallel test execution with pytest-xdist.
Creating an async client¶
Use create_async_client to get an httpx.AsyncClient configured for your FastAPI app:
from fastapi_toolsets.pytest import create_async_client
@pytest.fixture
async def http_client(db_session):
async def _override_get_db():
yield db_session
async with create_async_client(
app=app,
base_url="http://127.0.0.1/api/v1",
dependency_overrides={get_db: _override_get_db},
) as c:
yield c
Database sessions in tests¶
Use create_db_session to create an isolated AsyncSession for a test, combined with create_worker_database to set up a per-worker database:
from fastapi_toolsets.pytest import create_worker_database, create_db_session
@pytest.fixture(scope="session")
async def worker_db_url():
async with create_worker_database(
database_url=str(settings.SQLALCHEMY_DATABASE_URI)
) as url:
yield url
@pytest.fixture
async def db_session(worker_db_url):
async with create_db_session(
database_url=worker_db_url, base=Base, cleanup=True
) as session:
yield session
Info
In this example, the database is reset between each test using the argument cleanup=True.
Use worker_database_url to derive the per-worker URL manually if needed:
from fastapi_toolsets.pytest import worker_database_url
url = worker_database_url("postgresql+asyncpg://user:pass@localhost/test_db", default_test_db="test")
# e.g. "postgresql+asyncpg://user:pass@localhost/test_db_gw0" under xdist
Parallel testing with pytest-xdist¶
The examples above are already compatible with parallel test execution with pytest-xdist.
Cleaning up tables¶
If you want to manually clean up a database you can use cleanup_tables, this will truncate all tables between tests for fast isolation:
from fastapi_toolsets.db import cleanup_tables
@pytest.fixture(autouse=True)
async def clean(db_session):
yield
await cleanup_tables(session=db_session, base=Base)