Part 5: Testing

Tutorial Part 5: Testing Your API

Testing isn’t optional - it’s essential for production applications. In this part, we’ll build a comprehensive test suite that gives you confidence in your code, catches bugs before users do, and enables safe refactoring.

What We’ll Build

By the end of this part:

• Unit tests for business logic
• Integration tests for API endpoints
• Authentication and authorization testing
• Database transaction testing
• Mock external services
• Test fixtures and factories
• Coverage reporting
• CI/CD pipeline with GitHub Actions

Tip

Test-Driven Development (TDD): While we’re adding tests after implementation for this tutorial, consider writing tests first in real projects. It leads to better design and fewer bugs.

Testing Philosophy

Before diving into code, understand these testing principles:

1. Test Pyramid: Many unit tests, fewer integration tests, minimal E2E tests
2. Fast Feedback: Tests should run quickly to encourage frequent running
3. Isolation: Tests shouldn’t depend on each other
4. Clarity: Failed tests should clearly indicate what’s broken
5. Coverage: Aim for 80%+ coverage, but focus on critical paths

Setting Up the Test Environment

First, install testing dependencies:

# Using uv
uv add --dev pytest pytest-asyncio pytest-cov httpx faker factory-boy

# Or using pip
pip install pytest pytest-asyncio pytest-cov httpx faker factory-boy

Create pytest.ini in your project root:

[tool:pytest]
# Test discovery patterns
testpaths = tests
python_files = test_*.py
python_classes = Test*
python_functions = test_*

# Async support
asyncio_mode = auto

# Coverage settings
addopts =
    --verbose
    --cov=app
    --cov-report=html
    --cov-report=term-missing
    --cov-fail-under=80

# Environment for tests
env =
    ENVIRONMENT=test
    DATABASE_URL=sqlite+aiosqlite:///:memory:
    SECRET_KEY=test-secret-key-not-for-production

Test Database Configuration

Create tests/conftest.py for shared fixtures:

"""
Shared test fixtures and configuration.

This file is automatically loaded by pytest and provides
fixtures available to all tests.
"""

import asyncio
from typing import AsyncGenerator, Generator
import pytest
from httpx import AsyncClient
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlmodel import SQLModel
from app.main import app
from app.database import get_session
from app.models import User, Project, Task
from app.auth import create_access_token

# Use in-memory SQLite for tests
TEST_DATABASE_URL = "sqlite+aiosqlite:///:memory:"

@pytest.fixture(scope="session")
def event_loop() -> Generator:
    """
    Create event loop for async tests.

    This ensures all async tests use the same event loop,
    preventing issues with database connections.
    """
    loop = asyncio.get_event_loop_policy().new_event_loop()
    yield loop
    loop.close()

@pytest.fixture
async def engine():
    """
    Create test database engine.

    Uses in-memory SQLite for speed and isolation.
    Each test gets a fresh database.
    """
    engine = create_async_engine(
        TEST_DATABASE_URL,
        echo=False,  # Set to True for SQL debugging
        future=True,
        # SQLite specific settings for testing
        connect_args={
            "check_same_thread": False,
        }
    )

    # Create all tables
    async with engine.begin() as conn:
        await conn.run_sync(SQLModel.metadata.create_all)

    yield engine

    # Cleanup
    await engine.dispose()

@pytest.fixture
async def session(engine) -> AsyncGenerator[AsyncSession, None]:
    """
    Create database session for tests.

    Each test gets its own session with automatic
    rollback for isolation.
    """
    async with AsyncSession(engine) as session:
        yield session
        # Rollback any uncommitted changes
        await session.rollback()

@pytest.fixture
async def client(session) -> AsyncGenerator[AsyncClient, None]:
    """
    Create test client with dependency overrides.

    Overrides the database session to use test database.
    """
    # Override database dependency
    async def override_get_session():
        yield session

    app.dependency_overrides[get_session] = override_get_session

    # Create test client
    async with AsyncClient(app=app, base_url="http://test") as ac:
        yield ac

    # Clean up overrides
    app.dependency_overrides.clear()

@pytest.fixture
async def auth_headers(test_user) -> dict:
    """
    Create authentication headers for requests.

    Returns headers with valid JWT token.
    """
    token = create_access_token(
        data={
            "sub": test_user.email,
            "user_id": test_user.id,
            "role": test_user.role
        }
    )
    return {"Authorization": f"Bearer {token}"}

