Skip to content
You're not viewing the latest version. Click here to go to latest.

# `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.

## `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.

Derives a worker-specific database URL using :func:`worker_database_url`, then delegates to :func:`~fastapi_toolsets.db.create_database` to create and drop it. Intended for use as a session-scoped fixture.

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 (used as the server connection and as the base for the worker database name). | 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