Testing and Middleware

This document covers testing FastAPI applications with Pytest and TestClient, async testing patterns, mocking database dependencies, JWT authentication testing, and middleware implementation.

---

📚 Table of Contents

1. Part 1: Getting Started with Testing
2. Part 2: Minimal Unit Tests
3. Part 3: Async FastAPI Tests
4. Part 4: Testing with Mocked Database
5. Part 5: JWT Authentication Tests
6. Part 6: Test Folder Structure
7. Part 7: Custom Middleware
8. Part 8: Error Handling

---

Part 1: Getting Started with Testing

Why Test?

• Catch bugs early
• Ensure code works as expected
• Enable safe refactoring
• Document expected behavior

Required Packages

pip install pytest pytest-asyncio httpx fastapi

Key Testing Tools

Tool	Purpose
pytest	Test runner and framework
pytest-asyncio	Async test support
TestClient	Sync API testing
httpx.AsyncClient	Async API testing

---

Part 2: Minimal Unit Tests

Example FastAPI App (main.py)

from fastapi import FastAPI, HTTPException

app = FastAPI()

fake_db = {"alice": "Engineer", "bob": "Designer"}

@app.get("/users/{username}")
def get_user(username: str):
    if username not in fake_db:
        raise HTTPException(status_code=404, detail="User not found")
    return {"username": username, "job": fake_db[username]}

@app.post("/users")
def create_user(username: str, job: str):
    if username in fake_db:
        raise HTTPException(status_code=400, detail="User already exists")
    fake_db[username] = job
    return {"message": "User created", "username": username}

Unit Tests (test_main.py)

from fastapi.testclient import TestClient
from main import app

client = TestClient(app)

# Test GET endpoint - Success
def test_get_existing_user():
    response = client.get("/users/alice")
    assert response.status_code == 200
    data = response.json()
    assert data["username"] == "alice"
    assert data["job"] == "Engineer"

# Test GET endpoint - Not Found
def test_get_missing_user():
    response = client.get("/users/unknown")
    assert response.status_code == 404
    assert response.json() == {"detail": "User not found"}

# Test POST endpoint - Success
def test_create_new_user():
    response = client.post("/users", params={"username": "charlie", "job": "Manager"})
    assert response.status_code == 200
    assert response.json()["message"] == "User created"
    assert response.json()["username"] == "charlie"

# Test POST endpoint - Already Exists
def test_create_existing_user():
    response = client.post("/users", params={"username": "alice", "job": "Engineer"})
    assert response.status_code == 400
    assert response.json() == {"detail": "User already exists"}

Running Tests

pytest -v

---

Part 3: Async FastAPI Tests

For async endpoints, use httpx.AsyncClient instead of TestClient.

Async FastAPI App

from fastapi import FastAPI, HTTPException

app = FastAPI()

fake_db = {"alice": "Engineer", "bob": "Designer"}

@app.get("/users/{username}")
async def get_user(username: str):
    if username not in fake_db:
        raise HTTPException(status_code=404, detail="User not found")
    return {"username": username, "job": fake_db[username]}

@app.post("/users")
async def create_user(username: str, job: str):
    if username in fake_db:
        raise HTTPException(status_code=400, detail="User already exists")
    fake_db[username] = job
    return {"message": "User created", "username": username}

Async Tests (test_main_async.py)

import pytest
from httpx import AsyncClient
from main import app

@pytest.mark.asyncio
async def test_get_existing_user():
    async with AsyncClient(app=app, base_url="http://test") as client:
        response = await client.get("/users/alice")
    assert response.status_code == 200
    data = response.json()
    assert data["username"] == "alice"
    assert data["job"] == "Engineer"

@pytest.mark.asyncio
async def test_get_missing_user():
    async with AsyncClient(app=app, base_url="http://test") as client:
        response = await client.get("/users/unknown")
    assert response.status_code == 404
    assert response.json() == {"detail": "User not found"}

@pytest.mark.asyncio
async def test_create_new_user():
    async with AsyncClient(app=app, base_url="http://test") as client:
        response = await client.post(
            "/users", params={"username": "charlie", "job": "Manager"}
        )
    assert response.status_code == 200
    assert response.json()["message"] == "User created"

