Integration tests against a real database, and what coverage means

4 min read

Why it matters

Unit tests check logic in isolation. Many real bugs live between components: a SQL statement that is wrong, a status code that is not what the client expects, a constraint the database enforces but the code forgets. Integration tests catch these, and a fresh database per test keeps them reliable.

The concept

Test through the public interface. Call the HTTP endpoint with a test client rather than internal functions.
Use a real database of the same kind where you can. SQLite is fine for this example; for PostgreSQL or MySQL in production, run the real engine in a container (for example with Testcontainers).
Isolate every test. Each test gets its own fresh database (pytest's tmp_path), so order does not matter and tests can run in parallel.
Coverage tells you which lines ran during tests. Low coverage points at untested code. High coverage does not mean the tests check the right things.

Worked example: stock reservations

A reservation must never take stock below zero, even if two requests arrive together. The endpoint uses a single conditional UPDATE, so the check and the change happen atomically in the database.

app.py

import sqlite3
from contextlib import closing

from fastapi import Depends, FastAPI, HTTPException
from pydantic import BaseModel, Field

DB_PATH = "shop.db"


def get_db():
    conn = sqlite3.connect(DB_PATH)
    conn.row_factory = sqlite3.Row
    try:
        yield conn
    finally:
        conn.close()


def init_db(path: str) -> None:
    with closing(sqlite3.connect(path)) as conn:
        conn.execute("CREATE TABLE IF NOT EXISTS stock (sku TEXT PRIMARY KEY, qty INTEGER NOT NULL CHECK (qty >= 0))")
        conn.commit()


class Reservation(BaseModel):
    sku: str
    qty: int = Field(gt=0)


app = FastAPI()


@app.post("/reservations", status_code=201)
def reserve(body: Reservation, db: sqlite3.Connection = Depends(get_db)):
    cur = db.execute("UPDATE stock SET qty = qty - ? WHERE sku = ? AND qty >= ?", (body.qty, body.sku, body.qty))
    if cur.rowcount == 0:
        exists = db.execute("SELECT 1 FROM stock WHERE sku = ?", (body.sku,)).fetchone()
        raise HTTPException(409 if exists else 404, "not enough stock" if exists else "unknown sku")
    db.commit()
    left = db.execute("SELECT qty FROM stock WHERE sku = ?", (body.sku,)).fetchone()["qty"]
    return {"sku": body.sku, "reserved": body.qty, "left": left}

test_app.py

import sqlite3

import pytest
from fastapi.testclient import TestClient

import app as shop


@pytest.fixture
def client(tmp_path, monkeypatch):
    db_file = str(tmp_path / "test.db")
    shop.init_db(db_file)
    with sqlite3.connect(db_file) as conn:
        conn.execute("INSERT INTO stock VALUES ('SKU-1', 3)")
    monkeypatch.setattr(shop, "DB_PATH", db_file)
    return TestClient(shop.app)


def test_reserves_and_decrements_stock(client):
    r = client.post("/reservations", json={"sku": "SKU-1", "qty": 2})
    assert r.status_code == 201
    assert r.json() == {"sku": "SKU-1", "reserved": 2, "left": 1}


def test_cannot_oversell(client):
    client.post("/reservations", json={"sku": "SKU-1", "qty": 2})
    r = client.post("/reservations", json={"sku": "SKU-1", "qty": 2})
    assert r.status_code == 409
    assert r.json()["detail"] == "not enough stock"


def test_unknown_sku(client):
    assert client.post("/reservations", json={"sku": "NOPE", "qty": 1}).status_code == 404


@pytest.mark.parametrize("qty", [0, -5])
def test_validation(client, qty):
    assert client.post("/reservations", json={"sku": "SKU-1", "qty": qty}).status_code == 422

Run both lessons' tests with coverage (pip install pytest-cov):

shellsession

$ python3 -m pytest -q --cov=app --cov=pricing --cov-report=term-missing
..............                                                           [100%]
Name         Stmts   Miss  Cover   Missing
------------------------------------------
app.py          28      0   100%
pricing.py      26      1    96%   32
------------------------------------------
TOTAL           54      1    98%
14 passed in 0.47s

Line 32 of pricing.py is the real ExchangeRates.rate, which the unit tests replace with a mock. That is expected: it is covered by a separate, slower test against the real service (or a contract test), not by unit tests.

Notice what 100% coverage of app.py does not prove: the tests never send two reservations at the same time. The conditional UPDATE makes that safe, but only a concurrency test would demonstrate it.

Common mistakes

Shared test databases. Tests pass alone and fail together. Give each test fresh state.
Chasing a coverage number. A test that calls code without meaningful assertions raises coverage and catches nothing.
Mocking the database in integration tests. Then you are not testing the SQL, which is where the bug often is.
Slow suites nobody runs. Keep unit tests in milliseconds and run integration tests in CI on every push.

Interview angle

"How would you test an endpoint that talks to a database?", "what is a flaky test and how do you fix one?" (find the hidden shared state, timing or ordering dependency; never just retry), and "is 100% coverage a good goal?" (useful as a map of untested code, not as a target).

Practice task

Add a test that sends 10 concurrent reservations of 1 unit against stock of 3 using threads, and assert that exactly 3 succeed.
Add a GET /stock/{sku} endpoint and tests, keeping coverage at 100% for app.py with meaningful assertions.

#Why it matters

#The concept

#Worked example: stock reservations

#Common mistakes

#Interview angle

#Practice task

#Further reading