Skip to main content

Testing FastAPI Applications

Comprehensive guide to testing FastAPI applications using pytest, including unit tests, integration tests, test fixtures, and best practices.

Testing Setup​

Installation​

pip install pytest
pip install httpx  # For async testing
pip install pytest-asyncio  # For async test functions
pip install pytest-cov  # For code coverage

Project Structure​

myapp/
├── app/
│   ├── __init__.py
│   ├── main.py
│   ├── models.py
│   ├── schemas.py
│   └── database.py
├── tests/
│   ├── __init__.py
│   ├── conftest.py
│   ├── test_main.py
│   ├── test_auth.py
│   └── test_database.py
├── pytest.ini
└── requirements.txt

Basic Testing​

Test Client​

from fastapi.testclient import TestClient
from app.main import app

client = TestClient(app)

def test_read_root():
    response = client.get("/")
    assert response.status_code == 200
    assert response.json() == {"message": "Hello World"}

def test_read_item():
    response = client.get("/items/1")
    assert response.status_code == 200
    assert response.json() == {"item_id": 1}

def test_create_item():
    response = client.post(
        "/items/",
        json={"name": "Test Item", "price": 10.99}
    )
    assert response.status_code == 201
    assert response.json()["name"] == "Test Item"

Testing with Path Parameters​

def test_read_item_with_path_params():
    item_id = 42
    response = client.get(f"/items/{item_id}")

    assert response.status_code == 200
    data = response.json()
    assert data["item_id"] == item_id

def test_read_item_not_found():
    response = client.get("/items/999")
    assert response.status_code == 404

Testing with Query Parameters​

def test_read_items_with_pagination():
    response = client.get("/items/?skip=0&limit=10")

    assert response.status_code == 200
    data = response.json()
    assert len(data) <= 10

def test_search_items():
    response = client.get("/items/search/?q=test")

    assert response.status_code == 200
    assert isinstance(response.json(), list)

Testing with Fixtures​

Using pytest Fixtures​

conftest.py:

import pytest
from fastapi.testclient import TestClient
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from app.main import app
from app.database import Base, get_db

# Test database
SQLALCHEMY_DATABASE_URL = "sqlite:///./test.db"

engine = create_engine(
    SQLALCHEMY_DATABASE_URL,
    connect_args={"check_same_thread": False}
)

TestingSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

@pytest.fixture
def test_db():
    Base.metadata.create_all(bind=engine)
    yield
    Base.metadata.drop_all(bind=engine)

@pytest.fixture
def db_session(test_db):
    connection = engine.connect()
    transaction = connection.begin()
    session = TestingSessionLocal(bind=connection)
    yield session
    session.close()
    transaction.rollback()
    connection.close()

@pytest.fixture
def client(db_session):
    def override_get_db():
        try:
            yield db_session
        finally:
            db_session.close()

    app.dependency_overrides[get_db] = override_get_db
    yield TestClient(app)
    app.dependency_overrides.clear()

Using Fixtures in Tests​

def test_create_user(client, db_session):
    response = client.post(
        "/users/",
        json={
            "username": "testuser",
            "email": "test@example.com",
            "password": "password123"
        }
    )

    assert response.status_code == 201
    data = response.json()
    assert data["username"] == "testuser"
    assert "password" not in data

def test_read_user(client, db_session):
    # Create user first
    client.post(
        "/users/",
        json={"username": "testuser", "email": "test@example.com", "password": "pass"}
    )

    # Test reading user
    response = client.get("/users/1")
    assert response.status_code == 200
    assert response.json()["username"] == "testuser"

Testing Authentication​

Testing Login​

def test_login_success(client):
    # Create user
    client.post(
        "/register/",
        json={"username": "testuser", "password": "password123"}
    )

    # Test login
    response = client.post(
        "/token",
        data={"username": "testuser", "password": "password123"}
    )

    assert response.status_code == 200
    data = response.json()
    assert "access_token" in data
    assert data["token_type"] == "bearer"

def test_login_invalid_credentials(client):
    response = client.post(
        "/token",
        data={"username": "nonexistent", "password": "wrongpass"}
    )

    assert response.status_code == 401

Testing Protected Endpoints​

@pytest.fixture
def auth_client(client):
    # Create and login user
    client.post(
        "/register/",
        json={"username": "testuser", "password": "password123"}
    )

    response = client.post(
        "/token",
        data={"username": "testuser", "password": "password123"}
    )

    token = response.json()["access_token"]

    # Add authorization header to client
    client.headers = {
        **client.headers,
        "Authorization": f"Bearer {token}"
    }

    return client

