Skip to main content

Python 18 - Proving It Works: pytest and API Testing

Updated 21 min readMarkdown
Part of seriespython

Replace "I clicked through /docs" with a test suite that checks every route, with real users and a real database, in seconds.

Where StudyDesk is now

src/studydesk/
├── __init__.py, config.py, database.py, models.py, schemas.py, crud.py
├── dependencies.py, security.py, tasks.py, worker.py, seed.py
└── routers/auth.py, students.py, jobs.py
alembic/, Dockerfile, docker-compose.yml

No tests. By the end of this article, uv run pytest runs a suite that exercises authentication, every student route with staff and admin users, the job endpoints, and the CRUD layer directly, against a separate test database that is clean for every test.


Part 1: The Twelve-Route Checklist

After adding phone, Aman tested by hand: log in as staff, list, get, create, update, try to delete (403), log in as admin, delete, queue a job, check the job. Twelve routes, two users, maybe fifteen minutes. He will do it again after the next change. And the one after that. One day he will skip it, and that is the day something breaks.

Neha says, "A test is that checklist, written as code. You run it in five seconds, after every change, and it never forgets a step."


Part 2: pytest in Five Minutes

pytest is Python's standard test runner. The rules are simple:

  • Tests live in files named test_*.py.
  • Each test is a function named test_*.
  • A test passes if it runs without raising. It fails on an assert that is false or any exception.
# tests/test_example.py
def test_addition():
    assert 2 + 2 == 4
uv run pytest
tests/test_example.py .                                              [100%]
1 passed in 0.01s

Plain assert, no special methods. When an assertion fails, pytest shows both sides of the comparison:

    def test_addition():
>       assert 2 + 2 == 5
E       assert 4 == 5

Fixtures

Most tests need setup: a database session, a logged-in user, a client to send requests. pytest handles setup with fixtures: functions decorated with @pytest.fixture whose return value is injected into any test that names it as a parameter.

import pytest

@pytest.fixture
def numbers():
    return [1, 2, 3]

def test_sum(numbers):
    assert sum(numbers) == 6

pytest sees that test_sum has a parameter called numbers, finds the fixture with that name, calls it, and passes the result in. This is the same idea as FastAPI's Depends, and once you have used one, the other feels familiar.

Fixtures can use yield for setup and teardown, exactly like get_db():

@pytest.fixture
def temp_file():
    f = open("scratch.txt", "w")   # setup
    yield f                         # the test runs here
    f.close()                       # teardown, even if the test failed

Fixtures shared across files go in a file named conftest.py; pytest loads it automatically.

Install

uv add --dev pytest httpx

httpx is the HTTP client that FastAPI's TestClient is built on.

Add pytest settings to the project file:

pyproject.toml — add this section at the bottom

[tool.pytest.ini_options]
testpaths = ["tests"]

Part 3: The Test Database

Tests create, update, and delete rows. They must never touch the real studydesk database. We use a second database, studydesk_test, that tests own completely.

Create it

psql "postgresql://postgres:postgres@localhost:5432/postgres" -c "CREATE DATABASE studydesk_test;"

Config

.env — add one line

TEST_DATABASE_URL=postgresql+psycopg://postgres:postgres@localhost:5432/studydesk_test

.env.example — add

TEST_DATABASE_URL=postgresql+psycopg://USER:PASSWORD@localhost:5432/studydesk_test

src/studydesk/config.py

import os
from pathlib import Path

from dotenv import load_dotenv

BASE_DIR = Path(__file__).resolve().parent.parent.parent  # project root
load_dotenv(BASE_DIR / ".env")

DATABASE_URL: str = os.environ["DATABASE_URL"]
TEST_DATABASE_URL: str | None = os.getenv("TEST_DATABASE_URL")

ALLOWED_ORIGINS: list[str] = [
    origin.strip()
    for origin in os.getenv("ALLOWED_ORIGINS", "").split(",")
    if origin.strip()
]

HOST: str = os.getenv("HOST", "127.0.0.1")
PORT: int = int(os.getenv("PORT", "8000"))
RELOAD: bool = os.getenv("RELOAD", "true").lower() == "true"

