Module Project: Complete Test Pyramid

Project overview

You've learned the strategy (test pyramid and trade-offs), the technique of integration testing (FastAPI TestClient), database testing (fixtures), and E2E testing (complete flows). Now you're going to integrate it all by building a complete test pyramid for a task-management REST API.

The API is already implemented — your job is to create tests at all three levels with Claude Code: unit tests for the business logic, integration tests for the endpoints, and E2E tests for complete flows. Each level uses differentiated prompts and verifies different aspects of the system.

This project closes Phase 1 of the guide. By completing it, you'll have mastery of the three levels of testing and you'll know when to use each one. It's the exact template you'll replicate at a larger scale in the final project (Module 8).

---

Project Objective

Build a complete test pyramid with tests at all three levels for a REST API, using Claude Code as the generator and your strategic judgment to decide what to test at each level.

By completing this project:

• ✅ You'll have tests at all 3 levels: unit, integration, E2E
• ✅ You'll have used differentiated prompts for each level with Claude Code
• ✅ The pyramid will follow the recommended proportions (~70% unit, ~20% integration, ~10% E2E)
• ✅ The tests will be independent, organized by level, and with descriptive naming

---

Technical Specifications

Technology Stack

• Language: Python 3.10+
• Framework: FastAPI
• Testing: pytest + httpx (for TestClient)
• AI: Claude Code
• Dependencies: fastapi, uvicorn, httpx, pytest

Initial Setup

mkdir test-pyramid-project
cd test-pyramid-project

python -m venv venv
source venv/bin/activate

pip install fastapi uvicorn httpx pytest

Project Structure

test-pyramid-project/
├── main.py                ← REST API (given)
├── models.py              ← Models and business logic (given)
├── database.py            ← In-memory database (given)
├── tests/
│   ├── __init__.py
│   ├── conftest.py        ← Shared fixtures (you create it)
│   ├── unit/
│   │   ├── __init__.py
│   │   └── test_models.py ← Unit tests (you generate them with Claude Code)
│   ├── integration/
│   │   ├── __init__.py
│   │   └── test_endpoints.py ← Integration tests (you generate them)
│   └── e2e/
│       ├── __init__.py
│       └── test_flows.py  ← E2E tests (you generate them)
├── requirements.txt
└── venv/

---

The Code to Test

database.py

# database.py
"""Simple in-memory database for testing."""

_db: dict[str, list[dict]] = {"tasks": []}
_id_counter: dict[str, int] = {"tasks": 0}


def get_db():
    return _db


def reset_db():
    _db["tasks"] = []
    _id_counter["tasks"] = 0


def next_id(collection: str) -> int:
    _id_counter[collection] += 1
    return _id_counter[collection]

models.py

# models.py
"""Business logic and validation for tasks."""

from datetime import datetime
from typing import Optional


VALID_PRIORITIES = ["low", "medium", "high"]
VALID_STATUSES = ["pending", "in_progress", "completed"]


def validate_task(title: str, priority: str = "medium") -> dict:
    """Validate task data. Returns dict with is_valid and errors."""
    errors = []
    
    if not isinstance(title, str):
        errors.append("Title must be a string")
    elif not title.strip():
        errors.append("Title cannot be empty")
    elif len(title) > 200:
        errors.append("Title cannot exceed 200 characters")
    
    if priority not in VALID_PRIORITIES:
        errors.append(f"Priority must be one of: {', '.join(VALID_PRIORITIES)}")
    
    return {"is_valid": len(errors) == 0, "errors": errors}


def create_task_dict(
    task_id: int,
    title: str,
    priority: str = "medium",
    status: str = "pending",
) -> dict:
    """Create a task dictionary with all fields."""
    return {
        "id": task_id,
        "title": title.strip(),
        "priority": priority,
        "status": status,
        "created_at": datetime.utcnow().isoformat(),
        "updated_at": None,
    }


def can_transition_status(current: str, new: str) -> bool:
    """Check if status transition is valid."""
    allowed = {
        "pending": ["in_progress"],
        "in_progress": ["completed", "pending"],
        "completed": [],
    }
    return new in allowed.get(current, [])


def filter_tasks(
    tasks: list[dict],
    status: Optional[str] = None,
    priority: Optional[str] = None,
) -> list[dict]:
    """Filter tasks by status and/or priority."""
    result = tasks
    
    if status:
        if status not in VALID_STATUSES:
            raise ValueError(f"Invalid status: {status}")
        result = [t for t in result if t["status"] == status]
    
    if priority:
        if priority not in VALID_PRIORITIES:
            raise ValueError(f"Invalid priority: {priority}")
        result = [t for t in result if t["priority"] == priority]
    
    return result


def sort_tasks(tasks: list[dict], by: str = "created_at", reverse: bool = False) -> list[dict]:
    """Sort tasks by field."""
    valid_fields = ["created_at", "priority", "title"]
    if by not in valid_fields:
        raise ValueError(f"Cannot sort by '{by}'. Use: {', '.join(valid_fields)}")
    
    priority_order = {"high": 0, "medium": 1, "low": 2}
    
    if by == "priority":
        return sorted(tasks, key=lambda t: priority_order.get(t["priority"], 99), reverse=reverse)
    
    return sorted(tasks, key=lambda t: t.get(by, ""), reverse=reverse)

