← All articles

Testing FastAPI Applications: Unit, Integration, and E2E

December 20, 2025 · 5 min read

pythonpytestapi-testing

FastAPI's dependency injection system makes it unusually testable. Overriding dependencies at test time means you can swap databases, mock external services, and inject test users without changing application code. This article builds a full test pyramid from unit tests through integration tests with real services.

Project Structure

myapp/
  main.py              # FastAPI app
  models.py            # SQLAlchemy models
  schemas.py           # Pydantic schemas
  dependencies.py      # DI dependencies (get_db, get_current_user)
  routers/
    articles.py
    users.py
  services/
    article_service.py
  repositories/
    article_repo.py

tests/
  conftest.py          # Shared fixtures
  unit/
    test_services.py
    test_validators.py
  integration/
    test_article_api.py
    test_auth_api.py
  factories.py         # factory_boy factories

conftest.py Foundation

# tests/conftest.py
import pytest
import asyncio
from typing import AsyncGenerator
from httpx import AsyncClient, ASGITransport
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker
 
from myapp.main import app
from myapp.models import Base, User
from myapp.dependencies import get_db, get_current_user
 
TEST_DATABASE_URL = "postgresql+asyncpg://postgres:password@localhost/testdb"
 
@pytest.fixture(scope="session")
def event_loop():
    policy = asyncio.get_event_loop_policy()
    loop = policy.new_event_loop()
    yield loop
    loop.close()
 
@pytest.fixture(scope="session")
async def engine():
    engine = create_async_engine(TEST_DATABASE_URL, echo=False)
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    yield engine
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.drop_all)
    await engine.dispose()
 
@pytest.fixture
async def db(engine) -> AsyncGenerator[AsyncSession, None]:
    async_session = async_sessionmaker(engine, expire_on_commit=False)
    async with async_session() as session:
        async with session.begin():
            yield session
            await session.rollback()  # Roll back after each test
 
@pytest.fixture
async def client(db: AsyncSession) -> AsyncGenerator[AsyncClient, None]:
    async def override_get_db():
        yield db
 
    app.dependency_overrides[get_db] = override_get_db
 
    async with AsyncClient(
        transport=ASGITransport(app=app),
        base_url="http://test",
    ) as ac:
        yield ac
 
    del app.dependency_overrides[get_db]
 
@pytest.fixture
async def auth_client(client: AsyncClient, db: AsyncSession) -> AsyncClient:
    """Client authenticated as a regular user."""
    from tests.factories import UserFactory
 
    user = await UserFactory.create(db=db)
    app.dependency_overrides[get_current_user] = lambda: user
    yield client
    if get_current_user in app.dependency_overrides:
        del app.dependency_overrides[get_current_user]
 
@pytest.fixture
async def admin_client(client: AsyncClient, db: AsyncSession) -> AsyncClient:
    """Client authenticated as an admin."""
    from tests.factories import UserFactory
 
    admin = await UserFactory.create(db=db, role="admin")
    app.dependency_overrides[get_current_user] = lambda: admin
    yield client
    if get_current_user in app.dependency_overrides:
        del app.dependency_overrides[get_current_user]

Factory Boy for Test Data

# tests/factories.py
import factory
from factory import Faker
from myapp.models import User, Article
from myapp.auth import hash_password
 
class UserFactory(factory.AsyncAlchemyModelFactory):
    class Meta:
        model = User
        sqlalchemy_session_persistence = "commit"
 
    id = factory.Sequence(lambda n: n + 1)
    email = Faker("email")
    name = Faker("name")
    password_hash = factory.LazyFunction(lambda: hash_password("testpassword"))
    role = "user"
    is_active = True
 
    @classmethod
    async def create(cls, db, **kwargs):
        obj = cls.build(**kwargs)
        db.add(obj)
        await db.flush()
        return obj
 
class ArticleFactory(factory.AsyncAlchemyModelFactory):
    class Meta:
        model = Article
 
    id = factory.Sequence(lambda n: n + 1)
    title = Faker("sentence", nb_words=5)
    slug = factory.LazyAttribute(lambda obj: obj.title.lower().replace(" ", "-")[:50])
    content = Faker("paragraph", nb_sentences=10)
    published = False
    author_id = factory.SelfAttribute("author.id")
    author = factory.SubFactory(UserFactory)
 
    @classmethod
    async def create(cls, db, **kwargs):
        obj = cls.build(**kwargs)
        db.add(obj)
        await db.flush()
        return obj

Unit Tests (No Database)

Test service layer with pure mocks:

# tests/unit/test_article_service.py
import pytest
from unittest.mock import AsyncMock, MagicMock
from myapp.services.article_service import ArticleService
from myapp.schemas import CreateArticleSchema
 
@pytest.fixture
def article_repo():
    return AsyncMock()
 
@pytest.fixture
def email_service():
    return AsyncMock()
 
@pytest.fixture
def service(article_repo, email_service):
    return ArticleService(repo=article_repo, email_service=email_service)
 
