green snake
Photo by Pixabay on Pexels.com

FastAPI Testing Strategies to Raise Quality: pytest, TestClient/httpx, Dependency Overrides, DB Rollbacks, Mocks, Contract Tests, and Load Testing


Summary (Inverted Pyramid)

  • Prepare layered tests—unit, API, integration, contract, and load—and cover them step by step.
  • Design fixtures with pytest, and use TestClient vs. httpx.AsyncClient appropriately.
  • Leverage dependency injection (Depends) and swap dependencies via app.dependency_overrides succinctly.
  • Keep the DB clean with transaction rollbacks or a dedicated engine; mock external APIs with tools like respx.
  • Validate OpenAPI with contract tests, and add pytest-benchmark or locust to build operational confidence.

Who Benefits from This

  • Learner A (senior thesis / solo dev)
    Wants a small testing starter kit. First, automate two paths for each API: success and failure.
  • Small Team B (3-person contract shop)
    Frequent spec changes. Use dependency overrides, DB rollbacks, and external API mocks to pinpoint breakage quickly.
  • SaaS Team C (startup)
    Scared of production incidents. Wants OpenAPI-compliant contract tests and CI-based benchmarking/load tests.

1. The Testing Map (Start with the Whole Picture)

  • Unit tests: verify function/service logic fast.
  • API tests: verify input/output for GET/POST (success/failure/boundaries).
  • Integration tests: verify behavior using real resources (DB/external APIs).
  • Contract tests: verify conformance to OpenAPI and reproducibility of examples.
  • Load & benchmarks: understand latency/throughput trends.

Decision points

  • Start with unit & API. Next, isolate external effects with dependency overrides. Finally add integration & load.

2. Minimal Setup (pytest basics)

Example pytest.ini:

# pytest.ini
[pytest]
testpaths = tests
python_files = test_*.py
addopts = -q

Example dependencies (requirements):

pytest
httpx
pytest-asyncio
respx            # mock for httpx
pytest-benchmark # optional: benchmarks

Example project layout:

app/
  main.py
  routers/
    articles.py
  core/
    settings.py
    security.py
  db/
    session.py
tests/
  conftest.py
  test_articles_api.py
  test_services_unit.py

3. Spinning Up the App in Tests (Sync vs. Async)

  • Mostly sync endpoints: fastapi.testclient.TestClient is convenient.
  • For async endpoints/WebSocket or many async def: prefer httpx.AsyncClient.

Sample API (subject under test):

# app/routers/articles.py
from fastapi import APIRouter, HTTPException, status
from pydantic import BaseModel

router = APIRouter(prefix="/articles", tags=["articles"])

class Article(BaseModel):
    id: int
    title: str

FAKE = {1: {"id": 1, "title": "hello"}}

@router.get("/{aid}", response_model=Article)
def get_article(aid: int):
    a = FAKE.get(aid)
    if not a:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="not found")
    return a

main.py:

# app/main.py
from fastapi import FastAPI
from app.routers import articles

app = FastAPI(title="Testable API")
app.include_router(articles.router)

4. Quick Sync API Checks with TestClient

# tests/test_articles_api.py
from fastapi.testclient import TestClient
from app.main import app

client = TestClient(app)

def test_get_article_ok():
    r = client.get("/articles/1")
    assert r.status_code == 200
    body = r.json()
    assert body["id"] == 1
    assert "title" in body

def test_get_article_404():
    r = client.get("/articles/999")
    assert r.status_code == 404
    assert r.json()["detail"] == "not found"

Decision points

  • Always write a minimal pair of success + failure. Add boundary cases (min/max/nonexistent) to harden coverage.

5. Async Testing with httpx.AsyncClient

# tests/test_async_api.py
import pytest
from httpx import AsyncClient
from app.main import app

@pytest.mark.asyncio
async def test_get_article_async():
    async with AsyncClient(app=app, base_url="http://test") as ac:
        res = await ac.get("/articles/1")
        assert res.status_code == 200

Decision points

  • AsyncClient(app=app) runs in embedded ASGI mode—no external server, fast inner loop.

6. Dependency Overrides (The Power of Depends)

  • Dependencies like DB sessions or auth can be swapped with app.dependency_overrides.
  • This lets you isolate API behavior from auth/permissions/DB.