main.py

# main.py
"""FastAPI REST API for task management."""

from fastapi import FastAPI, HTTPException, Query
from pydantic import BaseModel, Field
from typing import Optional

from database import get_db, next_id, reset_db
from models import (
    validate_task, create_task_dict, can_transition_status,
    filter_tasks, sort_tasks, VALID_PRIORITIES, VALID_STATUSES,
)

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


class TaskCreate(BaseModel):
    title: str = Field(..., min_length=1, max_length=200)
    priority: str = Field(default="medium")


class TaskUpdate(BaseModel):
    title: Optional[str] = Field(None, min_length=1, max_length=200)
    priority: Optional[str] = None
    status: Optional[str] = None


@app.get("/health")
def health_check():
    return {"status": "healthy", "service": "task-manager"}


@app.get("/tasks")
def list_tasks(
    status: Optional[str] = Query(None),
    priority: Optional[str] = Query(None),
    sort_by: str = Query("created_at"),
):
    db = get_db()
    tasks = db["tasks"]
    
    try:
        if status or priority:
            tasks = filter_tasks(tasks, status=status, priority=priority)
        tasks = sort_tasks(tasks, by=sort_by)
    except ValueError as e:
        raise HTTPException(status_code=400, detail=str(e))
    
    return tasks


@app.get("/tasks/{task_id}")
def get_task(task_id: int):
    db = get_db()
    task = next((t for t in db["tasks"] if t["id"] == task_id), None)
    if not task:
        raise HTTPException(status_code=404, detail="Task not found")
    return task


@app.post("/tasks", status_code=201)
def create_task(task: TaskCreate):
    validation = validate_task(task.title, task.priority)
    if not validation["is_valid"]:
        raise HTTPException(status_code=400, detail=validation["errors"])
    
    db = get_db()
    task_id = next_id("tasks")
    new_task = create_task_dict(task_id, task.title, task.priority)
    db["tasks"].append(new_task)
    return new_task


@app.put("/tasks/{task_id}")
def update_task(task_id: int, updates: TaskUpdate):
    db = get_db()
    task = next((t for t in db["tasks"] if t["id"] == task_id), None)
    if not task:
        raise HTTPException(status_code=404, detail="Task not found")
    
    if updates.title is not None:
        task["title"] = updates.title.strip()
    
    if updates.priority is not None:
        if updates.priority not in VALID_PRIORITIES:
            raise HTTPException(status_code=400, detail=f"Invalid priority: {updates.priority}")
        task["priority"] = updates.priority
    
    if updates.status is not None:
        if not can_transition_status(task["status"], updates.status):
            raise HTTPException(
                status_code=400,
                detail=f"Cannot transition from '{task['status']}' to '{updates.status}'"
            )
        task["status"] = updates.status
    
    from datetime import datetime
    task["updated_at"] = datetime.utcnow().isoformat()
    return task


@app.delete("/tasks/{task_id}")
def delete_task(task_id: int):
    db = get_db()
    task = next((t for t in db["tasks"] if t["id"] == task_id), None)
    if not task:
        raise HTTPException(status_code=404, detail="Task not found")
    
    db["tasks"] = [t for t in db["tasks"] if t["id"] != task_id]
    return {"message": "Task deleted", "id": task_id}

---

What to Test at Each Level

Unit tests (tests/unit/test_models.py)

Test the functions in models.py in isolation — no HTTP, no database:

• validate_task: valid/invalid titles, priorities, empty strings, character limit
• create_task_dict: correct fields, title stripping, default values
• can_transition_status: valid and invalid transitions between statuses
• filter_tasks: filtering by status, priority, both, invalid values
• sort_tasks: sorting by different fields, reverse, an invalid field

Target: 15-20 unit tests

Integration tests (tests/integration/test_endpoints.py)

Test the HTTP endpoints with TestClient — status codes, response bodies:

• GET /health: returns 200 with a healthy status
• GET /tasks: returns a list, works with filters
• GET /tasks/{id}: returns a specific task, 404 if it doesn't exist
• POST /tasks: creates a task (201), fails with invalid data (400/422)
• PUT /tasks/{id}: updates fields, validates status transitions
• DELETE /tasks/{id}: deletes a task, 404 if it doesn't exist

Target: 8-12 integration tests

E2E tests (tests/e2e/test_flows.py)

Test complete flows that simulate a real user:

• Flow 1: Task lifecycle — create → get → update → complete → delete
• Flow 2: Filtering — create several tasks → filter by status → filter by priority
• Flow 3: Status transitions — create → start (in_progress) → complete → verify it can't go back to pending

Target: 3-5 E2E tests

---

Step-by-Step Process

Step 1: Create conftest.py

# tests/conftest.py
import pytest
from fastapi.testclient import TestClient
from main import app
from database import reset_db


