Pytest¶

Testing helpers for FastAPI applications: async HTTP client, database sessions, and parallel worker support.

Installation¶

uvpip

uv add "fastapi-toolsets[pytest]"

pip install "fastapi-toolsets[pytest]"

Async client¶

Use create_async_client to get an httpx.AsyncClient bound to 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

dependency_overrides mutates the app, not the client

app.dependency_overrides belongs to the app, which is normally a module-level singleton shared by every client. Each client layers its overrides on top of what was already there and unwinds them on exit, so clients compose in any order and leave nothing behind.

The app still has a single slot per dependency, so two clients open at once on the same key both resolve to the one applied most recently. When overlapping clients need to disagree, put the overrides in a test-scoped fixture and let the clients add none.

Any extra keyword arguments are forwarded to httpx.AsyncClient, so you can set default headers, authentication, timeouts, and more:

async with create_async_client(
    app=app,
    headers={"X-Api-Key": "secret"},
    timeout=10,
) as c:
    ...

Database sessions¶

Use create_worker_database + create_db_session to get a fully isolated AsyncSession for each test:

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

create_worker_database connects without specifying a database (asyncpg falls back to the username), so the target test database does not need to exist beforehand.

Info

cleanup=True truncates all tables between tests via TRUNCATE … RESTART IDENTITY CASCADE, which is faster than dropping and recreating tables.

The session is an EventSession

create_db_session builds its factory with class_=EventSession, so lifecycle events always fire in tests. Your application only dispatches them if its Database was built with session_class=EventSession, so this fixture cannot tell you whether the app is wired correctly.

Engine and session options¶

Pass engine_kwargs or session_kwargs to forward options to the underlying SQLAlchemy primitives:

async with create_db_session(
    database_url=worker_db_url,
    base=Base,
    engine_kwargs={"pool_size": 5, "connect_args": {"timeout": 10}},
    session_kwargs={"autoflush": False},
) as session:
    ...

Parallel testing with pytest-xdist¶

The fixtures above work with pytest-xdist out of the box. Each worker gets its own database named after the worker (e.g. gw0, gw1). Pass prefix to namespace the database (e.g. prefix="myapp" → myapp_gw0).

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/myapp", default_test_db="test"
)
# → "postgresql+asyncpg://user:pass@localhost/gw0" under xdist
# → "postgresql+asyncpg://user:pass@localhost/test" otherwise

url = worker_database_url(
    "postgresql+asyncpg://user:pass@localhost/myapp",
    default_test_db="test",
    prefix="myapp",
)
# → "postgresql+asyncpg://user:pass@localhost/myapp_gw0" under xdist
# → "postgresql+asyncpg://user:pass@localhost/myapp_test" otherwise

Manual table cleanup¶

cleanup_tables truncates all tables in a single statement and can be called directly when you need more control:

from fastapi_toolsets.pytest import cleanup_tables


@pytest.fixture(autouse=True)
async def clean(db_session):
    yield
    await cleanup_tables(session=db_session, base=Base)

API Reference