@pytest.fixture
async def test_user(session) -> User:
    """
    Create a test user.

    Provides a standard user for tests that need
    authentication.
    """
    from app.auth import hash_password

    user = User(
        email="test@example.com",
        name="Test User",
        password_hash=hash_password("testpass123"),
        is_active=True,
        is_verified=True,
        role="user"
    )
    session.add(user)
    await session.commit()
    await session.refresh(user)
    return user

@pytest.fixture
async def admin_user(session) -> User:
    """Create an admin user for testing."""
    from app.auth import hash_password

    user = User(
        email="admin@example.com",
        name="Admin User",
        password_hash=hash_password("adminpass123"),
        is_active=True,
        is_verified=True,
        role="admin"
    )
    session.add(user)
    await session.commit()
    await session.refresh(user)
    return user

Test Factories with Factory Boy

Create tests/factories.py for generating test data:

"""
Test factories for generating realistic test data.

Factories make it easy to create test objects with
sensible defaults while allowing customization.
"""

import factory
from factory import Faker, SubFactory, LazyAttribute
from datetime import datetime, timedelta
from app.models import User, Project, Task, UserRole
from app.auth import hash_password

class UserFactory(factory.Factory):
    """
    Factory for creating test users.

    Usage:
        user = UserFactory()
        admin = UserFactory(role=UserRole.ADMIN)
        users = UserFactory.create_batch(10)
    """
    class Meta:
        model = User

    # Faker providers for realistic data
    name = Faker("name")
    email = Faker("email")
    password_hash = LazyAttribute(
        lambda obj: hash_password("testpass123")
    )
    role = UserRole.USER
    is_active = True
    is_verified = True
    created_at = Faker("date_time_this_year")

    @factory.post_generation
    def projects(self, create, extracted, **kwargs):
        """Add projects after user creation if specified."""
        if not create:
            return

        if extracted:
            # Add provided projects
            for project in extracted:
                self.projects.append(project)

class ProjectFactory(factory.Factory):
    """Factory for creating test projects."""
    class Meta:
        model = Project

    name = Faker("catch_phrase")
    description = Faker("paragraph")
    owner = SubFactory(UserFactory)
    is_archived = False
    created_at = Faker("date_time_this_month")

    @factory.post_generation
    def tasks(self, create, extracted, **kwargs):
        """Add tasks after project creation."""
        if not create:
            return

        if extracted:
            for task in extracted:
                self.tasks.append(task)
        else:
            # Create 3 default tasks
            for _ in range(3):
                TaskFactory(project=self)

class TaskFactory(factory.Factory):
    """Factory for creating test tasks."""
    class Meta:
        model = Task

    title = Faker("sentence", nb_words=4)
    description = Faker("text", max_nb_chars=200)
    project = SubFactory(ProjectFactory)
    assignee = SubFactory(UserFactory)
    priority = Faker("random_int", min=1, max=5)
    is_completed = False
    due_date = LazyAttribute(
        lambda obj: datetime.utcnow() + timedelta(days=7)
    )
    created_at = Faker("date_time_this_week")

    @classmethod
    def create_overdue(cls, **kwargs):
        """Create an overdue task."""
        return cls(
            due_date=datetime.utcnow() - timedelta(days=1),
            is_completed=False,
            **kwargs
        )

    @classmethod
    def create_completed(cls, **kwargs):
        """Create a completed task."""
        return cls(
            is_completed=True,
            completed_at=datetime.utcnow(),
            **kwargs
        )

Unit Tests for Services

Create tests/unit/test_user_service.py:

"""
Unit tests for UserService.

These tests focus on business logic in isolation,
mocking external dependencies.
"""

import pytest
from unittest.mock import Mock, patch, AsyncMock
from datetime import datetime, timedelta
from app.services.users import UserService
from app.models import UserCreate, UserUpdate
from app.exceptions import ConflictError, ValidationError, NotFoundError
from tests.factories import UserFactory