SECRET_KEY: str = os.environ["SECRET_KEY"]
ALGORITHM: str = os.getenv("ALGORITHM", "HS256")
ACCESS_TOKEN_EXPIRE_MINUTES: int = int(os.getenv("ACCESS_TOKEN_EXPIRE_MINUTES", "60"))

ADMIN_EMAIL: str | None = os.getenv("ADMIN_EMAIL")
ADMIN_PASSWORD: str | None = os.getenv("ADMIN_PASSWORD")

The isolation problem

Thirty tests that all insert students into the same table will step on each other: test 7 expects three students and finds nine because tests 1–6 left theirs behind. Each test must start from a clean slate.

Dropping and recreating tables between every test works but is slow. The standard trick is faster: open a transaction before each test and roll it back after. Every insert, update, and delete the test did simply vanishes. The database is back to empty in a millisecond.


Part 4: conftest.py — The Fixtures

This file is the foundation. Read it slowly; every test will use it.

tests/conftest.py

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

from studydesk import app
from studydesk.config import TEST_DATABASE_URL
from studydesk.database import Base, get_db
from studydesk.models import User
from studydesk.security import create_access_token, hash_password

if not TEST_DATABASE_URL:
    raise RuntimeError("TEST_DATABASE_URL is not set in .env")

test_engine = create_engine(TEST_DATABASE_URL)
TestSessionLocal = sessionmaker(bind=test_engine, autoflush=False, autocommit=False)


# ---------- database ----------


@pytest.fixture(scope="session", autouse=True)
def create_test_tables():
    """Create all tables once before the test session, drop them after."""
    Base.metadata.create_all(bind=test_engine)
    yield
    Base.metadata.drop_all(bind=test_engine)


@pytest.fixture
def db() -> Session:
    """A session wrapped in a transaction that is rolled back after each test."""
    connection = test_engine.connect()
    transaction = connection.begin()
    session = TestSessionLocal(bind=connection)

    yield session

    session.close()
    transaction.rollback()
    connection.close()


# ---------- client ----------


@pytest.fixture
def client(db: Session) -> TestClient:
    """A TestClient whose routes use the test session instead of the real one."""

    def override_get_db():
        yield db

    app.dependency_overrides[get_db] = override_get_db
    with TestClient(app) as test_client:
        yield test_client
    app.dependency_overrides.clear()


# ---------- users and tokens ----------


def make_user(db: Session, email: str, password: str, role: str) -> User:
    user = User(email=email, hashed_password=hash_password(password), role=role)
    db.add(user)
    db.flush()
    return user


@pytest.fixture
def staff_user(db: Session) -> User:
    return make_user(db, "staff@test.com", "staff-pass-123", "staff")


@pytest.fixture
def admin_user(db: Session) -> User:
    return make_user(db, "admin@test.com", "admin-pass-123", "admin")


@pytest.fixture
def staff_headers(staff_user: User) -> dict:
    token = create_access_token(subject=staff_user.email, role=staff_user.role)
    return {"Authorization": f"Bearer {token}"}


@pytest.fixture
def admin_headers(admin_user: User) -> dict:
    token = create_access_token(subject=admin_user.email, role=admin_user.role)
    return {"Authorization": f"Bearer {token}"}


# ---------- sample data ----------


SAMPLE_STUDENT = {
    "name": "Priya Sharma",
    "email": "priya@example.com",
    "phone": "+919876543210",
    "age": 17,
    "city": "Ahmedabad",
    "marks": 88,
}


@pytest.fixture
def sample_student() -> dict:
    return dict(SAMPLE_STUDENT)

Reading it

test_engine and TestSessionLocal A second engine, pointing at studydesk_test. The real engine in database.py is never used by tests.

create_test_tables — scope="session", autouse=True Fixture scope controls how often it runs. The default is once per test. scope="session" means once for the whole run. autouse=True means every test uses it without naming it. So: tables are created once at the start, dropped once at the end. We use create_all here rather than Alembic because the test database is disposable and create_all is faster; the migrations are tested by the fact that alembic upgrade head runs in Compose.