async def test_create_article_sends_notification_when_published(service, article_repo, email_service):
    article_repo.create.return_value = MagicMock(
        id=1, title="Test", published=True, author_email="author@example.com"
    )
 
    schema = CreateArticleSchema(title="Test", content="Content " * 20, published=True)
    result = await service.create_article(schema, user_id=1)
 
    article_repo.create.assert_awaited_once()
    email_service.notify_subscribers.assert_awaited_once_with(article_id=1)
 
async def test_create_draft_does_not_send_notification(service, article_repo, email_service):
    article_repo.create.return_value = MagicMock(id=1, title="Draft", published=False)
 
    schema = CreateArticleSchema(title="Draft", content="Content " * 20, published=False)
    await service.create_article(schema, user_id=1)
 
    email_service.notify_subscribers.assert_not_awaited()
 
async def test_publish_article_validates_minimum_content_length(service):
    schema = CreateArticleSchema(title="Too Short", content="Short", published=True)
 
    with pytest.raises(ValueError, match="Content too short"):
        await service.create_article(schema, user_id=1)

Integration Tests (Real Database)

# tests/integration/test_article_api.py
import pytest
from tests.factories import ArticleFactory, UserFactory
 
class TestGetArticles:
    async def test_returns_only_published_articles(self, client, db):
        published = await ArticleFactory.create(db=db, published=True)
        await ArticleFactory.create(db=db, published=False)  # Should not appear
        await ArticleFactory.create(db=db, published=True)
 
        response = await client.get("/api/articles")
 
        assert response.status_code == 200
        articles = response.json()
        assert len(articles) == 2
        assert all(a["published"] for a in articles)
 
    async def test_paginates_results(self, client, db):
        for _ in range(15):
            await ArticleFactory.create(db=db, published=True)
 
        response = await client.get("/api/articles?page=1&limit=10")
        assert len(response.json()) == 10
 
        response = await client.get("/api/articles?page=2&limit=10")
        assert len(response.json()) == 5
 
    async def test_returns_empty_list_when_no_articles(self, client):
        response = await client.get("/api/articles")
        assert response.status_code == 200
        assert response.json() == []
 
class TestCreateArticle:
    async def test_requires_authentication(self, client):
        response = await client.post("/api/articles", json={
            "title": "Test", "content": "Content " * 20
        })
        assert response.status_code == 401
 
    async def test_creates_article_as_authenticated_user(self, auth_client, db):
        response = await auth_client.post("/api/articles", json={
            "title": "My Article",
            "content": "This is the content of my article. " * 5,
            "published": False,
        })
 
        assert response.status_code == 201
        article = response.json()
        assert article["title"] == "My Article"
        assert article["published"] is False
        assert "id" in article
 
    async def test_validates_title_length(self, auth_client):
        response = await auth_client.post("/api/articles", json={
            "title": "",  # Empty title
            "content": "Content " * 20,
        })
 
        assert response.status_code == 422
        errors = response.json()["detail"]
        assert any(e["loc"][-1] == "title" for e in errors)
 
    async def test_admin_can_create_for_other_user(self, admin_client, db):
        target_user = await UserFactory.create(db=db)
 
        response = await admin_client.post("/api/articles", json={
            "title": "Admin Article",
            "content": "Content " * 10,
            "author_id": target_user.id,
        })
 
        assert response.status_code == 201
        assert response.json()["author_id"] == target_user.id
 
class TestDeleteArticle:
    async def test_author_can_delete_own_article(self, auth_client, db):
        # The auth_client's user creates the article
        current_user = app.dependency_overrides[get_current_user]()
        article = await ArticleFactory.create(db=db, author_id=current_user.id)
 
        response = await auth_client.delete(f"/api/articles/{article.id}")
        assert response.status_code == 204
 
    async def test_cannot_delete_other_users_article(self, auth_client, db):
        other_user = await UserFactory.create(db=db)
        article = await ArticleFactory.create(db=db, author_id=other_user.id)
 
        response = await auth_client.delete(f"/api/articles/{article.id}")
        assert response.status_code == 403

Testing Authentication

# tests/integration/test_auth_api.py
async def test_login_with_valid_credentials(client, db):
    user = await UserFactory.create(db=db)
    # UserFactory sets password to "testpassword"
 
    response = await client.post("/api/auth/login", json={
        "email": user.email,
        "password": "testpassword",
    })
 
    assert response.status_code == 200
    data = response.json()
    assert "access_token" in data
    assert data["token_type"] == "bearer"
 
async def test_login_with_wrong_password(client, db):
    user = await UserFactory.create(db=db)
 
    response = await client.post("/api/auth/login", json={
        "email": user.email,
        "password": "wrongpassword",
    })
 
    assert response.status_code == 401
 
async def test_token_grants_access(client, db):
    user = await UserFactory.create(db=db)
 
    login_response = await client.post("/api/auth/login", json={
        "email": user.email,
        "password": "testpassword",
    })
    token = login_response.json()["access_token"]
 
    # Use token to access protected endpoint
    me_response = await client.get(
        "/api/me",
        headers={"Authorization": f"Bearer {token}"}
    )
    assert me_response.status_code == 200
    assert me_response.json()["email"] == user.email

This test pyramid — unit tests with mocks for service logic, integration tests for API behavior with a real DB — gives you fast feedback at the unit level and confidence at the integration level. The dependency injection override pattern is FastAPI's biggest testing advantage over Flask-style frameworks.