class TestUserService:
    """Test UserService business logic."""

    @pytest.fixture
    def service(self, session):
        """Create service with test session."""
        return UserService(session)

    async def test_create_user_success(self, service, session):
        """Test successful user creation."""
        # Arrange
        user_data = UserCreate(
            name="John Doe",
            email="john@example.com",
            password="SecurePass123"
        )

        # Act
        user = await service.create_user(user_data)

        # Assert
        assert user.id is not None
        assert user.email == "john@example.com"
        assert user.name == "John Doe"
        # Password should be hashed, not plain text
        assert user.password_hash != "SecurePass123"
        assert user.password_hash.startswith("$2b$")

    async def test_create_user_duplicate_email(self, service, session):
        """Test that duplicate emails are rejected."""
        # Create first user
        user1 = UserFactory(email="duplicate@example.com")
        session.add(user1)
        await session.commit()

        # Try to create second user with same email
        user_data = UserCreate(
            name="Another User",
            email="duplicate@example.com",
            password="Password123"
        )

        # Should raise ConflictError
        with pytest.raises(ConflictError) as exc_info:
            await service.create_user(user_data)

        assert "already registered" in str(exc_info.value)

    async def test_create_user_weak_password(self, service):
        """Test password validation."""
        user_data = UserCreate(
            name="Weak User",
            email="weak@example.com",
            password="123"  # Too short
        )

        with pytest.raises(ValidationError) as exc_info:
            await service.create_user(user_data)

        assert "at least 8 characters" in str(exc_info.value)

    async def test_get_user_found(self, service, session):
        """Test getting existing user."""
        # Create user
        user = UserFactory()
        session.add(user)
        await session.commit()

        # Get user
        found_user = await service.get_user(user.id)

        assert found_user.id == user.id
        assert found_user.email == user.email

    async def test_get_user_not_found(self, service):
        """Test getting non-existent user."""
        with pytest.raises(NotFoundError) as exc_info:
            await service.get_user(99999)

        assert "User 99999 not found" in str(exc_info.value)

    async def test_list_users_pagination(self, service, session):
        """Test user listing with pagination."""
        # Create 15 users
        users = UserFactory.create_batch(15)
        for user in users:
            session.add(user)
        await session.commit()

        # Get first page
        page1, total = await service.list_users(skip=0, limit=10)
        assert len(page1) == 10
        assert total == 15

        # Get second page
        page2, total = await service.list_users(skip=10, limit=10)
        assert len(page2) == 5
        assert total == 15

    async def test_list_users_search(self, service, session):
        """Test user search functionality."""
        # Create users with specific names
        alice = UserFactory(name="Alice Smith", email="alice@example.com")
        bob = UserFactory(name="Bob Jones", email="bob@example.com")
        charlie = UserFactory(name="Charlie Alice", email="charlie@example.com")

        session.add_all([alice, bob, charlie])
        await session.commit()

        # Search for "alice"
        results, total = await service.list_users(search="alice")

        assert total == 2
        emails = [user.email for user in results]
        assert "alice@example.com" in emails
        assert "charlie@example.com" in emails
        assert "bob@example.com" not in emails

    async def test_update_user_success(self, service, session):
        """Test updating user profile."""
        # Create user
        user = UserFactory(name="Old Name")
        session.add(user)
        await session.commit()

        # Update user
        update_data = UserUpdate(name="New Name")
        updated_user = await service.update_user(
            user.id,
            update_data,
            current_user_id=user.id  # User updating themselves
        )

        assert updated_user.name == "New Name"
        assert updated_user.updated_at > user.created_at

    async def test_update_user_permission_denied(self, service, session):
        """Test that users can't update others."""
        # Create two users
        user1 = UserFactory()
        user2 = UserFactory()
        session.add_all([user1, user2])
        await session.commit()

        # User1 tries to update User2
        update_data = UserUpdate(name="Hacked")

        with pytest.raises(PermissionError):
            await service.update_user(
                user2.id,
                update_data,
                current_user_id=user1.id
            )

    @patch('app.services.users.send_email')
    async def test_authentication_with_email_mock(
        self,
        mock_send_email,
        service,
        session
    ):
        """Test authentication flow with mocked email."""
        # Mock email sending
        mock_send_email.return_value = AsyncMock()

        # Create user
        user_data = UserCreate(
            name="Email Test",
            email="emailtest@example.com",
            password="TestPass123"
        )
        user = await service.create_user(user_data)

        # Verify email was "sent"
        mock_send_email.assert_called_once()
        call_args = mock_send_email.call_args
        assert "emailtest@example.com" in str(call_args)

        # Test authentication
        result = await service.authenticate(
            "emailtest@example.com",
            "TestPass123"
        )

        assert result is not None
        assert "access_token" in result
        assert result["user"].email == "emailtest@example.com"