db — the rollback trick

connection = test_engine.connect()
transaction = connection.begin()        # BEGIN
session = TestSessionLocal(bind=connection)
yield session                           # the test runs
session.close()
transaction.rollback()                  # everything the test did is undone

The session is bound to a connection that already has an open transaction. When CRUD functions call db.commit(), SQLAlchemy commits a nested savepoint inside the outer transaction; the outer one is never committed. At the end we roll it back and the table is empty again. Every test starts clean, and tests cannot see each other's data.

client — dependency override

app.dependency_overrides[get_db] = override_get_db

This is the line we promised in Article 7. app.dependency_overrides is a dictionary: "whenever a route asks for get_db, call override_get_db instead." Our override yields the rollback-wrapped test session. Every route — through DbSession, get_current_user, get_student_or_404 — now talks to the test database, and not one line of application code knows.

TestClient(app) gives an object with .get(), .post(), .put(), .delete() that call the app in-process, no server, no network. The with block runs the lifespan on entry and exit. After the test we clear the overrides so nothing leaks.

make_user uses db.flush(), not commit() flush() sends the INSERT so the row gets an id, but stays inside the transaction. We never commit in tests; the rollback would undo it anyway, but flush keeps the intent clear.

Tokens are minted directly create_access_token(...) from security.py makes a valid token without going through /auth/login. The login route gets its own tests; everywhere else, a ready-made header keeps tests short.

sample_student returns a copy dict(SAMPLE_STUDENT) so a test that modifies it does not affect the next test.


Part 5: Testing Authentication

tests/test_auth.py

from fastapi.testclient import TestClient


def test_register_creates_staff_user(client: TestClient):
    response = client.post(
        "/auth/register",
        json={"email": "ritu@test.com", "password": "ritu-pass-123"},
    )
    assert response.status_code == 201
    body = response.json()
    assert body["email"] == "ritu@test.com"
    assert body["role"] == "staff"
    assert body["is_active"] is True
    assert "password" not in body
    assert "hashed_password" not in body


def test_register_duplicate_email_is_409(client: TestClient):
    payload = {"email": "ritu@test.com", "password": "ritu-pass-123"}
    assert client.post("/auth/register", json=payload).status_code == 201
    assert client.post("/auth/register", json=payload).status_code == 409


def test_register_short_password_is_422(client: TestClient):
    response = client.post("/auth/register", json={"email": "x@test.com", "password": "short"})
    assert response.status_code == 422


def test_login_returns_token(client: TestClient, staff_user):
    response = client.post(
        "/auth/login",
        json={"email": "staff@test.com", "password": "staff-pass-123"},
    )
    assert response.status_code == 200
    body = response.json()
    assert body["token_type"] == "bearer"
    assert len(body["access_token"]) > 20


def test_login_wrong_password_is_401(client: TestClient, staff_user):
    response = client.post(
        "/auth/login",
        json={"email": "staff@test.com", "password": "wrong"},
    )
    assert response.status_code == 401


def test_login_unknown_email_is_401(client: TestClient):
    response = client.post(
        "/auth/login",
        json={"email": "nobody@test.com", "password": "whatever"},
    )
    assert response.status_code == 401


def test_me_without_token_is_401(client: TestClient):
    assert client.get("/auth/me").status_code == 401


def test_me_with_bad_token_is_401(client: TestClient):
    response = client.get("/auth/me", headers={"Authorization": "Bearer not-a-real-token"})
    assert response.status_code == 401


def test_me_returns_current_user(client: TestClient, staff_headers):
    response = client.get("/auth/me", headers=staff_headers)
    assert response.status_code == 200
    assert response.json()["email"] == "staff@test.com"

Reading it

Each test is one sentence of the checklist, named so that a failure reads like a bug report: test_login_wrong_password_is_401 FAILED tells you what broke without opening the file.

client.post(..., json=...) sends a JSON body, exactly like requests. response.json() parses the body. assert "hashed_password" not in body — this is the UserOut whitelist from Article 10, now guaranteed by a test.

