Skip to content

pytest

Here's the reference for all testing utilities and pytest fixtures.

You can import them directly from fastapi_toolsets.pytest:

from fastapi_toolsets.pytest import (
    register_fixtures,
    create_async_client,
    create_db_session,
    worker_database_url,
    create_worker_database,
    cleanup_tables,
)

fastapi_toolsets.pytest.plugin.register_fixtures(registry, namespace, *, prefix='fixture_', session_fixture='db_session', strategy=LoadStrategy.MERGE)

Register pytest fixtures from a FixtureRegistry.

Automatically creates pytest fixtures for each fixture in the registry. Dependencies are resolved via pytest fixture dependencies.

Parameters:

Name Type Description Default
registry FixtureRegistry

The FixtureRegistry containing fixtures

 required
namespace dict[str, Any]

The module's globals() dict to add fixtures to

 required
prefix str

Prefix for generated fixture names (default: "fixture_")

 'fixture_'
session_fixture str

Name of the db session fixture (default: "db_session")

 'db_session'
strategy LoadStrategy

Loading strategy for fixtures (default: MERGE)

 MERGE

Returns:

Type Description
list[str]

List of created fixture names
Example 
# conftest.py
from app.fixtures import fixtures
from fastapi_toolsets.pytest_plugin import register_fixtures

register_fixtures(fixtures, globals())

# Creates fixtures like:
# - fixture_roles
# - fixture_users (depends on fixture_roles if users depends on roles)
# - fixture_posts (depends on fixture_users if posts depends on users)

fastapi_toolsets.pytest.utils.create_async_client(app, base_url='http://test', dependency_overrides=None) async

Create an async httpx client for testing FastAPI applications.

Parameters:

Name Type Description Default
app Any

FastAPI application instance.

 required
base_url str

Base URL for requests. Defaults to "http://test".

 'http://test'
dependency_overrides dict[Callable[..., Any], Callable[..., Any]] | None

Optional mapping of original dependencies to their test replacements. Applied via app.dependency_overrides before yielding and cleaned up after.

 None

Yields:

Type Description
AsyncGenerator[AsyncClient, None]

An AsyncClient configured for the app.
Example 
from fastapi import FastAPI
from fastapi_toolsets.pytest import create_async_client

app = FastAPI()

@pytest.fixture
async def client():
    async with create_async_client(app) as c:
        yield c

async def test_endpoint(client: AsyncClient):
    response = await client.get("/health")
    assert response.status_code == 200
Example with dependency overrides 
from fastapi_toolsets.pytest import create_async_client, create_db_session
from app.db import get_db

@pytest.fixture
async def db_session():
    async with create_db_session(DATABASE_URL, Base, cleanup=True) as session:
        yield session

@pytest.fixture
async def client(db_session):
    async def override():
        yield db_session

    async with create_async_client(
        app, dependency_overrides={get_db: override}
    ) as c:
        yield c

fastapi_toolsets.pytest.utils.create_db_session(database_url, base, *, echo=False, expire_on_commit=False, drop_tables=True, cleanup=False) async

Create a database session for testing.

Creates tables before yielding the session and optionally drops them after. Each call creates a fresh engine and session for test isolation.

Parameters:

Name Type Description Default
database_url str

Database connection URL (e.g., "postgresql+asyncpg://...").

 required
base type[DeclarativeBase]

SQLAlchemy DeclarativeBase class containing model metadata.

 required
echo bool

Enable SQLAlchemy query logging. Defaults to False.

 False
expire_on_commit bool

Expire objects after commit. Defaults to False.

 False
drop_tables bool

Drop tables after test. Defaults to True.

 True
cleanup bool

Truncate all tables after test using :func:cleanup_tables. Defaults to False.

 False

Yields:

Type Description
AsyncGenerator[AsyncSession, None]

An AsyncSession ready for database operations.
Example 
from fastapi_toolsets.pytest import create_db_session
from app.models import Base

DATABASE_URL = "postgresql+asyncpg://user:pass@localhost/test_db"

@pytest.fixture
async def db_session():
    async with create_db_session(
        DATABASE_URL, Base, cleanup=True
    ) as session:
        yield session

async def test_create_user(db_session: AsyncSession):
    user = User(name="test")
    db_session.add(user)
    await db_session.commit()

fastapi_toolsets.pytest.utils.worker_database_url(database_url, default_test_db)

Derive a per-worker database URL for pytest-xdist parallel runs.

Appends _{worker_name} to the database name so each xdist worker operates on its own database. When not running under xdist, _{default_test_db} is appended instead.

The worker name is read from the PYTEST_XDIST_WORKER environment variable (set automatically by xdist in each worker process).

Parameters:

Name Type Description Default
database_url str

Original database connection URL.

 required
default_test_db str

Suffix appended to the database name when PYTEST_XDIST_WORKER is not set.

 required

Returns:

Type Description
str

A database URL with a worker- or default-specific database name.
Example 
# With PYTEST_XDIST_WORKER="gw0":
url = worker_database_url(
    "postgresql+asyncpg://user:pass@localhost/test_db",
    default_test_db="test",
)
# "postgresql+asyncpg://user:pass@localhost/test_db_gw0"

# Without PYTEST_XDIST_WORKER:
url = worker_database_url(
    "postgresql+asyncpg://user:pass@localhost/test_db",
    default_test_db="test",
)
# "postgresql+asyncpg://user:pass@localhost/test_db_test"

fastapi_toolsets.pytest.utils.create_worker_database(database_url, default_test_db='test_db') async

Create and drop a per-worker database for pytest-xdist isolation.

Intended for use as a session-scoped fixture. Connects to the server using the original database_url (with AUTOCOMMIT isolation for DDL), creates a dedicated database for the worker, and yields the worker-specific URL. On cleanup the worker database is dropped.

When running under xdist the database name is suffixed with the worker name (e.g. _gw0). Otherwise it is suffixed with default_test_db.

Parameters:

Name Type Description Default
database_url str

Original database connection URL.

 required
default_test_db str

Suffix appended to the database name when PYTEST_XDIST_WORKER is not set. Defaults to "test_db".

 'test_db'

Yields:

Type Description
AsyncGenerator[str, None]

The worker-specific database URL.
Example 
from fastapi_toolsets.pytest import (
    create_worker_database, create_db_session,
)

DATABASE_URL = "postgresql+asyncpg://postgres:postgres@localhost/test_db"

@pytest.fixture(scope="session")
async def worker_db_url():
    async with create_worker_database(DATABASE_URL) as url:
        yield url

@pytest.fixture
async def db_session(worker_db_url):
    async with create_db_session(
        worker_db_url, Base, cleanup=True
    ) as session:
        yield session

fastapi_toolsets.pytest.utils.cleanup_tables(session, base) async

Truncate all tables for fast between-test cleanup.

Executes a single TRUNCATE … RESTART IDENTITY CASCADE statement across every table in base's metadata, which is significantly faster than dropping and re-creating tables between tests.

This is a no-op when the metadata contains no tables.

Parameters:

Name Type Description Default
session AsyncSession

An active async database session.

 required
base type[DeclarativeBase]

SQLAlchemy DeclarativeBase class containing model metadata.

 required
Example 
@pytest.fixture
async def db_session(worker_db_url):
    async with create_db_session(worker_db_url, Base) as session:
        yield session
        await cleanup_tables(session, Base)