Example: override an auth dependency

# app/core/security.py (assumed)
from fastapi import Depends, HTTPException, status

def get_current_user():
    # In reality you’d verify JWT, etc.
    raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="unauthorized")

On the test side:

# tests/conftest.py
import pytest
from fastapi.testclient import TestClient
from app.main import app
from app.core import security

@pytest.fixture(autouse=True)
def override_auth():
    def fake_user():
        return {"sub": "test-user", "role": "admin"}
    app.dependency_overrides[security.get_current_user] = fake_user
    yield
    app.dependency_overrides.clear()

@pytest.fixture
def client():
    return TestClient(app)

Decision points

  • Always clear overrides at test end to prevent leakage between tests.

7. DB Testing: Avoid Pollution with Transactions

Three strategies:

  1. In-memory DB (learning / ultra-fast)
  2. Test DB + transaction rollbacks (closer to prod)
  3. Run migrations first (Alembic)

SQLAlchemy 2.x example:

# tests/db_utils.py
import pytest
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from app.db.session import Base  # Declarative Base
from app.db.session import get_db # dependency function

@pytest.fixture(scope="session")
def engine():
    # To be closer to production, use a PostgreSQL test DB
    eng = create_engine("sqlite:///:memory:", connect_args={"check_same_thread": False})
    Base.metadata.create_all(eng)
    return eng

@pytest.fixture
def db_session(engine):
    connection = engine.connect()
    trans = connection.begin()
    Session = sessionmaker(bind=connection, autoflush=False, autocommit=False)
    session = Session()
    yield session
    session.close()
    trans.rollback()
    connection.close()

@pytest.fixture(autouse=True)
def override_db(db_session):
    from app.main import app
    def _get_db():
        try:
            yield db_session
        finally:
            pass
    app.dependency_overrides[get_db] = _get_db
    yield
    app.dependency_overrides.clear()

Decision points

  • Begin a transaction per test → rollback on teardown. Fast and reproducible.

8. Mocking External HTTP (httpx × respx)

Cut the network for tests that depend on external APIs.

# tests/test_external_calls.py
import respx
from httpx import Response
import pytest
from httpx import AsyncClient
from app.main import app

@respx.mock
@pytest.mark.asyncio
async def test_external_call():
    respx.get("https://api.example.com/users/1").mock(
        return_value=Response(200, json={"id": 1, "name": "Alice"})
    )
    async with AsyncClient(app=app, base_url="http://test") as ac:
        r = await ac.get("/proxy/users/1")  # app’s proxy endpoint assumed
        assert r.status_code == 200
        assert r.json()["name"] == "Alice"

Decision points

  • Mock both normal and failure paths (5xx/timeouts) to verify retries/fallbacks.

9. Testing AuthN/AuthZ (Scopes & Access)

  • Login API: success and failures (email mismatch, password mismatch, frozen user).
  • Protected routes: no token → 401, insufficient scope → 403, sufficient scope → 200.

Example:

def test_protected_401(client):
    res = client.get("/protected")
    assert res.status_code == 401

def test_protected_403(client, monkeypatch):
    from app.core import security
    def low_scope_user():
        return {"sub": "u", "scopes": {"articles:read"}}
    app = security  # alias
    from app.main import app as fastapi_app
    fastapi_app.dependency_overrides[security.get_current_user] = low_scope_user
    res = client.post("/articles")  # assume write scope is required
    assert res.status_code == 403

Decision points

  • Keep required scopes minimal per route. Tests should clarify scope differences via overrides.

10. Validation & Exception Handlers

  • Fix the shape of Pydantic errors and standardized error responses.
  • Distinguish 422 Unprocessable Entity (input invalid) from custom (business) errors.
def test_validation_422(client):
    res = client.post("/users", json={"email": "not-an-email"})
    assert res.status_code == 422
    body = res.json()
    assert "detail" in body

Decision points

  • If error shapes change, clients break. Snapshot tests can help pin them down.

11. Contract Tests Against OpenAPI

  • Schema integrity: required/typing/enum values.
  • Example reproducibility: do sample requests behave as documented?