Integration Tests for API Endpoints

Create tests/integration/test_api_endpoints.py:

"""
Integration tests for API endpoints.

These tests verify the full request/response cycle
including middleware, validation, and database operations.
"""

import pytest
from httpx import AsyncClient
from app.main import app
from tests.factories import UserFactory, ProjectFactory, TaskFactory

class TestUserEndpoints:
    """Test user-related API endpoints."""

    async def test_create_user_endpoint(self, client: AsyncClient):
        """Test POST /users endpoint."""
        response = await client.post(
            "/users",
            json={
                "name": "New User",
                "email": "newuser@example.com",
                "password": "SecurePass123"
            }
        )

        assert response.status_code == 201
        data = response.json()
        assert data["email"] == "newuser@example.com"
        assert "password" not in data  # Password should never be in response
        assert "password_hash" not in data

    async def test_get_user_endpoint(self, client: AsyncClient, test_user):
        """Test GET /users/{id} endpoint."""
        response = await client.get(f"/users/{test_user.id}")

        assert response.status_code == 200
        data = response.json()
        assert data["id"] == test_user.id
        assert data["email"] == test_user.email

    async def test_list_users_with_pagination_headers(
        self,
        client: AsyncClient,
        session
    ):
        """Test that pagination headers are included."""
        # Create multiple users
        users = UserFactory.create_batch(25)
        for user in users:
            session.add(user)
        await session.commit()

        # Request first page
        response = await client.get("/users?skip=0&limit=10")

        assert response.status_code == 200
        assert "X-Total-Count" in response.headers
        assert int(response.headers["X-Total-Count"]) == 25

        data = response.json()
        assert len(data) == 10

    async def test_update_user_requires_auth(
        self,
        client: AsyncClient,
        test_user
    ):
        """Test that update requires authentication."""
        # Try without auth
        response = await client.patch(
            f"/users/{test_user.id}",
            json={"name": "Updated Name"}
        )

        assert response.status_code == 403  # Or 401 depending on your setup

    async def test_update_user_with_auth(
        self,
        client: AsyncClient,
        test_user,
        auth_headers
    ):
        """Test authenticated user update."""
        response = await client.patch(
            f"/users/{test_user.id}",
            json={"name": "Updated Name"},
            headers=auth_headers
        )

        assert response.status_code == 200
        data = response.json()
        assert data["name"] == "Updated Name"

class TestProjectEndpoints:
    """Test project-related endpoints."""

    async def test_create_project_requires_auth(self, client: AsyncClient):
        """Test that creating projects requires authentication."""
        response = await client.post(
            "/projects",
            json={"name": "Test Project", "description": "Test"}
        )

        assert response.status_code in [401, 403]

    async def test_create_project_with_auth(
        self,
        client: AsyncClient,
        auth_headers
    ):
        """Test authenticated project creation."""
        response = await client.post(
            "/projects",
            json={
                "name": "My Project",
                "description": "Project description"
            },
            headers=auth_headers
        )

        assert response.status_code == 201
        data = response.json()
        assert data["name"] == "My Project"
        assert data["owner"]["email"] == "test@example.com"

    async def test_list_projects_filtering(
        self,
        client: AsyncClient,
        session,
        test_user,
        auth_headers
    ):
        """Test project listing with filters."""
        # Create projects for different users
        project1 = ProjectFactory(owner=test_user, is_archived=False)
        project2 = ProjectFactory(owner=test_user, is_archived=True)
        other_user = UserFactory()
        project3 = ProjectFactory(owner=other_user)

        session.add_all([project1, project2, project3, other_user])
        await session.commit()

        # Get only my non-archived projects
        response = await client.get(
            "/projects?my_projects_only=true&include_archived=false",
            headers=auth_headers
        )

        assert response.status_code == 200
        data = response.json()
        assert len(data) == 1
        assert data[0]["id"] == project1.id