Run them:

uv run pytest tests/test_auth.py -v

-v prints one line per test:

tests/test_auth.py::test_register_creates_staff_user PASSED
tests/test_auth.py::test_register_duplicate_email_is_409 PASSED
...
9 passed in 1.84s

Part 6: Testing the Student Routes

The welcome email problem

POST /students queues send_welcome_email, which sleeps two seconds. TestClient runs background tasks synchronously before returning, so every create test would take two seconds. We do not want to test the email here; we want to test the route. So we replace the email function with a fake for the duration of the test, using pytest's monkeypatch fixture.

tests/test_students.py

import pytest
from fastapi.testclient import TestClient


@pytest.fixture(autouse=True)
def no_real_email(monkeypatch):
    """Replace the slow welcome email with an instant fake for every test in this file."""
    sent: list[tuple[str, str]] = []

    def fake_send(to_email: str, student_name: str) -> None:
        sent.append((to_email, student_name))

    monkeypatch.setattr("studydesk.routers.students.send_welcome_email", fake_send)
    return sent


# ---------- authentication on the whole router ----------


def test_list_requires_login(client: TestClient):
    assert client.get("/students").status_code == 401


def test_create_requires_login(client: TestClient, sample_student):
    assert client.post("/students", json=sample_student).status_code == 401


# ---------- create ----------


def test_create_student(client: TestClient, staff_headers, sample_student, no_real_email):
    response = client.post("/students", json=sample_student, headers=staff_headers)
    assert response.status_code == 201
    body = response.json()
    assert body["id"] > 0
    assert body["name"] == "Priya Sharma"
    assert body["phone"] == "+919876543210"
    assert body["grade"] == "A"
    assert no_real_email == [("priya@example.com", "Priya Sharma")]


def test_create_cleans_name_and_city(client: TestClient, staff_headers, sample_student):
    sample_student["name"] = "   priya sharma "
    sample_student["city"] = "ahmedabad"
    response = client.post("/students", json=sample_student, headers=staff_headers)
    assert response.status_code == 201
    assert response.json()["name"] == "Priya Sharma"
    assert response.json()["city"] == "Ahmedabad"


def test_create_duplicate_email_is_409(client: TestClient, staff_headers, sample_student):
    assert client.post("/students", json=sample_student, headers=staff_headers).status_code == 201
    assert client.post("/students", json=sample_student, headers=staff_headers).status_code == 409


@pytest.mark.parametrize(
    "field, bad_value",
    [
        ("email", "not-an-email"),
        ("age", 3),
        ("marks", 150),
        ("name", "P"),
        ("name", "Priya 123"),
        ("phone", "12345"),
    ],
)
def test_create_invalid_field_is_422(client: TestClient, staff_headers, sample_student, field, bad_value):
    sample_student[field] = bad_value
    response = client.post("/students", json=sample_student, headers=staff_headers)
    assert response.status_code == 422


# ---------- read ----------


def test_get_student(client: TestClient, staff_headers, sample_student):
    created = client.post("/students", json=sample_student, headers=staff_headers).json()
    response = client.get(f"/students/{created['id']}", headers=staff_headers)
    assert response.status_code == 200
    assert response.json()["email"] == "priya@example.com"


def test_get_missing_student_is_404(client: TestClient, staff_headers):
    assert client.get("/students/999", headers=staff_headers).status_code == 404


def test_get_invalid_id_is_422(client: TestClient, staff_headers):
    assert client.get("/students/abc", headers=staff_headers).status_code == 422
    assert client.get("/students/0", headers=staff_headers).status_code == 422


# ---------- list, filter, sort, paginate ----------


@pytest.fixture
def five_students(client: TestClient, staff_headers, sample_student):
    """Create five students with different cities and marks."""
    data = [
        ("Priya Sharma", "priya@example.com", "Ahmedabad", 88),
        ("Rahul Verma", "rahul@example.com", "Surat", 72),
        ("Sneha Patel", "sneha@example.com", "Ahmedabad", 95),
        ("Arjun Mehta", "arjun@example.com", "Vadodara", 58),
        ("Dev Joshi", "dev@example.com", "Surat", 45),
    ]
    for name, email, city, marks in data:
        payload = dict(sample_student, name=name, email=email, city=city, marks=marks)
        assert client.post("/students", json=payload, headers=staff_headers).status_code == 201


