Mocking vs using a real database in SQLAlchemy tests

Mock the layer above the session, never the session itself: a MagicMock standing in for AsyncSession asserts that your code called the ORM in a particular way, which is neither what you care about nor stable, while a real database in a rolled-back transaction is fast enough that the trade was never necessary. This is the testing-strategy question under testing async SQLAlchemy applications.

Quick Answer

Before — a mocked session:

from unittest.mock import AsyncMock, MagicMock


async def test_place_order_mocked():
    session = MagicMock()
    session.execute = AsyncMock(return_value=MagicMock())
    session.commit = AsyncMock()

    await place_order(session, customer_id=1, total=Decimal("42.00"))

    session.add.assert_called_once()      # asserts a method call, not a result
    session.commit.assert_awaited_once()  # passes even if the INSERT is invalid SQL

After — a real session in a rolled-back transaction:

import pytest
from sqlalchemy import select


@pytest.mark.asyncio
async def test_place_order_persists_the_row(session):
    customer = Customer(email="ada@example.com")
    session.add(customer)
    await session.flush()

    order = await place_order(session, customer.id, Decimal("42.00"))

    stored = (await session.execute(
        select(Order).where(Order.id == order.id)
    )).scalar_one()
    assert stored.total == Decimal("42.00")
    assert stored.customer_id == customer.id

The second test is a handful of milliseconds slower and catches an entire category the first cannot: invalid SQL, a violated constraint, a Decimal that lost its scale, a foreign key pointing nowhere, a transaction that never committed.

Execution Context & Async Workflow Integration

Mocking an AsyncSession is harder than mocking a synchronous one, and the extra difficulty is a useful signal rather than an obstacle to route around. session.execute() is awaitable, so it needs an AsyncMock; its result supports scalars(), which supports all() and first() and one(); and scalar_one() raises particular exceptions in particular cases that the mock will not.

# The mock a moderately realistic query needs — and it still proves nothing
from unittest.mock import AsyncMock, MagicMock


def fake_session_returning(rows: list[Order]) -> MagicMock:
    scalars = MagicMock()
    scalars.all.return_value = rows
    scalars.first.return_value = rows[0] if rows else None
    scalars.unique.return_value = scalars

    result = MagicMock()
    result.scalars.return_value = scalars

    session = MagicMock()
    session.execute = AsyncMock(return_value=result)
    session.commit = AsyncMock()
    session.flush = AsyncMock()
    return session

That builder is twenty lines that encode an assumption about SQLAlchemy's result API. When the code under test switches from scalars().all() to scalars().unique().all() — a correctness fix required by a joined eager load — the mock returns a MagicMock rather than a list and the test fails for a reason unrelated to the change. When the code switches to session.scalars(), the mock has no such method and the test fails again.

The deeper problem is that under asyncio the failures worth catching are precisely the ones a mock removes. MissingGreenlet from a relationship access, an expired attribute after commit, a session shared between two tasks — every one of those is a real-session behaviour with no mock equivalent. A suite built on mocked sessions passes cleanly and leaves all of them to production. The four sites where that error reliably appears are catalogued in fixing GreenletSpawnError in async SQLAlchemy workflows, and a mocked session hides all four.

Resolving Warnings, Errors & Common Mistakes

Advanced: The Repository Seam That Makes Mocking Honest

There is a legitimate place for a test double, and it is not the session. It is a narrow interface between the code that decides what should happen and the code that knows how to persist it.

from __future__ import annotations

from typing import Protocol


class OrderRepository(Protocol):
    """The seam: named operations, not ORM calls."""

    async def get(self, order_id: int) -> Order | None: ...
    async def add(self, order: Order) -> None: ...
    async def outstanding_for(self, customer_id: int) -> list[Order]: ...


class SqlOrderRepository:
    def __init__(self, session: AsyncSession) -> None:
        self._session = session

    async def get(self, order_id: int) -> Order | None:
        return await self._session.get(Order, order_id)

    async def add(self, order: Order) -> None:
        self._session.add(order)

    async def outstanding_for(self, customer_id: int) -> list[Order]:
        stmt = (
            select(Order)
            .where(Order.customer_id == customer_id, Order.status == "outstanding")
            .options(selectinload(Order.items))
        )
        return list((await self._session.execute(stmt)).scalars())


class FakeOrderRepository:
    """An in-memory implementation, not a mock — it behaves rather than records."""

    def __init__(self) -> None:
        self._rows: dict[int, Order] = {}
        self._next = 1

    async def get(self, order_id: int) -> Order | None:
        return self._rows.get(order_id)

    async def add(self, order: Order) -> None:
        order.id = order.id or self._next
        self._next += 1
        self._rows[order.id] = order

    async def outstanding_for(self, customer_id: int) -> list[Order]:
        return [o for o in self._rows.values()
                if o.customer_id == customer_id and o.status == "outstanding"]

Two properties make this honest where a mocked session is not. The fake behaves — it stores and returns objects — so a test using it asserts on outcomes rather than on calls, and refactoring the service does not break it. And the interface is small enough that the fake can be kept faithful, which is the condition every test-double argument depends on and which a mocked ORM never satisfies.

The remaining obligation is a contract test: SqlOrderRepository must be exercised against a real database, and ideally both implementations should be run through the same test body.

@pytest.fixture(params=["sql", "fake"])
def repository(request, session):
    return SqlOrderRepository(session) if request.param == "sql" else FakeOrderRepository()


@pytest.mark.asyncio
async def test_outstanding_excludes_settled(repository):
    await repository.add(Order(customer_id=1, status="outstanding", total=Decimal("5")))
    await repository.add(Order(customer_id=1, status="settled", total=Decimal("7")))
    assert len(await repository.outstanding_for(1)) == 1

Run against both implementations, this asserts that the fake and the real repository agree — which is the only thing that makes tests written against the fake meaningful.

Choosing a Strategy per Test, Not per Project

The useful question is not "mock or not" but "what is this test about", and the answer usually points at exactly one strategy.

A test about a rule — how a discount is calculated, when an order may be cancelled, which state transitions are legal — should not involve persistence at all. If it currently does, the rule is entangled with storage and extracting it is the real improvement; the test getting faster is a side effect.

A test about persistence — that a query returns the right rows, that a constraint holds, that a migration produced the expected schema — needs the database. There is no substitute that proves anything, and with rollback isolation it costs single-digit milliseconds.

A test about an interaction with something outside the process — a payment provider, an email gateway, a webhook — needs a double, because the real thing is slow, non-deterministic and has effects you cannot roll back. Prefer a fake with behaviour over a mock that records calls, and cover the real integration separately in a small number of tests that run against a sandbox.

A test about wiring — that the endpoint reaches the service, which reaches the database, and returns the right status code — needs the whole stack, and should be a small number of tests rather than a large one. These are where an override of the FastAPI session dependency earns its place, as described in the parent guide.

Applied consistently, this produces a suite where mocks appear only at genuine process boundaries, the database appears wherever the database is the subject, and most tests involve neither. The proportion that ends up mocked is small — and every one of them is mocking something that genuinely cannot be run in a test.

Frequently Asked Questions

Related

• Testing Async SQLAlchemy Applications — The parent guide: fixtures, rollback isolation, and the fast-versus-faithful suite split.
• Rolling back database state between async tests — Why a real database is fast enough that mocking it buys nothing.
• Running tests against a PostgreSQL testcontainer — The behaviours only a real server can exercise.
• Fixing GreenletSpawnError in async SQLAlchemy workflows — One of the async failure modes a mocked session cannot reproduce.