· Patryk Komow · 10 min read

Testing FastAPI with pytest, async SQLAlchemy and a real PostgreSQL

An async FastAPI app with async SQLAlchemy is easy to test once three things are right: which event loop the tests run on, which database they talk to, and how the state gets cleaned up. Get those wrong and you get the famous "attached to a different loop" errors. Get them right and more than 170 tests run against real PostgreSQL in under half a minute.

Contents

  1. Test against PostgreSQL, not SQLite
  2. Configure the app before importing it
  3. Event loops: one for the whole session
  4. The schema: real migrations
  5. Cleaning up between tests
  6. The client and the database fixtures
  7. Helpers that act like a browser
  8. Testing whole flows
  9. Faking external services

Test against PostgreSQL, not SQLite

An in-memory SQLite database is tempting, and it will lie to you. INSERT ... ON CONFLICT DO NOTHING from the PostgreSQL dialect, timezone-aware timestamps, TRUNCATE ... CASCADE, server defaults like now(): all of it behaves differently or doesn't exist. Tests that pass on SQLite and fail in production are worse than no tests.

With Docker Compose you already run PostgreSQL. Give the tests their own database on the same server, created the first time the volume is initialised:

-- docker/postgres/init/01-create-test-db.sql
-- Runs once, when the database volume is first created.
-- The test suite uses its own database so it never touches your dev data.
CREATE DATABASE app_test;
# docker-compose.yml, in the db service
    volumes:
      - pgdata:/var/lib/postgresql/data
      - ./docker/postgres/init:/docker-entrypoint-initdb.d:ro

Scripts in /docker-entrypoint-initdb.d only run when the data directory is empty. If your volume existed before you added the file, recreate it with docker compose down -v (which deletes local data) or run the CREATE DATABASE by hand once.

Configure the app before importing it

Settings are read once and cached, and the engine is created when app.db is imported. So the test configuration has to be in the environment before anything imports the app. The top of conftest.py does exactly that:

# tests/conftest.py
import os
import subprocess

# Point the app at the test database before anything imports app.config.
os.environ["APP_ENV"] = "test"
os.environ["EMAIL_BACKEND"] = "memory"
os.environ["SECRET_KEY"] = "test-secret-key-that-is-long-enough-for-hs256"
os.environ["STRIPE_SECRET_KEY"] = "sk_test_fake"
os.environ["STRIPE_WEBHOOK_SECRET"] = "whsec_test_secret"
os.environ["EU_MODE"] = "false"
if test_url := os.environ.get("TEST_DATABASE_URL"):
    os.environ["DATABASE_URL"] = test_url

import pytest  # noqa: E402
from httpx import ASGITransport, AsyncClient  # noqa: E402
from sqlalchemy import text  # noqa: E402

import app.models  # noqa: E402, F401  (all tables, for the cleanup below)
from app.db import Base, SessionLocal, engine  # noqa: E402
from app.main import app  # noqa: E402

The noqa: E402 comments tell the linter that the late imports are deliberate. TEST_DATABASE_URL lives in .env next to DATABASE_URL and points at app_test. Setting every value the tests rely on here, rather than trusting whatever is in a developer's .env, means the suite behaves the same on every machine and in CI.

Note the memory email backend: instead of sending, it appends each message to a list the tests can read. More on that below.

Event loops: one for the whole session

Most async test pain comes from event loops. pytest-asyncio gives each test its own loop by default, while a pooled asyncpg connection belongs to the loop that opened it. Reuse that connection in the next test and you get RuntimeError: ... attached to a different loop or a Future that never completes.

Two settings remove the problem. Run every test and fixture on one shared loop:

# pyproject.toml
[tool.pytest.ini_options]
asyncio_mode = "auto"
asyncio_default_fixture_loop_scope = "session"
# Tests and fixtures share one event loop, so DB connections never cross loops.
asyncio_default_test_loop_scope = "session"
testpaths = ["tests"]
pythonpath = ["."]

And don't pool connections in tests at all:

# app/db.py
engine = create_async_engine(
    settings.database_url,
    poolclass=NullPool if settings.app_env == "test" else None,
)

