📖 5 min read

Unit Testing FastAPI Applications#

This page shows how to write unit tests for FastAPI applications using pytest, covering synchronous and asynchronous test clients, database mocking, and testing JWT-protected endpoints.

Minimal FastAPI App to Test (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}

Sync Tests with TestClient#

Requires:

pip install pytest pytest-asyncio httpx fastapi

from fastapi.testclient import TestClient
from main import app

client = TestClient(app)


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"


def test_get_missing_user():
    response = client.get("/users/unknown")
    assert response.status_code == 404
    assert response.json() == {"detail": "User not found"}


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"


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"}

Run with:

pytest -v

Async Tests with httpx.AsyncClient#

For async FastAPI endpoints, use httpx.AsyncClient:

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"


@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"

Async Tests with Mocked Database#

For production apps using SQLAlchemy async sessions, override the dependency with an in-memory test database:

Setup (database.py)#

from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy.orm import sessionmaker, declarative_base

DATABASE_URL = "sqlite+aiosqlite:///./prod.db"

engine = create_async_engine(DATABASE_URL, echo=False, future=True)

async_session_maker = sessionmaker(
    engine, class_=AsyncSession, expire_on_commit=False
)

Base = declarative_base()

async def get_async_session():
    async with async_session_maker() as session:
        yield session

Test File with Dependency Override#

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
)


async def override_get_async_session():
    async with TestSessionLocal() as session:
        yield session


app.dependency_overrides[get_async_session] = override_get_async_session


@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)


@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():
    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"}

How this works:

  • Uses in-memory async SQLite — fast for tests, isolated from production

  • FastAPI’s Depends(get_async_session) is replaced with the test session

  • No real database I/O; everything runs in RAM

  • Tests are fully asynchronous with pytest.mark.asyncio

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)

FastAPI App with JWT Auth (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 Test Cases#

import pytest
from datetime import timedelta
from httpx import AsyncClient
from app.main import app
from app.auth import create_access_token


@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"


@pytest.mark.asyncio
async def test_expired_jwt():
    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"}


@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"


@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"}


@pytest.mark.asyncio
async def test_wrong_secret_jwt():
    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"}

Test Coverage Summary#

Test

Validates

test_valid_jwt

Proper access to protected routes

test_expired_jwt

Token expiration handling

test_missing_token

OAuth2 missing header logic

test_invalid_token

Token structure errors

test_wrong_secret_jwt

Signature mismatch handling