def test_protected_endpoint_without_auth(client):
    response = client.get("/users/me")
    assert response.status_code == 401

def test_protected_endpoint_with_auth(auth_client):
    response = auth_client.get("/users/me")
    assert response.status_code == 200
    assert response.json()["username"] == "testuser"

Async Testing​

Testing Async Endpoints​

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

@pytest.mark.asyncio
async def test_async_endpoint():
    async with AsyncClient(app=app, base_url="http://test") as ac:
        response = await ac.get("/async-items/")

    assert response.status_code == 200
    assert isinstance(response.json(), list)

@pytest.mark.asyncio
async def test_async_create_item():
    async with AsyncClient(app=app, base_url="http://test") as ac:
        response = await ac.post(
            "/items/",
            json={"name": "Test", "price": 10.0}
        )

    assert response.status_code == 201

Async Database Testing​

import pytest
from httpx import AsyncClient
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession

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

@pytest.mark.asyncio
async def test_async_create_user(async_client):
    response = await async_client.post(
        "/users/",
        json={"username": "testuser", "email": "test@example.com"}
    )

    assert response.status_code == 201

Mocking Dependencies​

Overriding Dependencies​

from unittest.mock import Mock

def test_with_mocked_dependency(client):
    # Mock dependency
    def mock_get_current_user():
        return {"username": "mockuser", "id": 1}

    app.dependency_overrides[get_current_user] = mock_get_current_user

    response = client.get("/users/me")

    assert response.status_code == 200
    assert response.json()["username"] == "mockuser"

    # Clean up
    app.dependency_overrides.clear()

Mocking External Services​

from unittest.mock import patch, MagicMock

def test_external_api_call(client):
    with patch('app.services.external_api.fetch_data') as mock_fetch:
        mock_fetch.return_value = {"data": "mocked"}

        response = client.get("/external-data/")

        assert response.status_code == 200
        assert response.json() == {"data": "mocked"}
        mock_fetch.assert_called_once()

Testing File Uploads​

Testing File Upload Endpoints​

from io import BytesIO

def test_file_upload(client):
    file_content = b"test file content"
    files = {"file": ("test.txt", BytesIO(file_content), "text/plain")}

    response = client.post("/upload/", files=files)

    assert response.status_code == 200
    assert "filename" in response.json()

def test_multiple_file_upload(client):
    files = [
        ("files", ("file1.txt", BytesIO(b"content1"), "text/plain")),
        ("files", ("file2.txt", BytesIO(b"content2"), "text/plain"))
    ]

    response = client.post("/upload-multiple/", files=files)

    assert response.status_code == 200
    assert len(response.json()["files"]) == 2

Testing Validation​

Testing Request Validation​

def test_create_item_invalid_data(client):
    response = client.post(
        "/items/",
        json={"name": "Test"}  # Missing required 'price' field
    )

    assert response.status_code == 422
    errors = response.json()["detail"]
    assert any(error["loc"] == ["body", "price"] for error in errors)

def test_create_item_invalid_type(client):
    response = client.post(
        "/items/",
        json={"name": "Test", "price": "not-a-number"}
    )

    assert response.status_code == 422

Testing Response Validation​

def test_response_model_filtering(client):
    # Create user with password
    response = client.post(
        "/users/",
        json={"username": "test", "password": "secret123"}
    )

    # Password should not be in response
    assert response.status_code == 201
    data = response.json()
    assert "password" not in data
    assert "hashed_password" not in data

Parametrized Tests​

Testing Multiple Scenarios​

@pytest.mark.parametrize("item_id,expected_status", [
    (1, 200),
    (2, 200),
    (999, 404),
    (-1, 422),
])
def test_read_item_various_ids(client, item_id, expected_status):
    response = client.get(f"/items/{item_id}")
    assert response.status_code == expected_status

@pytest.mark.parametrize("username,password,expected_status", [
    ("validuser", "validpass123", 200),
    ("short", "pass", 422),  # Too short
    ("", "", 422),  # Empty
    ("a" * 100, "pass", 422),  # Too long
])
def test_user_registration(client, username, password, expected_status):
    response = client.post(
        "/register/",
        json={"username": username, "password": password}
    )
    assert response.status_code == expected_status

Integration Tests​

End-to-End Testing​