class TestTaskEndpoints:
    """Test task-related endpoints."""

    async def test_create_task_in_project(
        self,
        client: AsyncClient,
        session,
        test_user,
        auth_headers
    ):
        """Test creating a task in a project."""
        # Create project
        project = ProjectFactory(owner=test_user)
        session.add(project)
        await session.commit()

        # Create task
        response = await client.post(
            "/tasks",
            json={
                "title": "New Task",
                "description": "Task description",
                "project_id": project.id,
                "priority": 3
            },
            headers=auth_headers
        )

        assert response.status_code == 201
        data = response.json()
        assert data["title"] == "New Task"
        assert data["project"]["id"] == project.id

    async def test_list_tasks_with_filters(
        self,
        client: AsyncClient,
        session,
        test_user
    ):
        """Test task listing with multiple filters."""
        # Create test data
        project = ProjectFactory(owner=test_user)
        completed_task = TaskFactory.create_completed(project=project)
        pending_task = TaskFactory(project=project, is_completed=False)
        overdue_task = TaskFactory.create_overdue(project=project)

        session.add_all([project, completed_task, pending_task, overdue_task])
        await session.commit()

        # Filter for pending tasks only
        response = await client.get(
            f"/tasks?project_id={project.id}&status=pending"
        )

        assert response.status_code == 200
        data = response.json()
        assert len(data) == 1
        assert data[0]["id"] == pending_task.id

    async def test_bulk_update_tasks(
        self,
        client: AsyncClient,
        session,
        test_user,
        auth_headers
    ):
        """Test bulk task update."""
        # Create project with tasks
        project = ProjectFactory(owner=test_user)
        tasks = TaskFactory.create_batch(5, project=project)

        session.add(project)
        for task in tasks:
            session.add(task)
        await session.commit()

        task_ids = [task.id for task in tasks]

        # Bulk update to mark as completed
        response = await client.post(
            "/tasks/bulk-update",
            json={
                "task_ids": task_ids,
                "update_data": {"is_completed": True}
            },
            headers=auth_headers
        )

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

Testing Authentication Flows

Create tests/integration/test_auth_flow.py:

"""
Test complete authentication flows.

These tests verify the full authentication journey
from registration to protected endpoints.
"""

import pytest
from datetime import datetime, timedelta
from jose import jwt
from app.config import settings

class TestAuthenticationFlow:
    """Test complete auth flows."""

    async def test_full_registration_login_flow(self, client):
        """Test complete registration and login flow."""
        # Step 1: Register
        register_response = await client.post(
            "/auth/register",
            json={
                "email": "newuser@example.com",
                "password": "SecurePass123",
                "name": "New User"
            }
        )
        assert register_response.status_code == 201

        # Step 2: Login
        login_response = await client.post(
            "/auth/login",
            json={
                "email": "newuser@example.com",
                "password": "SecurePass123"
            }
        )
        assert login_response.status_code == 200

        tokens = login_response.json()
        assert "access_token" in tokens
        assert "refresh_token" in tokens
        assert tokens["token_type"] == "bearer"

        # Step 3: Access protected endpoint
        headers = {"Authorization": f"Bearer {tokens['access_token']}"}
        protected_response = await client.get("/users/me", headers=headers)
        assert protected_response.status_code == 200

        user_data = protected_response.json()
        assert user_data["email"] == "newuser@example.com"

    async def test_token_refresh_flow(self, client, test_user):
        """Test token refresh mechanism."""
        # Login to get tokens
        login_response = await client.post(
            "/auth/login",
            json={
                "email": test_user.email,
                "password": "testpass123"
            }
        )

        tokens = login_response.json()
        refresh_token = tokens["refresh_token"]

        # Wait a moment (in real scenario, access token expires)
        import asyncio
        await asyncio.sleep(1)

        # Refresh the token
        refresh_response = await client.post(
            "/auth/refresh",
            json={"refresh_token": refresh_token}
        )

        assert refresh_response.status_code == 200
        new_tokens = refresh_response.json()
        assert "access_token" in new_tokens

        # Verify new token works
        headers = {"Authorization": f"Bearer {new_tokens['access_token']}"}
        response = await client.get("/users/me", headers=headers)
        assert response.status_code == 200

    async def test_password_reset_flow(self, client, test_user, monkeypatch):
        """Test complete password reset flow."""
        # Mock email sending
        emails_sent = []

        async def mock_send_email(to, subject, body):
            emails_sent.append({"to": to, "subject": subject, "body": body})

        monkeypatch.setattr("app.email.send_email", mock_send_email)

        # Step 1: Request password reset
        reset_request = await client.post(
            "/auth/password-reset",
            json={"email": test_user.email}
        )
        assert reset_request.status_code == 200

        # Verify email was "sent"
        assert len(emails_sent) == 1
        assert emails_sent[0]["to"] == test_user.email

        # Extract token from email (in real app, user clicks link)
        import re
        token_match = re.search(r"token=([a-zA-Z0-9_-]+)", emails_sent[0]["body"])
        assert token_match
        reset_token = token_match.group(1)

        # Step 2: Reset password with token
        reset_response = await client.post(
            "/auth/password-reset/confirm",
            json={
                "token": reset_token,
                "new_password": "NewSecurePass123"
            }
        )
        assert reset_response.status_code == 200

        # Step 3: Login with new password
        login_response = await client.post(
            "/auth/login",
            json={
                "email": test_user.email,
                "password": "NewSecurePass123"
            }
        )
        assert login_response.status_code == 200

    async def test_account_lockout(self, client, test_user):
        """Test account lockout after failed attempts."""
        # Make 5 failed login attempts
        for i in range(5):
            response = await client.post(
                "/auth/login",
                json={
                    "email": test_user.email,
                    "password": "wrongpassword"
                }
            )
            assert response.status_code == 401

        # 6th attempt should show lockout
        response = await client.post(
            "/auth/login",
            json={
                "email": test_user.email,
                "password": "testpass123"  # Even correct password
            }
        )

        assert response.status_code == 401
        assert "locked" in response.json()["detail"].lower()

