tests/conftest.py (Generic Example)

This page explains what belongs in tests/conftest.py, and provides a copy/paste template you can adapt.

What conftest.py is for

Use tests/conftest.py for shared, cross-module fixtures:

• client: an httpx.AsyncClient wired to your FastAPI ASGI app (no real network).
• db_session: an AsyncSession you can inject into CRUD and API tests.
• Auth helpers: auth_headers (reference backend builds this via a real login request).
• IDs / context: stable IDs like tenant_id for convenience in tests.

Keep it small and predictable:

• Put module-specific factories/fixtures in tests/{module}/fixtures.py (or tests/{module}/conftest.py).
• Don’t hide business logic in conftest.py; keep it focused on wiring.

Canonical fixture template (async DB + FastAPI client)

This template assumes:

• You’re using pytest + pytest-asyncio with asyncio_mode = strict (see docs/backend/04_testing/overview.md).
• Your FastAPI app exposes an ASGI app object.
• Migrations/schema setup are handled outside pytest (agents should not run migrations as part of test execution).

import pytest
import pytest_asyncio
from sqlalchemy.ext.asyncio import AsyncSession
from httpx import AsyncClient, ASGITransport

# Import your FastAPI app.
# Common patterns:
# - from main import app
# - from app.main import app
from main import app  # adjust to your project

# Import your database module that exposes the async engine + session factory.
# The reference backend ensures the engine is created before opening sessions.
import database.database as db_module  # adjust to your project


@pytest.fixture
def tenant_id() -> str:
    """Stable ID used across tests."""
    return "test-tenant-id"


@pytest_asyncio.fixture
async def auth_headers(client: AsyncClient, create_and_verify_user):
    """
    Authenticated request headers built via a real login flow.

    The reference backend performs a login request to obtain an access token and returns:
    - Authorization header
    - authenticated user id (optional convenience)
    """
    user = create_and_verify_user

    login_response = await client.post(
        "<LOGIN_ENDPOINT_PATH>",  # reference backend uses a form-encoded OAuth2-style login endpoint
        data={"username": user["email"], "password": user["password"]},
        headers={"Content-Type": "application/x-www-form-urlencoded"},
    )
    assert login_response.status_code == 200, login_response.text

    access_token = login_response.json()["access_token"]
    user_id = login_response.json()["content"]["id"]

    return {"Authorization": f"Bearer {access_token}", "user_id": user_id}


@pytest_asyncio.fixture(scope="function")
async def db_session() -> AsyncSession:
    """
    Per-test AsyncSession.

    Preferred isolation strategies:
    - Transaction rollback per test (best, if your DB layer supports it cleanly)
    - Explicit cleanup (acceptable, but more work)

    This example uses a simple session-per-test; pair it with cleanup or a rollback
    strategy in your DB layer.
    """
    db_module.get_async_engine()  # ensure engine + session factory exist
    async with db_module.AsyncSessionLocal() as session:
        try:
            yield session
        finally:
            await session.close()


@pytest_asyncio.fixture(scope="function")
async def client() -> AsyncClient:
    """
    Async client bound to the ASGI app (no real network).

    The reference backend uses `ASGITransport(app=app)` to bind the client to the app
    without making real network calls.
    """
    try:
        transport = ASGITransport(app=app)
        async with AsyncClient(transport=transport, base_url="http://test") as c:
            yield c
    finally:
        # The reference backend disposes engines after each test to prevent event loop conflicts
        engine = db_module.get_async_engine()
        if engine:
            await engine.dispose()
        worker_engine = db_module.get_worker_async_engine()
        if worker_engine:
            await worker_engine.dispose()


@pytest_asyncio.fixture(scope="function", autouse=True)
async def cancel_pending_async_tasks():
    """
    Reliability helper: cancel and await any pending tasks after each test.

    The reference backend uses this to prevent async task leakage between tests.
    """
    import asyncio
    from contextlib import suppress

    try:
        yield
    finally:
        loop = asyncio.get_running_loop()
        current = asyncio.current_task()
        pending = [task for task in asyncio.all_tasks(loop) if task is not current and not task.done()]
        for task in pending:
            task.cancel()
        for task in pending:
            with suppress(asyncio.CancelledError):
                await task

Where to put module fixtures

• Root-level shared wiring stays in tests/conftest.py.
• Module-specific creation helpers go in tests/{module}/fixtures.py (recommended) and are injected by importing them in tests or re-exporting them via tests/{module}/conftest.py.