def test_list_all(client: TestClient, staff_headers, five_students):
    response = client.get("/students", headers=staff_headers)
    assert response.status_code == 200
    body = response.json()
    assert body["total"] == 5
    assert body["count"] == 5
    assert body["total_pages"] == 1


def test_list_filter_by_city(client: TestClient, staff_headers, five_students):
    body = client.get("/students?city=surat", headers=staff_headers).json()
    assert body["total"] == 2
    assert {s["name"] for s in body["students"]} == {"Rahul Verma", "Dev Joshi"}


def test_list_filter_by_grade(client: TestClient, staff_headers, five_students):
    body = client.get("/students?grade=A", headers=staff_headers).json()
    assert body["total"] == 2
    assert all(s["grade"] == "A" for s in body["students"])


def test_list_sort_by_marks_desc(client: TestClient, staff_headers, five_students):
    body = client.get("/students?sort_by=marks&order=desc", headers=staff_headers).json()
    marks = [s["marks"] for s in body["students"]]
    assert marks == [95, 88, 72, 58, 45]


def test_list_pagination(client: TestClient, staff_headers, five_students):
    page1 = client.get("/students?limit=2&offset=0", headers=staff_headers).json()
    page3 = client.get("/students?limit=2&offset=4", headers=staff_headers).json()
    assert page1["total"] == 5
    assert page1["total_pages"] == 3
    assert page1["count"] == 2
    assert page3["count"] == 1


def test_list_bad_sort_field_is_422(client: TestClient, staff_headers):
    assert client.get("/students?sort_by=email", headers=staff_headers).status_code == 422


# ---------- update ----------


def test_update_partial(client: TestClient, staff_headers, sample_student):
    created = client.post("/students", json=sample_student, headers=staff_headers).json()
    response = client.put(f"/students/{created['id']}", json={"marks": 60}, headers=staff_headers)
    assert response.status_code == 200
    body = response.json()
    assert body["marks"] == 60
    assert body["grade"] == "C"
    assert body["name"] == "Priya Sharma"  # untouched


def test_update_empty_body_is_400(client: TestClient, staff_headers, sample_student):
    created = client.post("/students", json=sample_student, headers=staff_headers).json()
    assert client.put(f"/students/{created['id']}", json={}, headers=staff_headers).status_code == 400


def test_update_to_taken_email_is_409(client: TestClient, staff_headers, sample_student):
    first = client.post("/students", json=sample_student, headers=staff_headers).json()
    second = client.post(
        "/students", json=dict(sample_student, email="other@example.com"), headers=staff_headers
    ).json()
    response = client.put(
        f"/students/{second['id']}", json={"email": first["email"]}, headers=staff_headers
    )
    assert response.status_code == 409


# ---------- delete and authorization ----------


def test_delete_as_staff_is_403(client: TestClient, staff_headers, sample_student):
    created = client.post("/students", json=sample_student, headers=staff_headers).json()
    assert client.delete(f"/students/{created['id']}", headers=staff_headers).status_code == 403


def test_delete_as_admin(client: TestClient, staff_headers, admin_headers, sample_student):
    created = client.post("/students", json=sample_student, headers=staff_headers).json()
    assert client.delete(f"/students/{created['id']}", headers=admin_headers).status_code == 204
    assert client.get(f"/students/{created['id']}", headers=admin_headers).status_code == 404


def test_delete_missing_is_404(client: TestClient, admin_headers):
    assert client.delete("/students/999", headers=admin_headers).status_code == 404

Reading the new pieces

monkeypatch.setattr("studydesk.routers.students.send_welcome_email", fake_send) monkeypatch is a built-in pytest fixture that temporarily replaces an attribute and restores it automatically when the test ends. The string is the full path to the name as the router sees it: routers/students.py did from ..tasks import send_welcome_email, so the name lives in the router module, and that is where we replace it. Patching studydesk.tasks.send_welcome_email would not affect the router's already-imported reference.