@pytest.mark.asyncio
async def test_create_existing_user():
    async with AsyncClient(app=app, base_url="http://test") as client:
        response = await client.post(
            "/users", params={"username": "alice", "job": "Engineer"}
        )
    assert response.status_code == 400
    assert response.json() == {"detail": "User already exists"}

---

Part 4: Testing with Mocked Database

Project Structure

app/
├── database.py      # DB engine and session
├── models.py        # SQLAlchemy models
└── main.py          # FastAPI app
tests/
└── test_users_async_db.py

FastAPI App with Async DB Dependency

# app/main.py
from fastapi import FastAPI, Depends, HTTPException
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select
from app.database import get_async_session
from app.models import User

app = FastAPI()

@app.get("/users/{username}")
async def get_user(username: str, session: AsyncSession = Depends(get_async_session)):
    result = await session.execute(select(User).where(User.username == username))
    user = result.scalars().first()
    if not user:
        raise HTTPException(status_code=404, detail="User not found")
    return {"username": user.username, "job": user.job}

@app.post("/users")
async def create_user(
    username: str,
    job: str,
    session: AsyncSession = Depends(get_async_session),
):
    result = await session.execute(select(User).where(User.username == username))
    if result.scalars().first():
        raise HTTPException(status_code=400, detail="User already exists")
    new_user = User(username=username, job=job)
    session.add(new_user)
    await session.commit()
    return {"message": "User created", "username": username}

Test with Mocked Database

# tests/test_users_async_db.py
import pytest
from httpx import AsyncClient
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker

from app.main import app
from app.database import Base, get_async_session
from app.models import User

# Create TEST DB (in memory)
TEST_DATABASE_URL = "sqlite+aiosqlite:///:memory:"

test_engine = create_async_engine(
    TEST_DATABASE_URL, echo=False, future=True
)

TestSessionLocal = sessionmaker(
    test_engine, expire_on_commit=False, class_=AsyncSession
)

# Dependency override for tests
async def override_get_async_session():
    async with TestSessionLocal() as session:
        yield session

app.dependency_overrides[get_async_session] = override_get_async_session

# Create DB schema before tests
@pytest.fixture(scope="module", autouse=True)
async def prepare_database():
    async with test_engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    yield
    async with test_engine.begin() as conn:
        await conn.run_sync(Base.metadata.drop_all)

# Tests
@pytest.mark.asyncio
async def test_create_user():
    async with AsyncClient(app=app, base_url="http://test") as client:
        response = await client.post(
            "/users", params={"username": "alice", "job": "Engineer"}
        )
    assert response.status_code == 200
    assert response.json()["message"] == "User created"

@pytest.mark.asyncio
async def test_get_user():
    # Insert user manually into mocked DB
    async with TestSessionLocal() as session:
        session.add(User(username="bob", job="Designer"))
        await session.commit()

    async with AsyncClient(app=app, base_url="http://test") as client:
        response = await client.get("/users/bob")
    assert response.status_code == 200
    assert response.json() == {"username": "bob", "job": "Designer"}

@pytest.mark.asyncio
async def test_missing_user():
    async with AsyncClient(app=app, base_url="http://test") as client:
        response = await client.get("/users/unknown")
    assert response.status_code == 404
    assert response.json() == {"detail": "User not found"}

---

Part 5: JWT Authentication Tests

JWT Utility (auth.py)

from datetime import datetime, timedelta
from typing import Optional
import jwt

SECRET_KEY = "TEST_SECRET_KEY"
ALGORITHM = "HS256"

def create_access_token(data: dict, expires_delta: Optional[timedelta] = None):
    to_encode = data.copy()
    expire = datetime.utcnow() + (expires_delta or timedelta(minutes=15))
    to_encode.update({"exp": expire})
    return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)

Secured FastAPI Endpoint

# app/main.py
from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
import jwt

from app.auth import SECRET_KEY, ALGORITHM

app = FastAPI()
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

def verify_token(token: str = Depends(oauth2_scheme)):
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        return payload
    except jwt.ExpiredSignatureError:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Token expired"
        )
    except jwt.InvalidTokenError:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Invalid token"
        )