@pytest.fixture(autouse=True)
def clean_database():
    """Reset database before each test."""
    reset_db()
    yield
    reset_db()


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


@pytest.fixture
def sample_task(client):
    """Create and return a sample task."""
    response = client.post("/tasks", json={"title": "Sample Task", "priority": "medium"})
    return response.json()

Step 2: Generate unit tests with Claude Code

Prompt:

Generate professional unit tests for the functions in models.py.
Cover: happy path, edge cases (empty values, limits), error handling.
Use parametrize for validate_task and can_transition_status.
AAA pattern. Descriptive naming.

Step 3: Generate integration tests

Prompt:

Generate integration tests for the endpoints in main.py using
the FastAPI TestClient. Test status codes, response bodies.
Include: 200/201 success, 404 not found, 400 bad request, 422 validation.
Use the client fixture from conftest.py.

Step 4: Generate E2E tests

Prompt:

Generate E2E tests that validate the complete flows of the Task Manager API.
Flow 1: Complete lifecycle (create→read→update→delete).
Flow 2: Create several tasks and filter by status and priority.
Flow 3: Valid and invalid status transitions.
Each test simulates a real user using the API from start to finish.

Step 5: Evaluate, refine, run

# Run all the tests
pytest tests/ -v

# Run by level
pytest tests/unit/ -v
pytest tests/integration/ -v
pytest tests/e2e/ -v

# Check the count
pytest tests/ --co -q

---

Success Criteria

Your project is complete when:

• ✅ pytest tests/ -v → all green
• ✅ 15-20 unit tests in tests/unit/
• ✅ 8-12 integration tests in tests/integration/
• ✅ 3-5 E2E tests in tests/e2e/
• ✅ The pyramid proportions are respected (~70% unit, ~20% integration, ~10% E2E)
• ✅ Each test is independent (an autouse fixture resets the DB)
• ✅ Descriptive naming throughout the suite

---

Evaluation Rubric (100 points)

Unit Tests (35 points)

• (15 pts) All the functions in models.py are tested
• (10 pts) Edge cases covered (empty values, limits, invalid types)
• (5 pts) Parametrize used for validate_task and can_transition_status
• (5 pts) Focused tests (one assert per test)

Integration Tests (30 points)

• (15 pts) All the endpoints tested (GET, POST, PUT, DELETE)
• (10 pts) Status codes and response bodies verified
• (5 pts) Error cases: 404, 400, 422

E2E Tests (20 points)

• (10 pts) At least 2 complete flows (lifecycle + filtering)
• (5 pts) Status transitions tested end-to-end
• (5 pts) Independent and reproducible flows

Organization (15 points)

• (5 pts) Correct directory structure (unit/, integration/, e2e/)
• (5 pts) conftest.py with shared fixtures
• (5 pts) Descriptive naming throughout the suite

Extra Credit (up to +10 points)

• (+5 pts) A concurrent access test (create and delete in rapid succession)
• (+5 pts) An adversarial prompt that uncovers an uncovered edge case

---

Common Mistakes

Mistake 1: Tests that depend on each other through the database state

Cause: Not using an autouse fixture that resets the DB.

Solution: The clean_database fixture in conftest.py with autouse=True guarantees that each test starts with a clean DB.

Mistake 2: E2E tests that are really integration tests

Cause: A test that does a single POST and verifies the response is integration, not E2E.

Solution: E2E implies multiple sequential requests that form a flow: create → read → update → delete. If it's a single request, it's integration.

Mistake 3: Unit tests that import FastAPI

Cause: Unit tests shouldn't touch HTTP or the app. They only test pure functions from models.py.

Solution: test_models.py imports from models import ..., never from main import app.

Mistake 4: Not testing error cases in integration

Cause: Only testing the happy path (200, 201).

Solution: Each endpoint needs at least one error test: 404 for resources that don't exist, 400/422 for invalid inputs.

Mistake 5: Too many E2E tests

Cause: Writing 15 E2E tests instead of 3-5.

Solution: E2E tests are for critical flows. If you have more than 5, some of them should probably be integration tests.

---

Resources for the Project

1. FastAPI Testing - Official testing documentation
2. pytest Fixtures - Fixture reference
3. Martin Fowler: Test Pyramid - The original reference
4. httpx Documentation - The HTTP client used by TestClient
5. Ham Vocke: Practical Test Pyramid - A practical guide

---

Connection with the Next Module

What you built today gets used directly in Phase 2:

• Module 4 (TDD Workflow): You'll use the red-green-refactor cycle with tests at all 3 levels
• Module 5 (Coverage): You'll measure your pyramid's coverage and discover gaps
• Module 6 (Mocking): You'll learn to mock external services in integration tests
• Module 8 (Final Project): You'll build a complete pyramid at a larger scale

You completed Phase 1. You have spec-first, unit tests, integration tests, and E2E tests. Phase 2 integrates them into professional workflows.

---

Module 3, Capsule 06 — Testing with Claude Code Guide Your first complete test pyramid — from unit to E2E