Hands-onIntermediate 40 min· Automated tests · 10 of 19

Integration tests with a real database

Write integration tests for the coffee shop API that run against a real database, with a fresh database for every test. Learn dependency overrides, checking stored data, failure-path tests and how to run the same suite on Postgres.

In this tutorial you will write integration tests for the coffee shop API where the code, the web framework and a real database all run together. Unit tests with mocks can pass while the real SQL is wrong; integration tests find those bugs.

You need: Python 3.10 or newer, pytest basics, and a rough idea of what a database table is. You will use SQLite (it needs no installation) and see at the end how to run the same tests on Postgres. Related lessons: *Mocks, stubs and test doubles* (the opposite choice) and the database lessons in the Backend Development track if you want to go deeper on the app itself.

What an integration test is

A unit test checks one function with everything else replaced. An integration test checks that several real parts work together: here the API route, the validation, the SQLAlchemy models and the database. It is slower than a unit test (milliseconds to a few seconds) but it catches a different class of bugs:

What each kind of test can find

Bug	Unit test with a mock	Integration test
Typo in a column name	Passes (the mock does not care)	Fails
Forgot `db.commit()` so data is lost	Passes	Fails when you read the data back
Wrong filter in a query	Passes	Fails
Failed request leaves half-saved rows	Passes	Fails (you can count rows)
Two orders share data between tests	Not visible	Visible: you will design against it

Remember the test pyramid from *Why we test*: many unit tests, fewer integration tests, even fewer end-to-end tests. You do not need an integration test for every function, only for the places where your code meets the database.

Step 1: The app we will test

This is a small Brew & Bean API backed by SQLAlchemy 2.0. It stores orders and their items in two tables. The important design detail is in db.py: the route gets its database session through a function get_db. FastAPI calls this a dependency, and in tests we can replace it with one that points at a test database. Without that, tests would write into your real brewbean.db.

Set up

