Testing FastAPI Applications: Unit Tests to Integration

Testing FastAPI Applications: Unit Tests to Integration

Testing is a critical part of building reliable web applications. FastAPI, with its modern design and built-in testing utilities, makes it straightforward to test everything from individual functions to full HTTP endpoints. This tutorial walks you through the complete testing journey — from simple unit tests to comprehensive integration tests — so you can ship FastAPI applications with confidence.

Why Testing Matters in FastAPI

FastAPI applications often handle business logic, database operations, authentication, and external API calls. Without tests, small changes can silently break critical functionality. A well-structured test suite helps you:

• Catch regressions before they reach production
• Document expected behavior for future developers
• Refactor code with confidence
• Validate API contracts and response schemas
• Improve overall code design through testability

FastAPI is built on Starlette, which includes a powerful TestClient that lets you simulate HTTP requests without running an actual server. Combined with pytest, this creates an excellent testing workflow.

Project Setup

Let's start by setting up a sample FastAPI project with the necessary testing dependencies. Create the following project structure:

fastapi_app/
├── app/
│   ├── __init__.py
│   ├── main.py
│   ├── models.py
│   ├── database.py
│   └── services.py
├── tests/
│   ├── __init__.py
│   ├── conftest.py
│   ├── test_unit.py
│   └── test_integration.py
├── requirements.txt
└── pytest.ini

Install the required dependencies:

fastapi==0.104.1
uvicorn==0.24.0
sqlalchemy==2.0.23
pydantic==2.5.2
pytest==7.4.3
httpx==0.25.1
pytest-asyncio==0.21.1

The httpx package is required because FastAPI's TestClient uses it under the hood to make requests against your application in memory.

Building the Sample Application

Before writing tests, we need an application to test. Let's build a simple task management API with SQLite as the database. First, define the database connection in app/database.py:

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, declarative_base

SQLALCHEMY_DATABASE_URL = "sqlite:///./test.db"

engine = create_engine(
    SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False}
)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()


def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

Next, define the data models in app/models.py:

from sqlalchemy import Column, Integer, String, Boolean
from pydantic import BaseModel
from app.database import Base


class Task(Base):
    __tablename__ = "tasks"

    id = Column(Integer, primary_key=True, index=True)
    title = Column(String, nullable=False)
    description = Column(String, default="")
    completed = Column(Boolean, default=False)


class TaskCreate(BaseModel):
    title: str
    description: str = ""
    completed: bool = False


class TaskResponse(BaseModel):
    id: int
    title: str
    description: str
    completed: bool

    class Config:
        from_attributes = True

Now create a service layer in app/services.py that contains the business logic, separated from the HTTP layer:

from sqlalchemy.orm import Session
from app.models import Task, TaskCreate


def get_task(db: Session, task_id: int):
    return db.query(Task).filter(Task.id == task_id).first()


def get_tasks(db: Session, skip: int = 0, limit: int = 100):
    return db.query(Task).offset(skip).limit(limit).all()


def create_task(db: Session, task: TaskCreate):
    db_task = Task(
        title=task.title,
        description=task.description,
        completed=task.completed,
    )
    db.add(db_task)
    db.commit()
    db.refresh(db_task)
    return db_task


def mark_complete(db: Session, task_id: int):
    task = get_task(db, task_id)
    if task is None:
        return None
    task.completed = True
    db.commit()
    db.refresh(task)
    return task


def delete_task(db: Session, task_id: int):
    task = get_task(db, task_id)
    if task is None:
        return False
    db.delete(task)
    db.commit()
    return True

Finally, wire everything together in app/main.py:

from fastapi import FastAPI, Depends, HTTPException
from sqlalchemy.orm import Session
from app.database import engine, Base, get_db
from app.models import TaskCreate, TaskResponse
from app import services

Base.metadata.create_all(bind=engine)

app = FastAPI(title="Task Manager API")


@app.post("/tasks/", response_model=TaskResponse)
def create_task(task: TaskCreate, db: Session = Depends(get_db)):
    return services.create_task(db, task)