Simple check:

# tests/test_openapi_contract.py
from app.main import app

def test_openapi_basic():
    spec = app.openapi()
    assert spec["info"]["title"] == "Testable API"
    paths = spec["paths"]
    assert "/articles/{aid}" in paths

More powerful approaches include tools that auto-generate tests from the schema (see references).


12. Property-Based Testing (Auto-explore Boundaries)

Use hypothesis to auto-generate boundary-adjacent values.

Example (unit):

# tests/test_services_unit.py
from hypothesis import given, strategies as st

def safe_div(a: int, b: int) -> float | None:
    if b == 0:
        return None
    return a / b

@given(st.integers(), st.integers())
def test_safe_division(a, b):
    r = safe_div(a, b)
    if b == 0:
        assert r is None
    else:
        assert isinstance(r, float)

Decision points

  • Best for logic-heavy service layers rather than the API layer.

13. Benchmarks & Load Testing

Benchmarks (unit/API micro):

# tests/test_bench.py
import pytest
from fastapi.testclient import TestClient
from app.main import app

client = TestClient(app)

@pytest.mark.benchmark(min_rounds=5)
def test_get_article_bench(benchmark):
    def do():
        assert client.get("/articles/1").status_code == 200
    benchmark(do)

Load tests (separate process):

  • Use locust or k6 to describe scenarios (auth → list → detail → create).
  • Focus metrics on latency distribution, error rate, and throughput (RED metrics).

Decision points

  • In CI, run short benchmarks to detect regressions; run load tests in a staging environment.

14. Testing WebSocket/SSE

  • WebSocket: use websockets/starlette.testclient to connect and verify send/receive.
  • SSE: connect with httpx and parse events from the stream.

WebSocket example:

# tests/test_ws.py
from starlette.testclient import TestClient
from app.main import app

def test_ws_echo():
    with TestClient(app).websocket_connect("/realtime/ws?room=test") as ws:
        ws.send_json({"msg": "hi"})
        data = ws.receive_json()
        assert "echo" in data

Decision points

  • Simulate heartbeats and abnormal disconnects; detect leaked connections.

15. Running in CI & Artifacts

  • Test run: pytest -q
  • Coverage: visualize core modules with coverage.
  • Artifacts: store failing logs, OpenAPI JSON, screenshots (as needed).
  • Parallelism: pytest -n auto (pytest-xdist) to speed up.

GitHub Actions (outline):

name: test
on: [push, pull_request]
jobs:
  unit:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with: { python-version: '3.11' }
      - run: pip install -r requirements.txt
      - run: pip install pytest pytest-asyncio respx pytest-benchmark
      - run: pytest -q

Decision points

  • Tests must run on every PR. Consider exporting OpenAPI and generating SDKs here too.

16. Common Pitfalls & Remedies

Symptom Cause Remedy
Tests are slow Hitting real DB/external APIs Dependency overrides/mocks, transaction rollbacks, parallel runs
Flaky tests Time/random/order dependencies Fixed seeds, time injection, I/O isolation, pytest-randomly
Spec drift Docs not updated OpenAPI contract tests, validate examples
Prod-only failures Config/network differences Separate .env, inject settings as dependencies, shorter timeouts
Cleanup leaks Unclosed sessions / leftover overrides yield fixtures, app.dependency_overrides.clear()

17. Adoption Roadmap

  1. For key APIs, write two tests each with TestClient: success and failure.
  2. Introduce dependency overrides to decouple auth/DB and gain speed.
  3. Keep the DB clean with transaction rollbacks; mock externals with respx.
  4. Add OpenAPI contract tests to catch schema/example breakage early.
  5. Gradually add benchmarks and load tests; integrate into CI.

References


Takeaways

  • Split your test layers, and first lock down API success/failure with small tests.
  • Use Depends overrides, DB rollbacks, and external mocks to make tests fast and resilient.
  • Put OpenAPI contract tests and light benchmarks into CI to catch spec and performance regressions early.
  • The goal is safe change. Write two tests today, then add dependency overrides next.

By greeden

Leave a Reply

Your email address will not be published. Required fields are marked *

日本語が含まれない投稿は無視されますのでご注意ください。(スパム対策)