Performance Testing

Create tests/performance/test_performance.py:

"""
Performance tests to ensure API meets requirements.

These tests verify response times and throughput
under various loads.
"""

import pytest
import asyncio
import time
from httpx import AsyncClient

class TestPerformance:
    """Performance and load tests."""

    @pytest.mark.slow  # Mark as slow test
    async def test_endpoint_response_time(self, client):
        """Test that endpoints respond quickly."""
        start = time.time()
        response = await client.get("/health")
        end = time.time()

        assert response.status_code == 200
        assert end - start < 0.1  # Should respond in under 100ms

    @pytest.mark.slow
    async def test_concurrent_requests(self, client, session):
        """Test handling concurrent requests."""
        # Create test data
        from tests.factories import UserFactory
        users = UserFactory.create_batch(10)
        for user in users:
            session.add(user)
        await session.commit()

        # Make 50 concurrent requests
        async def make_request():
            return await client.get("/users")

        start = time.time()
        tasks = [make_request() for _ in range(50)]
        responses = await asyncio.gather(*tasks)
        end = time.time()

        # All should succeed
        assert all(r.status_code == 200 for r in responses)

        # Should complete in reasonable time
        assert end - start < 5  # 50 requests in under 5 seconds

    @pytest.mark.slow
    async def test_database_query_performance(self, session):
        """Test database query performance."""
        from app.services.users import UserService
        from tests.factories import UserFactory

        # Create 1000 users
        users = UserFactory.create_batch(1000)
        for user in users:
            session.add(user)
        await session.commit()

        # Test pagination performance
        service = UserService(session)

        start = time.time()
        result, total = await service.list_users(skip=0, limit=100)
        end = time.time()

        assert len(result) == 100
        assert total == 1000
        assert end - start < 0.5  # Should paginate quickly

    @pytest.mark.slow
    async def test_memory_usage(self, client, session):
        """Test that memory usage stays reasonable."""
        import psutil
        import os

        process = psutil.Process(os.getpid())
        initial_memory = process.memory_info().rss / 1024 / 1024  # MB

        # Make many requests
        for _ in range(100):
            await client.get("/health")

        final_memory = process.memory_info().rss / 1024 / 1024  # MB
        memory_increase = final_memory - initial_memory

        # Memory increase should be minimal
        assert memory_increase < 50  # Less than 50MB increase

Code Coverage

Create .coveragerc for coverage configuration:

[run]
source = app
omit =
    */tests/*
    */migrations/*
    */__init__.py
    */config.py

[report]
exclude_lines =
    pragma: no cover
    def __repr__
    raise AssertionError
    raise NotImplementedError
    if __name__ == .__main__.:
    if TYPE_CHECKING:
    @abstractmethod

[html]
directory = htmlcov

Run tests with coverage:

# Run all tests with coverage
pytest --cov=app --cov-report=html

# Run specific test file
pytest tests/unit/test_user_service.py -v

# Run only fast tests (exclude slow/performance tests)
pytest -m "not slow"

# Run with parallel execution
pytest -n 4  # Requires pytest-xdist

# Generate coverage badge for README
coverage-badge -o coverage.svg

Continuous Integration with GitHub Actions

Create .github/workflows/test.yml:

name: Tests

on:
  push:
    branches: [ main, develop ]
  pull_request:
    branches: [ main ]

jobs:
  test:
    runs-on: ubuntu-latest

    strategy:
      matrix:
        python-version: [3.11, 3.12]

    services:
      postgres:
        image: postgres:15
        env:
          POSTGRES_USER: testuser
          POSTGRES_PASSWORD: testpass
          POSTGRES_DB: testdb
        options: >-
          --health-cmd pg_isready
          --health-interval 10s
          --health-timeout 5s
          --health-retries 5
        ports:
          - 5432:5432

    steps:
    - uses: actions/checkout@v3

    - name: Set up Python ${{ matrix.python-version }}
      uses: actions/setup-python@v4
      with:
        python-version: ${{ matrix.python-version }}

    - name: Cache dependencies
      uses: actions/cache@v3
      with:
        path: ~/.cache/pip
        key: ${{ runner.os }}-pip-${{ hashFiles('**/requirements.txt') }}
        restore-keys: |
          ${{ runner.os }}-pip-

    - name: Install dependencies
      run: |
        python -m pip install --upgrade pip
        pip install -r requirements.txt
        pip install -r requirements-dev.txt

    - name: Run linting
      run: |
        # Format checking
        black app tests --check
        # Linting
        flake8 app tests
        # Type checking
        mypy app

    - name: Run tests
      env:
        DATABASE_URL: postgresql://testuser:testpass@localhost/testdb
        SECRET_KEY: test-secret-key
      run: |
        pytest --cov=app --cov-report=xml --cov-report=term

    - name: Upload coverage to Codecov
      uses: codecov/codecov-action@v3
      with:
        file: ./coverage.xml
        fail_ci_if_error: true

Best Practices and Tips

1. Test Organization

Structure tests to mirror your application:

tests/
├── unit/           # Business logic tests
│   ├── test_services.py
│   └── test_models.py
├── integration/    # API endpoint tests
│   ├── test_auth.py
│   └── test_crud.py
├── performance/    # Performance tests
└── e2e/           # End-to-end tests

2. Test Data Management

Use factories for consistent test data:

# Bad: Hardcoded test data
user = User(
    email="test@example.com",
    name="Test User",
    password_hash="$2b$12$..."
)

# Good: Factory with defaults
user = UserFactory()
admin = UserFactory(role=UserRole.ADMIN)
users = UserFactory.create_batch(10)

3. Async Testing

Always use async fixtures and tests:

# Bad: Sync test for async code
def test_async_function():
    result = asyncio.run(async_function())
    assert result == expected

# Good: Async test
async def test_async_function():
    result = await async_function()
    assert result == expected

4. Mocking External Services

Mock external dependencies for reliable tests:

@patch('app.services.email.send_email')
async def test_with_mock(mock_send_email):
    mock_send_email.return_value = None
    # Test code that sends email
    mock_send_email.assert_called_once()

Common Issues and Solutions

Issue: Tests failing due to database state

Solution: Use transactions and rollback, ensure test isolation with fixtures

Issue: Async tests hanging

Solution: Use pytest-asyncio and ensure event loop is properly configured

Issue: Slow test suite

Solution: Use in-memory database for tests, parallelize with pytest-xdist

What We Learned

In this part, you’ve implemented: Comprehensive test setup with pytest Unit tests for business logic Integration tests for API endpoints Test factories for data generation Mocking external dependencies Performance testing Code coverage reporting CI/CD with GitHub Actions

Next Steps

In Part 6: Background Jobs, we’ll:

• Implement task queues with Celery
• Send emails asynchronously
• Schedule periodic tasks
• Process long-running jobs
• Monitor job execution

Continue to Part 6

Ready to add background jobs? Continue to Background Jobs →

Additional Resources

• pytest Documentation
• Testing Best Practices
• Factory Boy

---

Questions? Check our FAQ or ask in GitHub Discussions.