The fixture returns the sent list so tests can check that the route did queue an email with the right arguments, without waiting two seconds.

autouse=True on a fixture inside a test file applies it to every test in that file only.

@pytest.mark.parametrize Runs one test function several times with different inputs. The names in the string ("field, bad_value") become parameters; each tuple in the list is one run. pytest reports them separately:

test_create_invalid_field_is_422[email-not-an-email] PASSED
test_create_invalid_field_is_422[age-3] PASSED
...

Six validation rules checked with one function.

five_students fixture Builds data through the API itself. Fixtures can depend on other fixtures (client, staff_headers), just like dependencies in FastAPI.

dict(sample_student, name=name, ...) Makes a copy with some keys replaced. Short and readable.

Sets for order-insensitive comparison {s["name"] for s in ...} == {"Rahul Verma", "Dev Joshi"} — the filter test should not care about order; the sort test does, so it uses a list.

Run:

uv run pytest tests/test_students.py -v

Twenty-six tests, around two seconds.


Part 7: Testing Jobs

tests/test_jobs.py

from fastapi.testclient import TestClient
from sqlalchemy.orm import Session

from studydesk import crud
from studydesk.worker import run_one


def test_queue_report_requires_admin(client: TestClient, staff_headers):
    assert client.post("/jobs/grade-report", headers=staff_headers).status_code == 403


def test_queue_report_returns_202_pending(client: TestClient, admin_headers):
    response = client.post("/jobs/grade-report?city=Surat", headers=admin_headers)
    assert response.status_code == 202
    body = response.json()
    assert body["status"] == "pending"
    assert body["payload"] == {"city": "Surat"}
    assert body["attempts"] == 0


def test_get_job(client: TestClient, admin_headers):
    created = client.post("/jobs/grade-report", headers=admin_headers).json()
    response = client.get(f"/jobs/{created['id']}", headers=admin_headers)
    assert response.status_code == 200
    assert response.json()["id"] == created["id"]


def test_get_missing_job_is_404(client: TestClient, admin_headers):
    assert client.get("/jobs/999", headers=admin_headers).status_code == 404


def test_worker_runs_grade_report(client: TestClient, admin_headers, staff_headers, sample_student, db: Session, monkeypatch):
    monkeypatch.setattr("studydesk.tasks.time.sleep", lambda seconds: None)

    client.post("/students", json=sample_student, headers=staff_headers)
    created = client.post("/jobs/grade-report", headers=admin_headers).json()

    assert run_one(db) is True

    job = crud.get_job(db, created["id"])
    assert job.status == "done"
    assert job.attempts == 1
    assert job.result["total"] == 1
    assert job.result["by_grade"]["A"] == 1


def test_worker_retries_then_fails(db: Session, monkeypatch):
    def always_fails(db, payload):
        raise RuntimeError("boom")

    monkeypatch.setitem(__import__("studydesk.tasks", fromlist=["TASKS"]).TASKS, "grade_report", always_fails)

    job = crud.create_job(db, kind="grade_report")

    for attempt in range(1, 4):
        assert run_one(db) is True
        db.refresh(job)
        assert job.attempts == attempt
        assert job.status == ("failed" if attempt == 3 else "pending")

    assert "boom" in job.error


def test_worker_with_empty_queue_returns_false(db: Session):
    assert run_one(db) is False

Reading it

Testing the worker without a worker process run_one(db) is a plain function. We call it directly with the test session, so the whole claim → run → finish cycle is tested in-process, with no polling loop.

monkeypatch.setattr("studydesk.tasks.time.sleep", lambda seconds: None) generate_grade_report sleeps five seconds. Replace sleep inside the tasks module with a function that does nothing.

monkeypatch.setitem(TASKS, "grade_report", always_fails) Replaces one entry of the TASKS dict with a function that always raises, restoring it afterwards. The test then walks through all three attempts and checks the status at each step: pending, pending, failed. This is the retry logic from Article 11, verified.

