FastAPI Unit Testing
Introduction
Unit testing is a critical part of software development that involves testing individual components or functions of your application in isolation. When working with FastAPI, proper unit testing helps ensure that your API endpoints, validation logic, and business functions work as expected.
In this guide, we'll explore how to write effective unit tests for FastAPI applications using pytest, which is the recommended testing framework for Python projects. By the end, you'll understand how to test different aspects of your FastAPI application and have confidence that your API works correctly.
Prerequisites
Before we begin, make sure you have:
- Basic knowledge of Python and FastAPI
- Python 3.7+ installed
- FastAPI and pytest installed
You can install the required packages using pip:
pip install fastapi pytest pytest-asyncio httpx
Setting Up Your Testing Environment
Project Structure
Let's start by looking at a typical project structure for a FastAPI application with tests:
my_fastapi_app/
├── app/
│ ├── __init__.py
│ ├── main.py
│ ├── models.py
│ ├── routers/
│ └── services/
├── tests/
│ ├── __init__.py
│ ├── conftest.py
│ ├── test_main.py
│ └── test_services.py
└── requirements.txt
Creating a Simple FastAPI Application
Let's create a simple FastAPI application that we'll test:
# app/main.py
from fastapi import FastAPI, HTTPException, Depends
from pydantic import BaseModel
from typing import List, Optional
app = FastAPI()
# Models
class Item(BaseModel):
id: Optional[int] = None
name: str
description: Optional[str] = None
price: float
# In-memory database
items_db = []
item_id_counter = 0
# Routes
@app.post("/items/", response_model=Item, status_code=201)
async def create_item(item: Item):
global item_id_counter
item_id_counter += 1
item.id = item_id_counter
items_db.append(item)
return item
@app.get("/items/", response_model=List[Item])
async def read_items():
return items_db
@app.get("/items/{item_id}", response_model=Item)
async def read_item(item_id: int):
for item in items_db:
if item.id == item_id:
return item
raise HTTPException(status_code=404, detail="Item not found")
Basic Unit Testing in FastAPI
Setting up the Test Client
FastAPI provides a TestClient class from fastapi.testclient that allows us to make requests to our application without running a server. Let's create our first test:
# tests/test_main.py
from fastapi.testclient import TestClient
from app.main import app
client = TestClient(app)
def test_create_item():
response = client.post(
"/items/",
json={"name": "Test Item", "description": "Test Description", "price": 10.5}
)
assert response.status_code == 201
data = response.json()
assert data["name"] == "Test Item"
assert data["description"] == "Test Description"
assert data["price"] == 10.5
assert "id" in data
Running Tests
To run the tests, use the pytest command:
pytest tests/test_main.py -v
This should output something like:
tests/test_main.py::test_create_item PASSED
Advanced Unit Testing Techniques
Using Fixtures with pytest
Fixtures in pytest help you set up and tear down resources for your tests. They are powerful tools for reusing test setup across multiple test functions.
Let's create a fixture that resets our database before each test:
# tests/conftest.py
import pytest
from app.main import items_db, item_id_counter
@pytest.fixture(autouse=True)
def reset_db():
# Clear the database before each test
items_db.clear()
global item_id_counter
item_id_counter = 0
Testing Multiple Endpoints
Now, let's write tests for our other endpoints:
# tests/test_main.py
from fastapi.testclient import TestClient
from app.main import app
client = TestClient(app)
def test_create_item():
response = client.post(
"/items/",
json={"name": "Test Item", "description": "Test Description", "price": 10.5}
)
assert response.status_code == 201
data = response.json()
assert data["name"] == "Test Item"
assert data["description"] == "Test Description"
assert data["price"] == 10.5
assert "id" in data
def test_read_items():
# First create some items
client.post("/items/", json={"name": "Item 1", "price": 5.5})
client.post("/items/", json={"name": "Item 2", "price": 10.5})
# Test reading all items
response = client.get("/items/")
assert response.status_code == 200
data = response.json()
assert len(data) == 2
assert data[0]["name"] == "Item 1"
assert data[1]["name"] == "Item 2"
def test_read_item():
# Create an item first
create_response = client.post(
"/items/",
json={"name": "Test Item", "price": 15.5}
)
item_id = create_response.json()["id"]
# Test reading the item by ID
response = client.get(f"/items/{item_id}")
assert response.status_code == 200
data = response.json()
assert data["name"] == "Test Item"
assert data["price"] == 15.5
assert data["id"] == item_id
def test_read_nonexistent_item():
response = client.get("/items/999")
assert response.status_code == 404
Testing Dependency Injection
FastAPI uses dependency injection for reusable components. Let's modify our app to include a dependency and then test it:
# app/main.py (modified version)
from fastapi import FastAPI, HTTPException, Depends
from pydantic import BaseModel
from typing import List, Optional
app = FastAPI()
# Models
class Item(BaseModel):
id: Optional[int] = None
name: str
description: Optional[str] = None
price: float
# In-memory database
items_db = []
item_id_counter = 0
# Dependencies
async def get_item_by_id(item_id: int):
for item in items_db:
if item.id == item_id:
return item
raise HTTPException(status_code=404, detail="Item not found")
# Routes
@app.post("/items/", response_model=Item, status_code=201)
async def create_item(item: Item):
global item_id_counter
item_id_counter += 1
item.id = item_id_counter
items_db.append(item)
return item
@app.get("/items/", response_model=List[Item])
async def read_items():
return items_db
@app.get("/items/{item_id}", response_model=Item)
async def read_item(item: Item = Depends(get_item_by_id)):
return item
Now let's test this dependency:
# tests/test_dependencies.py
import pytest
from fastapi.testclient import TestClient
from app.main import app, get_item_by_id, items_db
from fastapi import HTTPException
client = TestClient(app)
@pytest.mark.asyncio
async def test_get_item_by_id_exists():
# Add an item to the database
items_db.append({
"id": 1,
"name": "Dependency Test Item",
"price": 20.0
})
# Test the dependency directly
item = await get_item_by_id(1)
assert item["id"] == 1
assert item["name"] == "Dependency Test Item"
@pytest.mark.asyncio
async def test_get_item_by_id_not_exists():
with pytest.raises(HTTPException) as excinfo:
await get_item_by_id(999)
assert excinfo.value.status_code == 404
assert excinfo.value.detail == "Item not found"
Testing with Mocks
Sometimes, you want to isolate your unit tests from external dependencies like databases or third-party APIs. In these cases, you can use the unittest.mock module:
# app/services.py
import httpx
async def get_external_price(item_name: str):
async with httpx.AsyncClient() as client:
response = await client.get(f"https://api.prices.com/price/{item_name}")
if response.status_code == 200:
return response.json()["price"]
return None
# tests/test_services.py
import pytest
from unittest.mock import patch, AsyncMock
from app.services import get_external_price
@pytest.mark.asyncio
async def test_get_external_price():
# Create a mock response
mock_response = AsyncMock()
mock_response.status_code = 200
mock_response.json.return_value = {"price": 25.99}
# Patch the AsyncClient.get method
with patch("httpx.AsyncClient.get", return_value=mock_response):
price = await get_external_price("test_item")
assert price == 25.99
Real-World Example: Testing a User Authentication System
Let's create a more realistic example with user authentication:
# app/auth.py
from fastapi import Depends, FastAPI, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from pydantic import BaseModel
from typing import Optional
import jwt
from datetime import datetime, timedelta
# Secret key for JWT encoding
SECRET_KEY = "your-secret-key"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30
# User model
class User(BaseModel):
username: str
email: Optional[str] = None
full_name: Optional[str] = None
disabled: Optional[bool] = False
# Token model
class Token(BaseModel):
access_token: str
token_type: str
# Mock database
fake_users_db = {
"johndoe": {
"username": "johndoe",
"full_name": "John Doe",
"email": "[email protected]",
"disabled": False,
"password": "secret"
}
}
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
def get_user(db, username: str):
if username in db:
user_dict = db[username]
return User(**user_dict)
def authenticate_user(fake_db, username: str, password: str):
user = get_user(fake_db, username)
if not user:
return False
if fake_db[username]["password"] != password:
return False
return user
def create_access_token(data: dict, expires_delta: Optional[timedelta] = None):
to_encode = data.copy()
if expires_delta:
expire = datetime.utcnow() + expires_delta
else:
expire = datetime.utcnow() + timedelta(minutes=15)
to_encode.update({"exp": expire})
encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
return encoded_jwt
async def get_current_user(token: str = Depends(oauth2_scheme)):
credentials_exception = HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Could not validate credentials",
headers={"WWW-Authenticate": "Bearer"},
)
try:
payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
username: str = payload.get("sub")
if username is None:
raise credentials_exception
except jwt.PyJWTError:
raise credentials_exception
user = get_user(fake_users_db, username=username)
if user is None:
raise credentials_exception
return user
async def get_current_active_user(current_user: User = Depends(get_current_user)):
if current_user.disabled:
raise HTTPException(status_code=400, detail="Inactive user")
return current_user
# app/main.py (with auth added)
from fastapi import FastAPI, Depends, HTTPException
from fastapi.security import OAuth2PasswordRequestForm
from datetime import timedelta
from typing import List
from .auth import (
User, Token, fake_users_db, authenticate_user,
create_access_token, get_current_active_user,
ACCESS_TOKEN_EXPIRE_MINUTES
)
from .models import Item, items_db
app = FastAPI()
@app.post("/token", response_model=Token)
async def login_for_access_token(form_data: OAuth2PasswordRequestForm = Depends()):
user = authenticate_user(fake_users_db, form_data.username, form_data.password)
if not user:
raise HTTPException(
status_code=401,
detail="Incorrect username or password",
headers={"WWW-Authenticate": "Bearer"},
)
access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
access_token = create_access_token(
data={"sub": user.username}, expires_delta=access_token_expires
)
return {"access_token": access_token, "token_type": "bearer"}
@app.get("/users/me/", response_model=User)
async def read_users_me(current_user: User = Depends(get_current_active_user)):
return current_user
@app.get("/users/me/items/", response_model=List[Item])
async def read_own_items(current_user: User = Depends(get_current_active_user)):
# In a real app, you would filter items by user
return items_db
Now let's write tests for our authentication system:
# tests/test_auth.py
from fastapi.testclient import TestClient
import pytest
from app.main import app
from app.auth import fake_users_db
client = TestClient(app)
def test_login():
response = client.post(
"/token",
data={"username": "johndoe", "password": "secret"}
)
assert response.status_code == 200
data = response.json()
assert "access_token" in data
assert data["token_type"] == "bearer"
def test_login_incorrect_password():
response = client.post(
"/token",
data={"username": "johndoe", "password": "wrong"}
)
assert response.status_code == 401
assert response.json() == {"detail": "Incorrect username or password"}
def test_read_users_me():
# First login to get token
login_response = client.post(
"/token",
data={"username": "johndoe", "password": "secret"}
)
token = login_response.json()["access_token"]
# Use token to access protected endpoint
response = client.get(
"/users/me/",
headers={"Authorization": f"Bearer {token}"}
)
assert response.status_code == 200
data = response.json()
assert data["username"] == "johndoe"
assert data["email"] == "[email protected]"
def test_unauthorized_access():
response = client.get("/users/me/")
assert response.status_code == 401
response = client.get(
"/users/me/",
headers={"Authorization": "Bearer invalid-token"}
)
assert response.status_code == 401
Best Practices for FastAPI Unit Testing
-
Isolate tests: Each test should be independent and not rely on other tests.
-
Use fixtures: Fixtures help in setting up and tearing down test data.
-
Mock external dependencies: Use mocks to avoid calling external services during tests.
-
Test error cases: Test both successful scenarios and error conditions.
-
Test validation: Ensure that input validation works correctly.
-
Test authentication and authorization: Verify that protected routes can only be accessed with proper credentials.
-
Keep tests fast: Tests should run quickly to encourage frequent testing.
-
Use descriptive test names: Names should clearly indicate what is being tested.
Summary
In this guide, we covered the fundamentals of unit testing in FastAPI applications, including:
- Setting up a testing environment with pytest
- Creating a basic test client
- Testing API endpoints
- Working with fixtures and mocks
- Testing authentication and dependency injection
- Following best practices for effective unit tests
By applying these techniques to your FastAPI projects, you can ensure that your code is reliable, maintainable, and functions as expected.
Additional Resources
- FastAPI Testing Documentation
- Pytest Documentation
- FastAPI Testing Best Practices
- Python unittest.mock Library
Exercises
-
Write unit tests for a FastAPI application with CRUD operations for a "User" model.
-
Create a test suite that mocks a database connection using SQLAlchemy.
-
Implement tests for rate limiting middleware in a FastAPI application.
-
Write tests for file upload endpoints in FastAPI.
-
Create a CI/CD pipeline that runs FastAPI tests automatically on code changes.
If you spot any mistakes on this website, please let me know at [email protected]. I’d greatly appreciate your feedback! :)