Database Fixtures & Transactional Tests

A database test suite is fast and reliable, or it is neither. The difference is one design decision: whether a test leaves the database clean by undoing what it did or by deleting what it finds. Undoing is a rollback — constant time, immune to mid-test failures, and correct regardless of what the test touched. Deleting is a cleanup routine that must know every table, respect every foreign key, and run even when the test raised, which it usually does not.

Prerequisites

A real database engine, ideally started by Testcontainers so every developer and every CI job gets the same version.
SQLAlchemy >= 2.0 for join_transaction_mode, or Django 4.2+ where TestCase provides the equivalent out of the box.
pytest >= 8.0, plus pytest-xdist if the suite runs in parallel.
Migrations that can be applied from empty, since the test database is built by running them.

Core concept: isolation is a transaction, not a cleanup

Every test needs to see a known starting state and leave no trace. There are three ways to achieve that and they differ by two orders of magnitude in cost.

The rollback wins on both axes. Its only complication is what happens when the code under test commits, which is what savepoints exist for.

The complication is real, though. Application code calls session.commit() because that is what application code does, and a naive rollback fixture is defeated by the first commit. The resolution is to give the session a savepoint rather than the outer transaction: the application commits and releases its savepoint, the outer transaction is untouched, and the fixture rolls the whole thing back at teardown.

Step-by-step implementation

1. One engine, built from migrations

import pytest
from alembic import command
from alembic.config import Config
from sqlalchemy import create_engine


@pytest.fixture(scope="session")
def engine(postgres_dsn):
    # One engine and one pool for the entire run.
    engine = create_engine(postgres_dsn, pool_pre_ping=True)

    # Build the schema by running the real migrations, so an unapplyable
    # revision fails the suite immediately rather than at deploy time.
    alembic_cfg = Config("alembic.ini")
    alembic_cfg.set_main_option("sqlalchemy.url", postgres_dsn)
    command.upgrade(alembic_cfg, "head")

    yield engine
    engine.dispose()

2. A transaction per test, with a savepoint inside it

import pytest
from sqlalchemy.orm import Session


@pytest.fixture
def db_session(engine):
    connection = engine.connect()
    outer = connection.begin()

    # The session runs inside a SAVEPOINT of the outer transaction, so
    # session.commit() releases the savepoint and never touches `outer`.
    session = Session(bind=connection, join_transaction_mode="create_savepoint")
    try:
        yield session
    finally:
        session.close()
        outer.rollback()      # undoes every commit the test made
        connection.close()

3. Seed reference data once, outside the rolled-back transaction

import pytest
from sqlalchemy.orm import Session


@pytest.fixture(scope="session", autouse=True)
def reference_data(engine):
    # Currencies, countries, feature flags: immutable rows every test assumes.
    # Committed for real, once, so they survive each test's rollback.
    with Session(engine) as session:
        session.add_all([Currency(code="GBP"), Currency(code="USD")])
        session.commit()

Reference data must be committed outside the per-test transaction, or every test would have to re-create it. The rule that keeps this safe is that reference data is immutable: a test that modifies a currency row breaks every later test, so such rows should be treated as read-only by convention and, where the engine supports it, by permission.

4. Isolate parallel workers

import os

import pytest
from sqlalchemy import create_engine, text


@pytest.fixture(scope="session")
def postgres_dsn(base_dsn):
    worker = os.environ.get("PYTEST_XDIST_WORKER", "master")   # "gw0", "gw1", …
    dbname = f"test_{worker}"

    admin = create_engine(base_dsn, isolation_level="AUTOCOMMIT")
    with admin.connect() as conn:
        conn.execute(text(f'DROP DATABASE IF EXISTS "{dbname}"'))
        conn.execute(text(f'CREATE DATABASE "{dbname}"'))
    admin.dispose()

    return base_dsn.rsplit("/", 1)[0] + f"/{dbname}"

Creating one database per worker costs well under a second and eliminates an entire class of failure — two workers truncating the same table, one worker's rollback racing another's read — that otherwise presents as intermittent application bugs.

5. Django's equivalent

import pytest
from django.test import TestCase


class OrderTests(TestCase):     # not SimpleTestCase, not TransactionTestCase
    """TestCase wraps each test method in an atomic block and rolls it back."""

    def test_total_excludes_cancelled_lines(self):
        order = Order.objects.create(customer=self.customer)
        OrderLine.objects.create(order=order, amount=100, cancelled=True)
        self.assertEqual(order.total(), 0)

Django's TestCase already implements exactly the pattern above; the class you choose is the whole configuration. TransactionTestCase disables it and truncates tables instead, which is roughly fifty times slower and only necessary for tests that genuinely need committed data visible to another connection. With pytest-django, @pytest.mark.django_db gives the rollback behaviour and @pytest.mark.django_db(transaction=True) opts out of it.

Verification