(The __import__ form is only there to get a handle on the TASKS dict in one line; from studydesk.tasks import TASKS at the top of the file works just as well.)


Part 8: Testing the CRUD Layer Directly

Routes are the outside of the house. The CRUD functions are the plumbing, and they can be tested without HTTP at all: just a session and a call. These tests are the fastest and the most precise.

tests/test_crud.py

import pytest
from sqlalchemy.exc import IntegrityError
from sqlalchemy.orm import Session

from studydesk import crud
from studydesk.schemas import StudentCreate, UserCreate


def make(db: Session, **overrides) -> int:
    data = {
        "name": "Priya Sharma",
        "email": "priya@example.com",
        "age": 17,
        "city": "Ahmedabad",
        "marks": 88,
    }
    data.update(overrides)
    return crud.create_student(db, StudentCreate(**data)).id


def test_create_and_get(db: Session):
    student_id = make(db)
    student = crud.get_student(db, student_id)
    assert student is not None
    assert student.name == "Priya Sharma"


def test_get_by_email_is_case_insensitive(db: Session):
    make(db)
    assert crud.get_student_by_email(db, "PRIYA@EXAMPLE.COM") is not None


def test_unique_email_enforced_by_database(db: Session):
    make(db)
    with pytest.raises(IntegrityError):
        make(db, name="Someone Else")
    db.rollback()


def test_list_filters_sorts_and_paginates(db: Session):
    make(db, email="a@x.com", city="Surat", marks=72)
    make(db, email="b@x.com", city="Surat", marks=45)
    make(db, email="c@x.com", city="Ahmedabad", marks=95)

    page, total = crud.list_students(db, city="surat", sort_by="marks", order="desc", limit=1)
    assert total == 2
    assert len(page) == 1
    assert page[0].marks == 72


def test_update_and_delete(db: Session):
    student_id = make(db)
    student = crud.get_student(db, student_id)

    crud.update_student(db, student, {"marks": 40})
    assert crud.get_student(db, student_id).marks == 40

    crud.delete_student(db, student)
    assert crud.get_student(db, student_id) is None


def test_authenticate_user(db: Session):
    crud.create_user(db, UserCreate(email="Ritu@Test.com", password="ritu-pass-123"))
    assert crud.authenticate_user(db, "ritu@test.com", "ritu-pass-123") is not None
    assert crud.authenticate_user(db, "ritu@test.com", "wrong") is None
    assert crud.authenticate_user(db, "nobody@test.com", "ritu-pass-123") is None


def test_claim_next_job_takes_oldest_first(db: Session):
    first = crud.create_job(db, kind="grade_report", payload={"n": 1})
    crud.create_job(db, kind="grade_report", payload={"n": 2})

    claimed = crud.claim_next_job(db)
    assert claimed.id == first.id
    assert claimed.status == "running"
    assert claimed.attempts == 1

Reading it

pytest.raises(IntegrityError) A context manager that passes only if the block raises that exception. This test proves the unique=True on email really is enforced by PostgreSQL, not just by our pre-check. The db.rollback() afterwards clears the failed state so the fixture's own rollback can run cleanly.

No client, no headers, no JSON These tests call Python functions and inspect Python objects. When a route test fails, a CRUD test tells you whether the problem is in the data layer or in the HTTP layer.


Part 9: Running the Whole Suite

uv run pytest
tests/test_auth.py .........                                         [ 18%]
tests/test_crud.py .......                                           [ 32%]
tests/test_jobs.py .......                                           [ 46%]
tests/test_students.py ..........................                    [100%]

49 passed in 3.41s

Forty-nine checks, three and a half seconds. The fifteen-minute checklist is gone.

Useful flags

Flag What it does
-v one line per test
-x stop at the first failure
-k delete run only tests whose name contains "delete"
--lf re-run only the tests that failed last time
-q quieter output
tests/test_auth.py::test_login_returns_token run exactly one test

Watch a test catch a bug

In dependencies.py, change if user.role != "admin": to if user.role != "staff": (as if someone inverted the check by mistake). Run uv run pytest -x:

FAILED tests/test_students.py::test_delete_as_staff_is_403 - assert 204 == 403

One line tells you a staff user can now delete students. Put it back; green again. That is what the suite is for.

Tests in Docker

The test database is on your laptop, not in Compose, and tests do not need to run inside the container. Running them on the host before docker compose up --build is the normal flow. CI systems (GitHub Actions, for example) start a PostgreSQL service, set TEST_DATABASE_URL, and run uv run pytest on every push; the same conftest.py works unchanged.


Where StudyDesk is after this article

studydesk/
├── pyproject.toml           ← pytest, httpx (dev); [tool.pytest.ini_options]
├── .env                     ← TEST_DATABASE_URL
├── tests/
│   ├── conftest.py          ← test engine, rollback session, client override, users, tokens
│   ├── test_auth.py         ← 9 tests
│   ├── test_crud.py         ← 7 tests
│   ├── test_jobs.py         ← 7 tests
│   └── test_students.py     ← 26 tests
└── src/studydesk/
    ├── config.py            ← TEST_DATABASE_URL
    └── ... (unchanged)
Command Result
uv run pytest 49 passed
uv run pytest -x -v verbose, stop on first failure
uv run pytest tests/test_students.py -k delete only the delete tests

Concept Quiz

Answers are below. No coding needed.

Q1. What makes a function a test in pytest, and what makes it pass or fail?

Q2. What is a fixture, and how does a test receive one?

Q3. How does the db fixture keep every test isolated without dropping tables?

Q4. What does app.dependency_overrides[get_db] = override_get_db do, and why does no application code need to change?

Q5. What does scope="session" mean on create_test_tables, and why is autouse=True added?

Q6. Why is send_welcome_email patched at studydesk.routers.students.send_welcome_email and not at studydesk.tasks.send_welcome_email?

Q7. What does @pytest.mark.parametrize do, and what did it save us in test_students.py?

Q8. How do the job tests exercise the worker without starting the worker process?

Q9. What does pytest.raises(IntegrityError) prove in test_crud.py that the route tests do not?

Q10. Why is TestClient able to run background tasks synchronously, and what problem did that cause?


Answers

  1. A function named test_* in a file named test_*.py. It passes if it finishes without raising; a false assert or any exception fails it.
  2. A function decorated with @pytest.fixture whose return (or yield) value is injected into any test that lists the fixture's name as a parameter.
  3. It opens a transaction before yielding the session and rolls it back after the test. All inserts and updates inside the test are undone; the tables are empty again in milliseconds.
  4. It tells FastAPI to call override_get_db instead of get_db whenever a dependency asks for it. Routes only ever declared db: DbSession; they never knew how the session was made, so swapping the source is invisible to them.
  5. The fixture runs once for the whole test run instead of once per test. autouse=True makes it apply to every test automatically, so tables exist without each test requesting it.
  6. The router imported the name with from ..tasks import send_welcome_email, so it holds its own reference. Patching must target the name where it is looked up, which is the router module.
  7. It runs one test function once per set of inputs, reporting each separately. Six validation rules were tested with one function instead of six copies.
  8. They call run_one(db) directly with the test session. The claim, run, finish, and retry logic is ordinary Python and needs no polling loop or separate process.
  9. That the database's unique constraint itself rejects a duplicate email. The route tests only show a 409, which could come from the Python pre-check alone.
  10. TestClient calls the app in-process and waits for the whole request cycle, including background tasks, to finish. That made every create test wait two seconds for the simulated email until it was patched out.

Epilogue

Aman changes phone's pattern, runs uv run pytest, and sees one red line in three seconds: the parametrized case ("phone", "12345") now passes validation when it should not. He fixes the pattern, runs again, green. He did not open /docs once.

Neha brings the next requirement. "The institute runs courses. A student enrols in several; a course has many students. And when a student is enrolled in a course that has no seats left, nothing may be saved — not the enrollment, not the seat count."

In the next article: SQLAlchemy relationships and transactions — courses, enrollments, one-to-many and many-to-many, and doing several writes as one all-or-nothing unit.

No comments yet. Be the first to comment.