1Create a folder bb-integration and a virtual environment (python -m venv .venv, then activate it).
2Install packages: pip install fastapi httpx sqlalchemy pytest. (httpx is needed by FastAPI's TestClient.)
3Create the three files below: db.py, models.py, main.py.

db.py

import os

from sqlalchemy import create_engine
from sqlalchemy.orm import DeclarativeBase, sessionmaker

DATABASE_URL = os.getenv("DATABASE_URL", "sqlite:///./brewbean.db")

engine = create_engine(DATABASE_URL, connect_args={"check_same_thread": False})
SessionLocal = sessionmaker(bind=engine, autoflush=False)


class Base(DeclarativeBase):
    pass


def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

The engine, a session factory, and get_db, which gives each request its own session and always closes it.

models.py

from datetime import datetime, timezone

from sqlalchemy import ForeignKey
from sqlalchemy.orm import Mapped, mapped_column, relationship

from db import Base


def utcnow() -> datetime:
    return datetime.now(timezone.utc)


class Order(Base):
    __tablename__ = "orders"

    id: Mapped[int] = mapped_column(primary_key=True)
    customer_name: Mapped[str]
    phone: Mapped[str]
    total: Mapped[int]
    status: Mapped[str] = mapped_column(default="received")
    created_at: Mapped[datetime] = mapped_column(default=utcnow)
    items: Mapped[list["OrderItem"]] = relationship(
        back_populates="order", cascade="all, delete-orphan"
    )


class OrderItem(Base):
    __tablename__ = "order_items"

    id: Mapped[int] = mapped_column(primary_key=True)
    order_id: Mapped[int] = mapped_column(ForeignKey("orders.id"))
    item_id: Mapped[str]
    qty: Mapped[int]
    order: Mapped[Order] = relationship(back_populates="items")

Two tables: orders and order_items. The relationship with cascade makes items belong to their order.

main.py

from contextlib import asynccontextmanager

from fastapi import Depends, FastAPI, HTTPException
from pydantic import BaseModel, Field
from sqlalchemy.orm import Session

import models
from db import Base, engine, get_db

PRICES = {"espresso": 120, "cappuccino": 180, "cold-brew": 220, "filter-coffee": 80,
          "masala-chai": 90, "croissant": 150, "banana-bread": 130, "veg-sandwich": 160}



@asynccontextmanager
async def lifespan(app: FastAPI):
    Base.metadata.create_all(engine)
    yield


app = FastAPI(title="Brew & Bean API", lifespan=lifespan)


class ItemIn(BaseModel):
    item_id: str
    qty: int = Field(ge=1, le=20)


class OrderIn(BaseModel):
    customer_name: str = Field(min_length=1)
    phone: str = Field(pattern=r"^[6-9]\d{9}$")
    items: list[ItemIn] = Field(min_length=1)


def to_dict(order: models.Order) -> dict:
    return {
        "id": order.id,
        "customer_name": order.customer_name,
        "phone": order.phone,
        "items": [{"item_id": i.item_id, "qty": i.qty} for i in order.items],
        "total": order.total,
        "status": order.status,
    }


@app.post("/orders", status_code=201)
def create_order(body: OrderIn, db: Session = Depends(get_db)):
    unknown = [i.item_id for i in body.items if i.item_id not in PRICES]
    if unknown:
        raise HTTPException(422, f"Unknown items: {unknown}")
    order = models.Order(
        customer_name=body.customer_name,
        phone=body.phone,
        total=sum(PRICES[i.item_id] * i.qty for i in body.items),
        items=[models.OrderItem(item_id=i.item_id, qty=i.qty) for i in body.items],
    )
    db.add(order)
    db.commit()
    db.refresh(order)
    return to_dict(order)


@app.get("/orders/{order_id}")
def get_order(order_id: int, db: Session = Depends(get_db)):
    order = db.get(models.Order, order_id)
    if order is None:
        raise HTTPException(404, "Order not found")
    return to_dict(order)


@app.get("/orders")
def list_orders(status: str | None = None, db: Session = Depends(get_db)):
    query = db.query(models.Order)
    if status:
        query = query.filter(models.Order.status == status)
    return [to_dict(o) for o in query.order_by(models.Order.id).all()]

POST /orders validates, computes the total on the server and saves; GET /orders/{id} reads it back; GET /orders?status= filters. Tables are created on startup.

Step 2: A fresh database for every test

This is the heart of integration testing. Each test must start from a known, empty state, otherwise tests influence each other and fail in confusing ways ("it passes alone but fails with the others").

The fixture db_session creates a new SQLite file in a temporary folder for each test (tmp_path is a built-in pytest fixture that gives each test its own folder), creates the tables, and closes everything afterwards. The client fixture then overrides get_db so the app uses this session. app.dependency_overrides.clear() removes the override when the test ends.

conftest.py

import pytest
from fastapi.testclient import TestClient
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

from db import Base, get_db
from main import app


@pytest.fixture
def db_session(tmp_path):
    """A brand-new SQLite file for every test, so tests never share data."""
    engine = create_engine(
        f"sqlite:///{tmp_path / 'test.db'}", connect_args={"check_same_thread": False}
    )
    Base.metadata.create_all(engine)
    TestingSession = sessionmaker(bind=engine, autoflush=False)
    session = TestingSession()
    try:
        yield session
    finally:
        session.close()
        engine.dispose()


@pytest.fixture
def client(db_session):
    def override_get_db():
        yield db_session

    app.dependency_overrides[get_db] = override_get_db
    yield TestClient(app)
    app.dependency_overrides.clear()

Every test that asks for client gets an app wired to its own private database.

Step 3: Write the tests

Look at what the first test does: it uses the API to create an order, then uses the database session to check what really got stored. That second step is what a pure API test cannot do. The third test checks the failure path: a rejected order must leave zero rows behind.

test_orders_db.py

import models

PAYLOAD = {
    "customer_name": "Asha",
    "phone": "9876543210",
    "items": [{"item_id": "espresso", "qty": 2}, {"item_id": "croissant", "qty": 1}],
}


def test_order_is_really_saved(client, db_session):
    response = client.post("/orders", json=PAYLOAD)
    assert response.status_code == 201

    saved = db_session.get(models.Order, response.json()["id"])
    assert saved is not None
    assert saved.total == 390
    assert saved.status == "received"
    assert sorted((i.item_id, i.qty) for i in saved.items) == [("croissant", 1), ("espresso", 2)]


def test_get_reads_back_from_the_database(client):
    created = client.post("/orders", json=PAYLOAD).json()

    response = client.get(f"/orders/{created['id']}")

    assert response.status_code == 200
    assert response.json() == created


def test_rejected_order_leaves_no_rows(client, db_session):
    bad = {**PAYLOAD, "items": [{"item_id": "pizza", "qty": 1}]}

    assert client.post("/orders", json=bad).status_code == 422

    assert db_session.query(models.Order).count() == 0
    assert db_session.query(models.OrderItem).count() == 0


def test_filter_by_status(client, db_session):
    client.post("/orders", json=PAYLOAD)
    second = client.post("/orders", json=PAYLOAD).json()
    row = db_session.get(models.Order, second["id"])
    row.status = "ready"
    db_session.commit()

    ready = client.get("/orders", params={"status": "ready"}).json()

    assert [o["id"] for o in ready] == [second["id"]]


def test_tests_do_not_share_data(client):
    assert client.get("/orders").json() == []

Five tests covering save, read-back, failure leaves no rows, filtering, and isolation between tests.

Terminal

pytest -v

Expected output (a deprecation warning from a library may also appear; times will differ)

test_orders_db.py::test_order_is_really_saved PASSED
test_orders_db.py::test_get_reads_back_from_the_database PASSED
test_orders_db.py::test_rejected_order_leaves_no_rows PASSED
test_orders_db.py::test_filter_by_status PASSED
test_orders_db.py::test_tests_do_not_share_data PASSED

5 passed in 0.3s

Assert on stored state

Check the database directly (db_session.get(...)) in at least one test per write endpoint. A response that says 201 Created only proves the code *believed* it saved the data.

Why a separate test database matters

Never run tests against a database that holds anything you care about. Tests create and delete data freely. The pattern above is safe because the database file lives in a temporary folder and the app's real DATABASE_URL is never touched while the override is active.

Other common isolation strategies, from simplest to fastest:

New database per test (what we did): simplest and very reliable. Fine for hundreds of tests on SQLite.

Empty the tables after each test: faster on bigger databases, but you must remember every table.

Wrap each test in a transaction and roll it back: the fastest, a little harder to set up, and it does not work for code that itself commits unless you use nested transactions.

Step 4: Run the same tests on Postgres

SQLite is great for learning and fast feedback, but production usually runs Postgres. They differ in details: type strictness, locking, some SQL functions. If production uses Postgres, you should run your integration tests on Postgres too, at least in CI. The only change is where the test database lives. Start a throw-away Postgres in Docker on a different port from any real one, and let the fixture read its address from an environment variable.

Terminal

docker run --name bb-test-db -e POSTGRES_PASSWORD=test -e POSTGRES_DB=brewbean_test -p 5433:5432 -d postgres:16
pip install "psycopg[binary]"

Starts a disposable Postgres 16 container on port 5433 and installs the Python driver. Remove it later with docker rm -f bb-test-db.

conftest.py (replace the db_session fixture)

import os

import pytest
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

from db import Base


@pytest.fixture
def db_session(tmp_path):
    url = os.getenv("TEST_DATABASE_URL") or f"sqlite:///{tmp_path / 'test.db'}"
    kwargs = {"connect_args": {"check_same_thread": False}} if url.startswith("sqlite") else {}
    engine = create_engine(url, **kwargs)
    Base.metadata.drop_all(engine)
    Base.metadata.create_all(engine)
    session = sessionmaker(bind=engine, autoflush=False)()
    try:
        yield session
    finally:
        session.close()
        Base.metadata.drop_all(engine)
        engine.dispose()

Uses SQLite by default and Postgres when TEST_DATABASE_URL is set. Keep the client fixture from Step 2 unchanged.

Run the tests on Postgres (macOS, Linux, Git Bash)

TEST_DATABASE_URL=postgresql+psycopg://postgres:test@localhost:5433/brewbean_test pytest -q

On Windows PowerShell, first run $env:TEST_DATABASE_URL = "postgresql+psycopg://postgres:test@localhost:5433/brewbean_test" and then pytest -q.

This deletes data

drop_all deletes every table the models define. Point TEST_DATABASE_URL only at a database made for tests. Never reuse your development or production database URL here.

For bigger projects, a library called Testcontainers can start and stop the Postgres container from inside your test run, so teammates do not need to run docker by hand. The idea is the same as above; check its official documentation for the Python setup.

Using AI to help

Prompt · Review my integration tests

You are a senior SDET reviewing integration tests for a [FastAPI / Express / Spring] API that uses [Postgres / SQLite / MongoDB].

Here are my fixtures and tests:
[paste conftest.py and test file]

Check for: tests that share data, missing cleanup, tests that only check the HTTP response and never the stored data, missing failure-path tests (what is left in the database after a rejected request), and slow patterns. List the problems in order of importance and show a fix for each.

What to expect: Look for concrete, fixable items. If the AI suggests mocking the database, remember the point of this lesson: keep your own database real.

How this comes up in interviews

Expect: "What is the difference between unit and integration tests?", "How do you keep tests from sharing data?", "Would you use an in-memory database or the real one?" (same engine as production when possible; SQLite is fine for speed with a Postgres run in CI) and "How do you test that a failed request does not leave partial data?". Mentioning dependency overrides, per-test databases or transaction rollback shows real experience.

Try it: Test a status change

Add a PATCH /orders/{id}/status endpoint to main.py that updates the status and returns the order (404 if it does not exist). Then write a test that changes an order to preparing and checks the database, not just the response.

Try it: Make the status flow strict

Write a failing test first: moving an order directly from received to picked_up must return 409 and leave the status unchanged. Then change the endpoint so the test passes. Keep the allowed flow: received to preparing to ready to picked_up, and cancel only from received or preparing.

Integration test checklist

Each test starts from an empty database it owns.
The app is pointed at the test database through a dependency override (or an environment variable), never the real one.
At least one assertion per write endpoint reads the stored data.
There is a failure-path test that checks nothing was left behind.
The tests also run on the same database engine as production, at least in CI.

Key takeaways

Integration tests run your code with real collaborators, such as a real database, and find bugs mocks hide.
Isolation is the main challenge: give every test a fresh, known database state.
FastAPI dependency overrides let tests swap the database without changing app code.
Assert on the stored data as well as the HTTP response, and test that failures leave no partial data.
SQLite is fast for local runs; run the suite on your production engine (for example Postgres in Docker) in CI.
Never point tests at a database with real data: drop_all and cleanup code delete things.

Check yourself

Finishing the quiz marks the lesson as done.

1. Which bug is an integration test with a real database most likely to find that a mocked unit test would miss?

2. Why does the db_session fixture create a new database for every test?

3. What does app.dependency_overrides[get_db] = override_get_db do?

4. A test posts an invalid order and gets 422. What else should an integration test check?

5. Your production database is Postgres. What is a sensible plan for integration tests?

Saved on this device. Create a free account to keep it on every device.

Mocks, stubs and test doubles Contract testing and OpenAPI validation

Practise this with Sam

A real mock interview, out loud, with an honest scorecard. Free to start.

Start free