@app.get("/secure")
async def secure_endpoint(payload: dict = Depends(verify_token)):
    return {"message": "Access granted", "payload": payload}

JWT Authentication Tests

# tests/test_jwt_auth.py
import pytest
from datetime import timedelta
from httpx import AsyncClient
import jwt

from app.main import app
from app.auth import create_access_token

# Test valid JWT
@pytest.mark.asyncio
async def test_valid_jwt():
    token = create_access_token({"sub": "alice"}, expires_delta=timedelta(minutes=5))

    async with AsyncClient(app=app, base_url="http://test") as client:
        response = await client.get(
            "/secure",
            headers={"Authorization": f"Bearer {token}"}
        )
    assert response.status_code == 200
    assert response.json()["message"] == "Access granted"
    assert response.json()["payload"]["sub"] == "alice"

# Test expired JWT
@pytest.mark.asyncio
async def test_expired_jwt():
    # Token expired 1 minute ago
    token = create_access_token(
        {"sub": "user"},
        expires_delta=timedelta(minutes=-1)
    )

    async with AsyncClient(app=app, base_url="http://test") as client:
        response = await client.get(
            "/secure",
            headers={"Authorization": f"Bearer {token}"}
        )
    assert response.status_code == 401
    assert response.json() == {"detail": "Token expired"}

# Test missing token
@pytest.mark.asyncio
async def test_missing_token():
    async with AsyncClient(app=app, base_url="http://test") as client:
        response = await client.get("/secure")
    assert response.status_code == 401
    assert response.json()["detail"] == "Not authenticated"

# Test invalid token
@pytest.mark.asyncio
async def test_invalid_token():
    invalid_token = "abc.def.ghi"

    async with AsyncClient(app=app, base_url="http://test") as client:
        response = await client.get(
            "/secure",
            headers={"Authorization": f"Bearer {invalid_token}"}
        )
    assert response.status_code == 401
    assert response.json() == {"detail": "Invalid token"}

# Test wrong secret key
@pytest.mark.asyncio
async def test_wrong_secret_jwt():
    # Token signed with a different key
    fake_token = jwt.encode({"sub": "hacker"}, "WRONG_KEY", algorithm="HS256")

    async with AsyncClient(app=app, base_url="http://test") as client:
        response = await client.get(
            "/secure",
            headers={"Authorization": f"Bearer {fake_token}"}
        )
    assert response.status_code == 401
    assert response.json() == {"detail": "Invalid token"}

JWT Test Summary

Test	Purpose
test_valid_jwt	Token validation works correctly
test_expired_jwt	Token expiration is handled
test_missing_token	Missing Authorization header returns 401
test_invalid_token	Malformed tokens are rejected
test_wrong_secret_jwt	Wrong signature is detected

---

Part 6: Test Folder Structure

Recommended Structure

tests/
├── __init__.py
├── conftest.py          # Global pytest fixtures
├── factories/           # Data builders / model factories
│   ├── __init__.py
│   └── user_factory.py
├── utils/               # Shared testing utilities
│   ├── __init__.py
│   └── jwt_helpers.py
├── unit/                # Pure unit tests (no DB, no FastAPI)
│   ├── __init__.py
│   └── test_helpers.py
├── integration/         # API + DB tests
│   ├── __init__.py
│   ├── test_users.py
│   └── test_auth.py
└── e2e/                 # End-to-end tests
    ├── __init__.py
    └── test_full_flow.py

conftest.py - Global Fixtures

import pytest
from httpx import AsyncClient
from app.main import app

@pytest.fixture
async def client():
    async with AsyncClient(app=app, base_url="http://test") as c:
        yield c

factories/user_factory.py

def user_payload(username="alice", job="Engineer"):
    return {"username": username, "job": job}

utils/jwt_helpers.py

from app.auth import create_access_token

def get_token(user="test_user"):
    return create_access_token({"sub": user})

unit/test_helpers.py

from app.utils.math import add

def test_add():
    assert add(2, 3) == 5

integration/test_users.py

import pytest