The fixture is correct when a test that commits leaves nothing behind. Prove it with a pair of tests that must both pass in either order:

def test_commits_are_rolled_back(db_session):
    db_session.add(Widget(name="leaked"))
    db_session.commit()                       # a real commit, inside the savepoint
    assert db_session.query(Widget).count() == 1


def test_sees_a_clean_database(db_session):
    # If this fails, the previous test's commit escaped the fixture.
    assert db_session.query(Widget).count() == 0

Then run them in both orders — pytest -p no:randomly for the declared order and pytest -p randomly for a shuffled one — and confirm both pass. This two-test pair is worth keeping permanently; it fails loudly the day someone changes the fixture to use engine directly instead of a bound connection.

Troubleshooting

Symptom	Root cause	Fix
Rows survive between tests	Session bound to the engine, not the connection	Bind `Session(bind=connection)`
`InvalidRequestError: transaction already deassociated`	Application called `commit()` without a savepoint	Set `join_transaction_mode="create_savepoint"`
Deadlocks under `-n auto`	Workers sharing one database	One database per `PYTEST_XDIST_WORKER`
Reference data missing after the first test	Seeded inside the rolled-back transaction	Seed in a session-scoped fixture committed on its own connection
DDL in a test breaks the outer transaction	MySQL commits DDL implicitly	Mark those tests; give them a disposable database
First test slow, rest fast	Pool warm-up plus migrations, as designed	Nothing to fix; keep the engine session-scoped

What a savepoint actually does

Understanding the mechanism removes most of the guesswork. A savepoint is a named marker inside an open transaction. ROLLBACK TO SAVEPOINT s undoes everything after the marker while leaving the transaction open; RELEASE SAVEPOINT s discards the marker and keeps the work, subject to the outer transaction's eventual fate.

That last clause is the whole trick. When the ORM session "commits", SQLAlchemy issues RELEASE SAVEPOINT, not COMMIT. The application's data is now durable within the outer transaction, so subsequent queries in the same test see it — which is what makes the test realistic. But the outer transaction has never been committed, so the fixture's final ROLLBACK discards all of it.

Because the outer BEGIN was never matched by a COMMIT, the database has no record that any of it happened.

Two limits follow from the mechanism. A second connection — a background thread, a subprocess, a live server running in another process — cannot see uncommitted data, so tests that exercise cross-connection visibility need real commits and therefore real cleanup. And engines that do not support transactional DDL, notably MySQL, will implicitly commit on CREATE TABLE, silently ending the outer transaction and leaving everything before it durable.

When a rollback is not enough

Three situations defeat the transactional fixture, and each has a specific remedy that does not require abandoning the pattern for the rest of the suite.

A second connection must see the data. A live server fixture, a background worker thread, or a subprocess running the CLI all open their own connections, and an uncommitted transaction is invisible across connections by definition. These tests need real commits and therefore real cleanup. Keep them few, mark them, and give them a truncation fixture of their own:

import pytest
from sqlalchemy import text


@pytest.fixture
def committed_db(engine):
    """For the handful of tests where another connection must see the rows."""
    yield engine
    with engine.begin() as conn:
        # One statement, foreign keys handled by CASCADE, reference data excluded.
        conn.execute(text("TRUNCATE TABLE order_line, \"order\", widget CASCADE"))

The engine does not roll back DDL. MySQL commits implicitly on CREATE, ALTER and DROP, which silently ends the outer transaction and makes everything before it durable. A test that creates a temporary table therefore poisons the ones after it. The remedy is a marker plus a dedicated database recreated for those tests; the alternative — dropping transactional isolation everywhere because MySQL cannot support it in one case — costs the whole suite its speed.

The test asserts on isolation behaviour itself. Verifying that two concurrent transactions produce a serialization failure, or that a SELECT FOR UPDATE actually blocks, requires two genuine connections with genuine transactions. Those tests are worth writing and they cannot use the fixture; they belong in their own module with explicit connection management and a timeout, since a mis-written lock test is a deadlocked suite.

The exceptions are real but rare. Marking them keeps the cost visible and stops the truncation fixture spreading to tests that never needed it.

Layering fixtures so the suite stays readable

A mature database suite ends up with four layers, and naming them explicitly prevents the single-fixture sprawl that makes later tests hard to read.

At the bottom is the engine, session-scoped, built from migrations, shared by everything. Above it the session, function-scoped, transactional, the only thing tests are given directly. Above that, factories — not fixtures at all, but callables a test invokes to build exactly the rows it needs. And at the top, a small number of scenario fixtures that compose factories for a situation used by many tests, such as "a customer with an open subscription".

import pytest


@pytest.fixture
def subscribed_customer(db_session):
    """A composed scenario — one obvious meaning, used by many tests."""
    customer = CustomerFactory(country="GB")
    SubscriptionFactory(customer=customer, status="active")
    db_session.flush()          # assign primary keys without committing
    return customer