def test_user_flow(client, db_session):
    # 1. Register user
    register_response = client.post(
        "/register/",
        json={"username": "testuser", "email": "test@example.com", "password": "pass123"}
    )
    assert register_response.status_code == 201

    # 2. Login
    login_response = client.post(
        "/token",
        data={"username": "testuser", "password": "pass123"}
    )
    assert login_response.status_code == 200
    token = login_response.json()["access_token"]

    # 3. Access protected endpoint
    client.headers = {"Authorization": f"Bearer {token}"}
    profile_response = client.get("/users/me")
    assert profile_response.status_code == 200
    assert profile_response.json()["username"] == "testuser"

    # 4. Update profile
    update_response = client.put(
        "/users/me",
        json={"email": "newemail@example.com"}
    )
    assert update_response.status_code == 200

Testing Error Handling​

Testing Exception Handlers​

def test_http_exception_handler(client):
    response = client.get("/items/999")
    assert response.status_code == 404
    assert "detail" in response.json()

def test_validation_exception(client):
    response = client.post("/items/", json={"invalid": "data"})
    assert response.status_code == 422
    assert "detail" in response.json()

Code Coverage​

Running Coverage​

# Run tests with coverage
pytest --cov=app tests/

# Generate HTML coverage report
pytest --cov=app --cov-report=html tests/

# Show missing lines
pytest --cov=app --cov-report=term-missing tests/

Coverage Configuration​

pytest.ini:

[pytest]
testpaths = tests
python_files = test_*.py
python_classes = Test*
python_functions = test_*
addopts =
    --cov=app
    --cov-report=term-missing
    --cov-report=html
    --strict-markers
    -v

[coverage:run]
omit =
    */tests/*
    */venv/*
    */__pycache__/*

Performance Testing​

Load Testing with Locust​

pip install locust

locustfile.py:

from locust import HttpUser, task, between

class APIUser(HttpUser):
    wait_time = between(1, 3)

    def on_start(self):
        # Login
        response = self.client.post(
            "/token",
            data={"username": "testuser", "password": "password123"}
        )
        self.token = response.json()["access_token"]
        self.client.headers = {"Authorization": f"Bearer {self.token}"}

    @task(3)
    def read_items(self):
        self.client.get("/items/")

    @task(1)
    def create_item(self):
        self.client.post(
            "/items/",
            json={"name": "Test", "price": 10.0}
        )

    @task(2)
    def read_item(self):
        self.client.get("/items/1")

Run: locust -f locustfile.py

Test Utilities​

Custom Test Fixtures​

@pytest.fixture
def sample_user():
    return {
        "username": "testuser",
        "email": "test@example.com",
        "password": "password123"
    }

@pytest.fixture
def sample_items():
    return [
        {"name": "Item 1", "price": 10.0},
        {"name": "Item 2", "price": 20.0},
        {"name": "Item 3", "price": 30.0}
    ]

def test_with_fixtures(client, sample_user, sample_items):
    # Use fixtures in test
    user_response = client.post("/users/", json=sample_user)
    assert user_response.status_code == 201

    for item in sample_items:
        item_response = client.post("/items/", json=item)
        assert item_response.status_code == 201

Test Helpers​

def create_test_user(client, username="testuser", password="password123"):
    """Helper function to create a test user"""
    response = client.post(
        "/register/",
        json={"username": username, "password": password}
    )
    return response.json()

def get_auth_token(client, username="testuser", password="password123"):
    """Helper function to get authentication token"""
    response = client.post(
        "/token",
        data={"username": username, "password": password}
    )
    return response.json()["access_token"]

def test_with_helpers(client):
    create_test_user(client)
    token = get_auth_token(client)

    client.headers = {"Authorization": f"Bearer {token}"}
    response = client.get("/users/me")
    assert response.status_code == 200

Best Practices​

  1. Isolate Tests: Each test should be independent
  2. Use Fixtures: Reuse common setup code with fixtures
  3. Test Database: Use a separate test database
  4. Mock External Services: Don't depend on external APIs in tests
  5. Test Edge Cases: Test boundary conditions and error cases
  6. Descriptive Names: Use clear, descriptive test names
  7. Coverage Goals: Aim for high test coverage (80%+)
  8. Fast Tests: Keep tests fast by using in-memory databases
  9. CI/CD Integration: Run tests automatically in CI/CD pipeline
  10. Organize Tests: Structure tests to mirror application structure

Running Tests​

# Run all tests
pytest

# Run specific test file
pytest tests/test_main.py

# Run specific test
pytest tests/test_main.py::test_read_root

# Run with verbose output
pytest -v

# Run with coverage
pytest --cov=app

# Run and stop on first failure
pytest -x

# Run tests matching pattern
pytest -k "test_user"

Testing is crucial for building reliable FastAPI applications. A comprehensive test suite gives you confidence to refactor and add features safely.