@pytest.mark.asyncio
async def test_create_user(client):
    response = await client.post("/users", params={"username": "bob", "job": "Manager"})
    assert response.status_code == 200

e2e/test_full_flow.py

@pytest.mark.asyncio
async def test_user_full_flow(client):
    # Create user
    res = await client.post("/users", params={"username": "eva", "job": "Dev"})
    assert res.status_code == 200

    # Fetch user
    res = await client.get("/users/eva")
    assert res.json()["job"] == "Dev"

Folder Summary

Folder	Purpose
unit/	Fast, isolated logic tests
integration/	DB + API + Async tests
e2e/	Full workflow scenarios
factories/	Model and payload generators
utils/	Reusable test helpers
conftest.py	Global fixtures

---

Part 7: Custom Middleware

Middleware executes code before and after each request.

Request Logging Middleware

from fastapi import FastAPI, Request
import time

app = FastAPI()

@app.middleware("http")
async def log_requests(request: Request, call_next):
    start_time = time.time()

    # Before request
    print(f"Request: {request.method} {request.url}")

    # Call the next middleware/endpoint
    response = await call_next(request)

    # After request
    process_time = time.time() - start_time
    print(f"Response: {response.status_code} - Time: {process_time:.4f}s")

    # Add custom header
    response.headers["X-Process-Time"] = str(process_time)

    return response

Adding Request ID Middleware

import uuid

@app.middleware("http")
async def add_request_id(request: Request, call_next):
    request_id = str(uuid.uuid4())
    request.state.request_id = request_id

    response = await call_next(request)
    response.headers["X-Request-ID"] = request_id

    return response

Middleware Order

Middleware is executed in the order it's added. The first middleware added is the outermost layer.

---

Part 8: Error Handling

Custom Exception Handler

from fastapi import FastAPI, HTTPException, Request
from fastapi.responses import JSONResponse

app = FastAPI()

class CustomException(Exception):
    def __init__(self, name: str, detail: str):
        self.name = name
        self.detail = detail

@app.exception_handler(CustomException)
async def custom_exception_handler(request: Request, exc: CustomException):
    return JSONResponse(
        status_code=400,
        content={"error": exc.name, "detail": exc.detail}
    )

@app.exception_handler(HTTPException)
async def http_exception_handler(request: Request, exc: HTTPException):
    return JSONResponse(
        status_code=exc.status_code,
        content={"error": "HTTPException", "detail": exc.detail}
    )

@app.get("/error")
async def raise_error():
    raise CustomException(name="TestError", detail="This is a test error")

Global Exception Handler

from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
import traceback

app = FastAPI()

@app.exception_handler(Exception)
async def global_exception_handler(request: Request, exc: Exception):
    # Log the error
    traceback.print_exc()

    return JSONResponse(
        status_code=500,
        content={
            "error": "Internal Server Error",
            "detail": "An unexpected error occurred"
        }
    )

---

📖 References

Resource	Link	Duration
FastAPI Testing Tutorial	fastapi.tiangolo.com	~20 min
FastAPI Middleware	fastapi.tiangolo.com	~10 min
FastAPI Error Handling	fastapi.tiangolo.com	~10 min
Pytest Documentation	docs.pytest.org	~30 min

Learning Checklist

• Write unit tests using Pytest and verify endpoints with TestClient
• Test both success and failure cases for API routes
• Implement async tests using httpx.AsyncClient
• Mock database dependencies for isolated testing
• Test JWT authentication (valid, expired, invalid, missing tokens)
• Organize tests in a proper folder structure
• Implement custom middleware for logging requests
• Add error handlers for HTTPException and custom exceptions
• Ensure middleware doesn't block async execution
• Document testing strategy and include coverage for edge cases

---

Summary

This document covers:

1. Basic Testing: TestClient for sync endpoints
2. Async Testing: httpx.AsyncClient with pytest-asyncio
3. Database Mocking: In-memory SQLite with dependency overrides
4. JWT Testing: Authentication scenarios (valid, expired, invalid)
5. Test Structure: Organized folders for unit, integration, and e2e tests
6. Middleware: Request logging and custom headers
7. Error Handling: Custom exception handlers