SQLAlchemy 2.0 Textbook Ch. 11 — Testing

Chapter 11: testing SQLAlchemy without mocks. Real DBs, per-test transactions, factories, async fixtures, CI integration.

Why real DBs

Mocking SQLAlchemy is tempting — and produces false confidence. Real DBs catch:

Constraint violations.
Type coercion.
Migration / model drift.
Index usage.
Transaction semantics.

Mocks pass; production fails. Always test against the real DB.

testcontainers

import pytest
from testcontainers.postgres import PostgresContainer

@pytest.fixture(scope="session")
def postgres():
    with PostgresContainer("postgres:17") as pg:
        yield pg

Spins up a Postgres in Docker for the test session. Clean; isolated; fast (~3s startup).

For MySQL / SQLite: similar containers exist.

Engine fixture

from sqlalchemy.ext.asyncio import create_async_engine

@pytest.fixture(scope="session")
async def engine(postgres):
    url = postgres.get_connection_url().replace("psycopg2", "asyncpg")
    engine = create_async_engine(url, echo=False)
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    yield engine
    await engine.dispose()

Create schema once per session.

Per-test transactions

The fast pattern:

@pytest.fixture
async def db_session(engine):
    connection = await engine.connect()
    transaction = await connection.begin()
    async with AsyncSession(bind=connection, expire_on_commit=False) as session:
        nested = await session.begin_nested()
        
        @event.listens_for(session.sync_session, "after_transaction_end")
        def end_savepoint(session, transaction):
            nonlocal nested
            if not nested.is_active:
                nested = session.begin_nested()
        
        yield session
    
    await transaction.rollback()
    await connection.close()

Each test starts a savepoint. After the test: rollback. DB returns to clean state.

For simpler tests:

@pytest.fixture
async def db_session(engine):
    async with AsyncSession(engine, expire_on_commit=False) as session:
        yield session
        await session.rollback()

Less robust if test commits something; truncate tables between tests.

Factory boy

import factory

class UserFactory(factory.Factory):
    class Meta:
        model = User
    
    email = factory.Sequence(lambda n: f"user{n}@example.com")
    full_name = factory.Faker("name")
    is_active = True

class PostFactory(factory.Factory):
    class Meta:
        model = Post
    
    title = factory.Faker("sentence")
    body = factory.Faker("text")
    author = factory.SubFactory(UserFactory)

async def test_user_create(db_session):
    user = UserFactory.build()
    db_session.add(user)
    await db_session.commit()
    
    fetched = await db_session.get(User, user.id)
    assert fetched.email == user.email

build() creates; add() to insert. Fixtures stay short.

Pytest-asyncio / pytest-anyio

import pytest

@pytest.mark.asyncio
async def test_x(db_session):
    ...

Or with anyio:

@pytest.fixture(scope="session")
def anyio_backend():
    return "asyncio"

@pytest.mark.anyio
async def test_x(db_session):
    ...

Both work. anyio is more common in modern FastAPI projects.

Testing relationships

async def test_posts_loaded(db_session):
    user = UserFactory.build()
    user.posts = [PostFactory.build() for _ in range(3)]
    db_session.add(user)
    await db_session.commit()
    
    stmt = select(User).options(selectinload(User.posts)).where(User.id == user.id)
    result = await db_session.scalar(stmt)
    assert len(result.posts) == 3

Testing constraints

async def test_email_unique(db_session):
    db_session.add(UserFactory.build(email="[email protected]"))
    await db_session.commit()
    
    db_session.add(UserFactory.build(email="[email protected]"))
    with pytest.raises(IntegrityError):
        await db_session.commit()

Verifies the DB constraint actually fires.

Testing migrations

def test_no_drift(engine):
    from alembic.autogenerate import compare_metadata
    from alembic.migration import MigrationContext
    
    with engine.sync_engine.connect() as conn:
        ctx = MigrationContext.configure(conn)
        diff = compare_metadata(ctx, Base.metadata)
        assert diff == [], f"Drift: {diff}"

Catches forgotten migrations.

For migration up/down:

def test_migrations(postgres):
    cfg = Config("alembic.ini")
    cfg.set_main_option("sqlalchemy.url", postgres.get_connection_url())
    
    command.upgrade(cfg, "head")
    command.downgrade(cfg, "base")
    command.upgrade(cfg, "head")

Performance tests

For critical queries, test query count:

async def test_no_n_plus_1(db_session):
    [UserFactory.build() for _ in range(10)]
    await db_session.commit()
    
    queries = []
    @event.listens_for(db_session.sync_session, "do_orm_execute")
    def collect(orm_execute_state):
        queries.append(orm_execute_state.statement)
    
    stmt = select(User).options(selectinload(User.posts))
    users = (await db_session.execute(stmt)).scalars().all()
    for u in users:
        _ = u.posts  # access
    
    assert len(queries) == 2  # users + posts (no N+1)

Tests that selectinload is being used.

Fixture composition

@pytest.fixture
async def admin_user(db_session):
    user = UserFactory.build(role="admin")
    db_session.add(user)
    await db_session.commit()
    return user

@pytest.fixture
async def auth_client(client, admin_user):
    token = make_token(admin_user.id)
    client.headers["Authorization"] = f"Bearer {token}"
    return client

async def test_admin_endpoint(auth_client):
    r = await auth_client.get("/admin/users")
    assert r.status_code == 200

Compose fixtures up the stack.

CI integration

services:
  postgres:
    image: postgres:17
    env:
      POSTGRES_PASSWORD: test
    ports: ["5432:5432"]
    options: >-
      --health-cmd pg_isready
      --health-interval 10s

Or testcontainers does this automatically.

Cross-DB testing

For DB-agnostic libraries, parametrize:

@pytest.fixture(params=["postgres", "mysql", "sqlite"])
def engine(request):
    if request.param == "postgres":
        with PostgresContainer() as pg:
            yield create_async_engine(pg.get_connection_url())
    elif ...

Run tests against each. Slower; catches dialect bugs.

Common mistakes

1. Mocking the session

session = AsyncMock()
session.get.return_value = User(...)

Tests pass; SQL never runs; constraint bugs ship. Use real DB.

2. Shared state across tests

Test A inserts; test B sees row; test order matters. Use per-test transactions.

3. Slow test setup

Spin up Postgres for each test → minutes. Session-scoped fixture; per-test transaction.

4. Not testing migrations

Migration breaks; CI passes; deploy fails. Run migrations in CI.

5. SQLite for tests when prod is Postgres

Different dialects; different behavior. Test against the prod DB.

What’s next

Chapter 12: Production patterns.

Why real DBs#

testcontainers#

Engine fixture#

Per-test transactions#

Factory boy#

Pytest-asyncio / pytest-anyio#

Testing relationships#

Testing constraints#

Testing migrations#

Performance tests#

Fixture composition#

CI integration#

Cross-DB testing#

Common mistakes#

1. Mocking the session#

2. Shared state across tests#

3. Slow test setup#

4. Not testing migrations#

5. SQLite for tests when prod is Postgres#

What’s next#

Read this next#