@app.get("/tasks/", response_model=list[TaskResponse])
def list_tasks(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
    return services.get_tasks(db, skip=skip, limit=limit)


@app.get("/tasks/{task_id}", response_model=TaskResponse)
def get_task(task_id: int, db: Session = Depends(get_db)):
    task = services.get_task(db, task_id)
    if task is None:
        raise HTTPException(status_code=404, detail="Task not found")
    return task


@app.patch("/tasks/{task_id}/complete", response_model=TaskResponse)
def complete_task(task_id: int, db: Session = Depends(get_db)):
    task = services.mark_complete(db, task_id)
    if task is None:
        raise HTTPException(status_code=404, detail="Task not found")
    return task


@app.delete("/tasks/{task_id}")
def delete_task(task_id: int, db: Session = Depends(get_db)):
    deleted = services.delete_task(db, task_id)
    if not deleted:
        raise HTTPException(status_code=404, detail="Task not found")
    return {"message": "Task deleted"}

Configuring pytest

Create a pytest.ini file at the project root to configure pytest behavior:

[pytest]
testpaths = tests
python_files = test_*.py
python_functions = test_*
asyncio_mode = auto

Writing Unit Tests

Unit tests focus on testing individual functions or components in isolation. In our application, the service layer is a perfect candidate for unit testing because it contains pure business logic that depends only on a database session. We can use an in-memory SQLite database to keep tests fast and isolated.

Create tests/conftest.py to define shared fixtures:

import pytest
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from app.database import Base
from app.models import Task


@pytest.fixture
def db_session():
    engine = create_engine(
        "sqlite:///:memory:",
        connect_args={"check_same_thread": False},
    )
    Base.metadata.create_all(engine)
    TestingSessionLocal = sessionmaker(
        autocommit=False, autoflush=False, bind=engine
    )
    session = TestingSessionLocal()
    try:
        yield session
    finally:
        session.close()
        Base.metadata.drop_all(engine)


@pytest.fixture
def sample_task(db_session):
    task = Task(title="Learn FastAPI", description="Read the docs", completed=False)
    db_session.add(task)
    db_session.commit()
    db_session.refresh(task)
    return task

Now write unit tests for the service layer in tests/test_unit.py:

from app.models import TaskCreate
from app import services


def test_create_task(db_session):
    task_data = TaskCreate(title="Write tests", description="Cover all endpoints")
    task = services.create_task(db_session, task_data)

    assert task.id is not None
    assert task.title == "Write tests"
    assert task.description == "Cover all endpoints"
    assert task.completed is False


def test_get_task(db_session, sample_task):
    task = services.get_task(db_session, sample_task.id)

    assert task is not None
    assert task.title == "Learn FastAPI"


def test_get_task_not_found(db_session):
    task = services.get_task(db_session, 9999)

    assert task is None


def test_get_tasks(db_session, sample_task):
    # Add a second task
    services.create_task(db_session, TaskCreate(title="Second task"))
    tasks = services.get_tasks(db_session)

    assert len(tasks) == 2


def test_mark_complete(db_session, sample_task):
    assert sample_task.completed is False
    task = services.mark_complete(db_session, sample_task.id)

    assert task.completed is True


def test_mark_complete_not_found(db_session):
    result = services.mark_complete(db_session, 9999)

    assert result is None


def test_delete_task(db_session, sample_task):
    result = services.delete_task(db_session, sample_task.id)

    assert result is True
    assert services.get_task(db_session, sample_task.id) is None


def test_delete_task_not_found(db_session):
    result = services.delete_task(db_session, 9999)

    assert result is False

These tests run entirely in memory, execute in milliseconds, and verify that each service function behaves correctly. Because each test gets a fresh database session through the fixture, there are no side effects between tests.

Writing Integration Tests

Integration tests verify that multiple components work together correctly. For a FastAPI application, this typically means testing the full request-response cycle through the API endpoints. FastAPI's TestClient makes this easy by simulating real HTTP requests against your app without starting a server.

Update tests/conftest.py to add an integration test client fixture that overrides the database dependency:

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


@pytest.fixture
def test_db():
    engine = create_engine(
        "sqlite:///:memory:",
        connect_args={"check_same_thread": False},
    )
    Base.metadata.create_all(engine)
    TestingSessionLocal = sessionmaker(
        autocommit=False, autoflush=False, bind=engine
    )

    def override_get_db():
        db = TestingSessionLocal()
        try:
            yield db
        finally:
            db.close()

    app.dependency_overrides[get_db] = override_get_db
    yield TestingSessionLocal()
    app.dependency_overrides.clear()
    Base.metadata.drop_all(engine)


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

The key technique here is dependency overriding. FastAPI allows you to replace any dependency — in this case get_db — with a custom implementation during tests. This lets us swap the production database for an in-memory SQLite database without changing application code.

Now write the integration tests in tests/test_integration.py:

def test_create_task_endpoint(client):
    response = client.post(
        "/tasks/",
        json={"title": "Integration test", "description": "Test the API"},
    )

    assert response.status_code == 201 or response.status_code == 200
    data = response.json()
    assert data["title"] == "Integration test"
    assert data["description"] == "Test the API"
    assert data["completed"] is False
    assert "id" in data


def test_list_tasks_endpoint(client):
    client.post("/tasks/", json={"title": "Task 1"})
    client.post("/tasks/", json={"title": "Task 2"})

    response = client.get("/tasks/")

    assert response.status_code == 200
    data = response.json()
    assert len(data) == 2


def test_get_task_endpoint(client):
    create_response = client.post(
        "/tasks/",
        json={"title": "Get me", "description": "Single task"},
    )
    task_id = create_response.json()["id"]

    response = client.get(f"/tasks/{task_id}")

    assert response.status_code == 200
    assert response.json()["title"] == "Get me"


def test_get_task_not_found(client):
    response = client.get("/tasks/9999")

    assert response.status_code == 404
    assert response.json()["detail"] == "Task not found"


def test_complete_task_endpoint(client):
    create_response = client.post("/tasks/", json={"title": "Complete me"})
    task_id = create_response.json()["id"]

    response = client.patch(f"/tasks/{task_id}/complete")

    assert response.status_code == 200
    assert response.json()["completed"] is True


def test_complete_task_not_found(client):
    response = client.patch("/tasks/9999/complete")

    assert response.status_code == 404


def test_delete_task_endpoint(client):
    create_response = client.post("/tasks/", json={"title": "Delete me"})
    task_id = create_response.json()["id"]

    response = client.delete(f"/tasks/{task_id}")

    assert response.status_code == 200
    assert response.json()["message"] == "Task deleted"

    # Verify it's actually gone
    get_response = client.get(f"/tasks/{task_id}")
    assert get_response.status_code == 404


def test_delete_task_not_found(client):
    response = client.delete("/tasks/9999")

    assert response.status_code == 404


def test_create_task_validation_error(client):
    response = client.post("/tasks/", json={"description": "Missing title"})

    assert response.status_code == 422

These integration tests exercise the entire stack: routing, dependency injection, Pydantic validation, service logic, and database operations. They verify not just that the logic works, but that the API contract is correct — status codes, response shapes, and error handling all behave as expected.

Testing Authentication

Most real-world FastAPI applications include authentication. Let's look at how to test protected endpoints. Add a simple token-based auth dependency to app/main.py:

from fastapi import Header
import secrets

VALID_TOKENS = {"secret-token-123"}


async def verify_token(x_token: str = Header()):
    if x_token not in VALID_TOKENS:
        raise HTTPException(status_code=401, detail="Invalid token")
    return x_token


@app.get("/secure/tasks/", response_model=list[TaskResponse])
def list_secure_tasks(
    db: Session = Depends(get_db),
    token: str = Depends(verify_token),
):
    return services.get_tasks(db)

Now test both authenticated and unauthenticated access:

def test_secure_endpoint_with_valid_token(client):
    client.post("/tasks/", json={"title": "Secure task"})

    response = client.get(
        "/secure/tasks/",
        headers={"X-Token": "secret-token-123"},
    )

    assert response.status_code == 200
    assert len(response.json()) == 1


def test_secure_endpoint_without_token(client):
    response = client.get("/secure/tasks/")

    assert response.status_code == 422


def test_secure_endpoint_with_invalid_token(client):
    response = client.get(
        "/secure/tasks/",
        headers={"X-Token": "wrong-token"},
    )

    assert response.status_code == 401
    assert response.json()["detail"] == "Invalid token"

Testing Async Endpoints

FastAPI supports async route handlers. To test async code, use pytest-asyncio and AsyncClient from httpx. Here is an example of testing an async endpoint:

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


@pytest.mark.asyncio
async def test_async_endpoint():
    transport = ASGITransport(app=app)
    async with AsyncClient(transport=transport, base_url="http://test") as ac:
        response = await ac.post("/tasks/", json={"title": "Async task"})

    assert response.status_code in (200, 201)
    assert response.json()["title"] == "Async task"

Note that when using AsyncClient, you need to handle dependency overrides the same way as with TestClient. The app.dependency_overrides dictionary works regardless of which client you use.

Best Practices for Testing FastAPI

To keep your test suite maintainable and effective, follow these best practices:

• Separate unit and integration tests: Keep pure logic tests fast and isolated. Reserve slower integration tests for verifying API contracts. Consider using pytest markers to run them separately in CI.
• Use dependency overrides for databases: Never let tests touch your production database. Always override the database dependency with an in-memory or test database.
• Test the happy path and edge cases: Cover successful operations, missing resources (404s), validation errors (422s), and unauthorized access (401s).
• Keep tests independent: Each test should set up and tear down its own state. Avoid relying on execution order or shared mutable state between tests.
• Test response schemas, not just status codes: Verify that the JSON response contains the expected fields and values, not just that the request succeeded.
• Mock external services: If your app calls third-party APIs, use unittest.mock or libraries like respx to mock those calls in tests.
• Aim for meaningful coverage: Focus on testing critical business logic and complex code paths rather than chasing 100% line coverage of trivial code.
• Use descriptive test names: Names like test_create_task_with_missing_title_returns_422 clearly communicate what each test verifies.

Running the Tests

Run the entire test suite with a simple command:

pytest

To run only unit tests or integration tests, use the -k flag to filter by filename:

pytest tests/test_unit.py
pytest tests/test_integration.py

To see verbose output with test names, add the -v flag. To measure coverage, install pytest-cov and run:

pytest --cov=app --cov-report=term-missing

This shows which lines of your application code are covered by tests and highlights gaps that need attention.

Conclusion

Testing FastAPI applications does not have to be complicated. By leveraging pytest fixtures, FastAPI's dependency injection system, and the built-in TestClient, you can build a robust test suite that covers both individual functions and full API workflows. Start with unit tests for your service layer to verify business logic quickly, then layer in integration tests to validate the complete request-response cycle. As your application grows, continue to maintain test independence, mock external dependencies, and focus on meaningful coverage of critical paths. A well-tested FastAPI application gives you the confidence to refactor freely, add features safely, and deploy to production knowing your code behaves exactly as intended.