flush() rather than commit() is the detail that keeps scenario fixtures compatible with the rollback: it sends the INSERTs so identifiers are populated and later queries see the rows, without ending any transaction.

The discipline that keeps this layering useful is to resist adding fields to a scenario fixture for one test's benefit. When a test needs a customer with a lapsed subscription, it builds one from the factories rather than adding a status parameter to subscribed_customer; the parametrised scenario fixture that tries to serve every case is how a suite acquires a fixture nobody can safely change. Each layer should be replaceable without touching the ones above it, which is the same separation of construction from configuration argued for in wiring test doubles through a factory function.

Keeping the fixture honest as the suite grows

Two habits prevent this design from eroding over the following year.

The first is a guard against direct engine use. Any fixture or helper that takes engine and opens its own connection bypasses the outer transaction, and the resulting leakage appears as a mysteriously order-dependent test somewhere else entirely. Making engine private to the conftest.py that defines db_session, and exposing only the session, removes the temptation.

The second is a periodic check that the migrations and the models still agree. Because the test database is built from migrations, a model change with no matching revision produces a failure at the first query rather than a silent divergence — but only if nobody has quietly switched the fixture back to Base.metadata.create_all() for speed. A test that autogenerates a revision and asserts it is empty makes the invariant explicit, and the full setup is in testing Alembic migrations in CI.

Finally, resist the pull toward a shared "kitchen sink" fixture that creates a customer, an order and three products for every test. It makes the suite slower, couples unrelated tests to one data shape, and hides which values a given test actually depends on. Explicit per-test construction through factories costs a line or two and makes each test readable on its own — the same argument made for fixture design generally in taming autouse fixtures in large suites.

Measuring what the change bought

Restructuring a database suite is worth measuring, both to confirm the gain and to catch the regression when someone reintroduces a per-test truncation.

# Before and after, same machine, same database, same tests.
pytest tests/db -q --durations=0 | tail -n 40

The number to watch is the ratio of setup time to call time. A healthy transactional suite spends almost nothing in setup — the connection is already open, the transaction is a single BEGIN — so the durations report is dominated by the call phase, where the actual queries run. A suite still doing per-test cleanup shows the opposite: setup and teardown together outweighing the test bodies by three or four to one.

Two other numbers are worth recording in the same pass. Total wall clock for the database subset, since that is what people feel; and peak database connection count during the run, which is what breaks first when the suite is parallelised. SELECT count(*) FROM pg_stat_activity sampled during a run gives the second, and if it is anywhere near the server's max_connections divided by the worker count, the pool is sized too generously for test use.

Keep the figures in the repository next to the fixture, as a comment or a short note. The value is not the benchmark itself but the record of intent: the next engineer who wonders why the engine is session-scoped and the session is not finds the answer rather than re-deriving it, and the one who is tempted to add a TRUNCATE to "make sure things are clean" sees what that costs before doing it. That is the same reasoning behind recording enforcement thresholds rather than leaving them to habit, as in coverage measurement and enforcement.

Frequently Asked Questions

Why does my rollback fixture stop working when the code calls commit()? Because the application's commit ends the transaction your fixture intended to roll back. The fix is to run the session inside a savepoint: SQLAlchemy 2.0's join_transaction_mode="create_savepoint" does this directly, and Django's TestCase wraps each test in an atomic block for the same reason. The application then commits a savepoint rather than the outer transaction.

Should the test database be built from migrations or from metadata? From migrations. Building with create_all() means the migrations are never executed until a deploy, so drift between models and migration history goes undetected. Running migrations once per session costs a few seconds and turns every suite run into a smoke test of the migration chain.

How do I isolate parallel xdist workers from each other? Give each worker its own database or schema, named from PYTEST_XDIST_WORKER. Sharing one database across workers means one worker's rollback races another's read, producing failures that look like application bugs. Creating eight small databases at session start costs a second and removes the whole class of problem.

Is SQLite an acceptable stand-in for Postgres in tests? Only for code that touches no database-specific behaviour, which in practice is almost nothing. SQLite differs in type affinity, constraint enforcement timing, concurrency, JSON operators and window-function support, so a suite that passes on SQLite and deploys to Postgres is testing a different program. Run the real engine in a container.

What about tests that need DDL, which cannot be rolled back everywhere? Postgres is transactional for DDL, so CREATE TABLE inside a test rolls back cleanly. MySQL is not: DDL commits implicitly and destroys the surrounding transaction. For MySQL, mark those tests and give them a dedicated database that is recreated between them, rather than weakening isolation for the whole suite.

Implement the rollback fixture end to end in rolling back every test with nested transactions.
Keep models and schema in step with testing Alembic migrations in CI.
Start the engine the fixture connects to using starting Postgres with testcontainers-python.
Make each test's preconditions explicit with test data factories and builders.
For async engines, match the fixture's loop to the pool's using pytest-asyncio in depth.

← Back to Integration, Database & Service Testing