With NullPool, every session opens a fresh connection and closes it when done, so no connection outlives the code that made it. It's slightly slower per query and much faster to debug. asyncio_mode = "auto" treats every async def test and fixture as asyncio without a decorator on each one. (asyncio_default_test_loop_scope needs a recent pytest-asyncio; check your version's docs if the option isn't recognised.)

The schema: real migrations

@pytest.fixture(scope="session", autouse=True)
def migrated_database():
    """Builds the test database with the real migrations (down, then up), so the
    migrations are tested too."""
    for command in (["alembic", "downgrade", "base"], ["alembic", "upgrade", "head"]):
        subprocess.run(command, check=True, capture_output=True)

Once per test run, the database is torn down to nothing and rebuilt with your Alembic migrations, including every downgrade() on the way down. Base.metadata.create_all() would be faster by a second and would hide the one bug that only shows up in production: a model column no migration creates. Alembic runs in a subprocess because its async env.py calls asyncio.run(), which can't run inside pytest's already-running loop.

Cleaning up between tests

The popular pattern is "wrap each test in a transaction and roll it back". It breaks as soon as the code under test calls commit(), which real service code does. Truncating every table after each test is simpler and works whatever the code does:

@pytest.fixture(autouse=True)
async def clean_state():
    yield
    tables = ", ".join(table.name for table in Base.metadata.sorted_tables)
    async with engine.begin() as conn:
        await conn.execute(text(f"TRUNCATE {tables} RESTART IDENTITY CASCADE"))
    email.outbox.clear()
    login_limiter.reset()
    email_limiter.reset()

The client and the database fixtures

@pytest.fixture
async def client():
    async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as c:
        yield c


@pytest.fixture
async def db():
    async with SessionLocal() as session:
        yield session


@pytest.fixture
def make_user(db):
    async def make(email="ada@example.com", password=PASSWORD, verified=True, **fields):
        user = await service.create_user(db, email, password, locale="en")
        if verified:
            await service.mark_email_verified(db, user)
        for name, value in fields.items():
            setattr(user, name, value)
        await db.commit()
        return user

    return make

httpx.AsyncClient with ASGITransport calls the app in-process, on the same event loop as the test, with no server and no network. It keeps cookies between requests like a browser, so a login in one request carries over to the next. It doesn't follow redirects by default, which is what you want: tests can assert on the 303 and its location header.

One thing ASGITransport doesn't do is run the app's lifespan (startup and shutdown). If your app opens resources there, open them in a fixture or use a lifespan manager in tests.

make_user is a factory fixture: it returns a function, so one test can create several users with different fields: await make_user("bob@example.com", verified=False, is_admin=True). It goes through the real service code, so the password is hashed exactly as in production and login works with it.

Helpers that act like a browser

If your forms are protected by CSRF tokens, a test can't just POST to them. Instead of switching CSRF off in tests (and never testing it), read the token from a page the way a browser would:

# tests/helpers.py
PASSWORD = "correct horse battery"


async def get_csrf(client) -> str:
    """Loads a page and returns its CSRF token (every page carries it for HTMX)."""
    response = await client.get("/")
    return re.search(r'"X-CSRF-Token": "([^"]+)"', response.text).group(1)


async def post_form(client, url: str, data: dict | None = None, **kwargs):
    """POSTs a form with a valid CSRF token, like a browser would."""
    data = {**(data or {}), "csrf_token": await get_csrf(client)}
    return await client.post(url, data=data, **kwargs)


async def login(client, email="ada@example.com", password=PASSWORD):
    return await post_form(client, "/login", {"email": email, "password": password})


def last_email_link() -> str:
    return re.search(r"https?://\S+", email.outbox[-1].text).group(0)

last_email_link is where the memory email backend pays off. Verification and reset flows send a link by email; the test reads it from the outbox and follows it, exactly like a user clicking it in their inbox.

Testing whole flows

With those pieces, a test reads like the user story it checks. The whole password reset, from the form to logging in with the new password:

async def test_reset_flow(client, make_user):
    await make_user()

    response = await request_reset(client, "ADA@example.com")
    assert response.status_code == 303
    assert email.outbox[-1].to == "ada@example.com"
    link = last_email_link()

    assert (await client.get(link)).status_code == 200
    response = await reset_with_link(client, link)
    assert response.status_code == 303
    assert response.headers["location"] == "/account"

    await post_form(client, "/logout")
    assert (await login(client)).status_code == 400  # old password
    assert (await login(client, password=NEW_PASSWORD)).status_code == 303

Note the uppercase ADA@example.com: a small check that emails are normalised, inside a test about something else. Two clients make "two devices" tests easy, for example that a password reset signs the user out everywhere else:

async def test_reset_signs_out_other_sessions(client, make_user):
    await make_user()
    async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as laptop:
        await login(laptop)
        assert (await laptop.get("/account")).status_code == 200

        await request_reset(client)
        await reset_with_link(client, last_email_link())

        assert (await laptop.get("/account")).status_code == 303

Test the security properties as explicitly as the happy path: the reset link works only once, an unknown email gets the same answer as a known one and no email is sent, a short password is rejected. Each is a five-line test, and each is a bug you'd otherwise find in production.

Faking external services

Tests shouldn't call Stripe, or any paid API. The trick that keeps fakes simple is to route every call to the service through one module, and fake that module's functions, nothing deeper:

# tests/fake_stripe.py (the start of it)
class FakeStripe:
    def __init__(self):
        self._ids = itertools.count(1)
        self.customers: list[dict] = []
        self.checkout_params: list[dict] = []
        self.subscriptions: dict[str, dict] = {}

    def install(self, monkeypatch) -> None:
        for name in ("create_customer", "create_checkout_session", "retrieve_subscription"):
            monkeypatch.setattr(stripe_gateway, name, getattr(self, name))

    async def create_customer(self, email: str, user_id: int) -> str:
        customer_id = f"cus_{next(self._ids)}"
        self.customers.append({"id": customer_id, "email": email, "user_id": user_id})
        return customer_id
# tests/conftest.py
@pytest.fixture(autouse=True)
def stripe_fake(monkeypatch) -> FakeStripe:
    """Replaces every function in app.billing.stripe_gateway that talks to Stripe."""
    fake = FakeStripe()
    fake.install(monkeypatch)
    return fake

The fake is autouse, so no test can reach the real API by accident, and it records what the app sent (checkout_params), so tests can assert on it. Because only the gateway is faked, the webhook signature check still runs for real. Tests sign their events the way Stripe does:

def signed_webhook(event: dict, secret: str = "whsec_test_secret") -> tuple[bytes, dict]:
    """Returns the body and headers Stripe would send, signed the way Stripe signs them."""
    payload = json.dumps(event).encode()
    timestamp = int(time.time())
    signature = hmac.new(
        secret.encode(), f"{timestamp}.".encode() + payload, hashlib.sha256
    ).hexdigest()
    headers = {
        "Stripe-Signature": f"t={timestamp},v1={signature}",
        "Content-Type": "application/json",
    }
    return payload, headers
async def test_rejects_bad_signature(client):
    payload, headers = signed_webhook(make_event("invoice.paid", {}), secret="whsec_wrong")

    response = await client.post("/stripe/webhook", content=payload, headers=headers)

    assert response.status_code == 400


async def test_duplicate_events_are_applied_once(client, make_user, db, stripe_fake):
    _, subscription = await subscribed_user(make_user, db, stripe_fake)
    event = make_event("customer.subscription.deleted", subscription, event_id="evt_same")
    subscription["status"] = "canceled"

    await send(client, event)
    await send(client, event)

    assert len(email.outbox) == 1
    assert await db.scalar(select(func.count()).select_from(StripeEvent)) == 1

Duplicate deliveries, events out of order, a webhook racing the success page: the cases that are nearly impossible to reproduce by hand against Stripe each become a test that runs in milliseconds.

One small gotcha when a test reads rows the app changed: the test's db session may still hold the old objects. Call db.expire_all() before re-reading, or query in a new session.

Checklist

This test setup comes from getpylaunch, a FastAPI SaaS starter I sell, which ships with these fixtures and more than 170 tests.