Testing and Monitoring APIs

pip install pytest pytest-asyncio httpx

project/
├── app/
│   ├── main.py
│   ├── models.py
│   └── routes.py
├── tests/
│   ├── __init__.py
│   ├── conftest.py      # Shared fixtures
│   ├── test_routes.py   # API endpoint tests
│   └── test_models.py   # Model/business logic tests
└── pytest.ini           # pytest configuration

# app/main.py
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel

app = FastAPI()

class User(BaseModel):
    username: str
    email: str

users_db = {}

@app.post("/users", status_code=201)
async def create_user(user: User):
    if user.username in users_db:
        raise HTTPException(status_code=400, detail="User already exists")
    users_db[user.username] = user
    return user

@app.get("/users/{username}")
async def get_user(username: str):
    if username not in users_db:
        raise HTTPException(status_code=404, detail="User not found")
    return users_db[username]

# tests/test_routes.py
from fastapi.testclient import TestClient
from app.main import app

client = TestClient(app)


def test_create_user_success():
    """Test successful user creation"""
    response = client.post("/users", json={
        "username": "alice",
        "email": "alice@example.com"
    })

    assert response.status_code == 201
    assert response.json() == {
        "username": "alice",
        "email": "alice@example.com"
    }


def test_create_user_duplicate():
    """Test creating duplicate user returns 400"""
    # Create first user
    client.post("/users", json={
        "username": "bob",
        "email": "bob@example.com"
    })

    # Try to create same user again
    response = client.post("/users", json={
        "username": "bob",
        "email": "bob2@example.com"
    })

    assert response.status_code == 400
    assert "already exists" in response.json()["detail"]


def test_get_user_success():
    """Test retrieving existing user"""
    # Create user
    client.post("/users", json={
        "username": "charlie",
        "email": "charlie@example.com"
    })

    # Get user
    response = client.get("/users/charlie")

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


def test_get_user_not_found():
    """Test retrieving non-existent user returns 404"""
    response = client.get("/users/nonexistent")

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

pytest tests/ -v

# tests/conftest.py
import pytest
from fastapi.testclient import TestClient
from app.main import app, users_db


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


@pytest.fixture(autouse=True)
def clear_database():
    """Clear database before each test"""
    users_db.clear()
    yield  # Test runs here
    users_db.clear()  # Cleanup after test


@pytest.fixture
def sample_user():
    """Create sample user data"""
    return {
        "username": "testuser",
        "email": "test@example.com"
    }

# tests/test_routes.py
def test_create_user(client, sample_user):
    """Test using fixtures"""
    response = client.post("/users", json=sample_user)

    assert response.status_code == 201
    assert response.json() == sample_user


def test_isolation(client):
    """Test that database is cleared between tests"""
    # Database should be empty due to clear_database fixture
    response = client.get("/users/testuser")
    assert response.status_code == 404

# app/routes.py
import httpx

async def fetch_github_user(username: str):
    """Fetch user from GitHub API"""
    async with httpx.AsyncClient() as client:
        response = await client.get(f"https://api.github.com/users/{username}")
        response.raise_for_status()
        return response.json()

# tests/test_routes.py
import pytest
import httpx
from unittest.mock import AsyncMock, patch


@pytest.mark.asyncio
async def test_fetch_github_user_success():
    """Test GitHub API call with mock"""
    # Mock response
    mock_response = AsyncMock()
    mock_response.json.return_value = {
        "login": "octocat",
        "name": "The Octocat"
    }
    mock_response.raise_for_status = AsyncMock()

    # Mock httpx.AsyncClient
    with patch("httpx.AsyncClient.get", return_value=mock_response):
        result = await fetch_github_user("octocat")

        assert result["login"] == "octocat"
        assert result["name"] == "The Octocat"


@pytest.mark.asyncio
async def test_fetch_github_user_not_found():
    """Test GitHub API 404 error"""
    mock_response = AsyncMock()
    mock_response.raise_for_status.side_effect = httpx.HTTPStatusError(
        "404 Not Found",
        request=None,
        response=None
    )

    with patch("httpx.AsyncClient.get", return_value=mock_response):
        with pytest.raises(httpx.HTTPStatusError):
            await fetch_github_user("nonexistent")

# tests/conftest.py
import pytest
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from app.database import Base
from app.main import app
from fastapi.testclient import TestClient

# Test database URL (use in-memory SQLite for speed)
TEST_DATABASE_URL = "sqlite:///./test.db"

@pytest.fixture(scope="function")
def test_db():
    """Create test database for each test"""
    engine = create_engine(TEST_DATABASE_URL, connect_args={"check_same_thread": False})

    # Create all tables
    Base.metadata.create_all(bind=engine)

    # Create session
    TestingSessionLocal = sessionmaker(bind=engine)
    db = TestingSessionLocal()

    try:
        yield db
    finally:
        db.close()
        # Drop all tables after test
        Base.metadata.drop_all(bind=engine)

@pytest.fixture(scope="function")
def client(test_db):
    """Create test client with test database"""
    def override_get_db():
        try:
            yield test_db
        finally:
            test_db.close()

    app.dependency_overrides[get_db] = override_get_db

    with TestClient(app) as test_client:
        yield test_client

    app.dependency_overrides.clear()

# tests/test_integration.py
def test_create_and_retrieve_user(client, test_db):
    """Test full user lifecycle: create, retrieve, update"""
    # 1. Create user
    create_response = client.post("/users", json={
        "username": "alice",
        "email": "alice@example.com"
    })
    assert create_response.status_code == 201
    user_id = create_response.json()["id"]

    # 2. Verify user exists in database
    from app.models import User
    db_user = test_db.query(User).filter(User.id == user_id).first()
    assert db_user is not None
    assert db_user.username == "alice"

    # 3. Retrieve via API
    get_response = client.get(f"/users/{user_id}")
    assert get_response.status_code == 200
    assert get_response.json()["username"] == "alice"

    # 4. Update user
    update_response = client.put(f"/users/{user_id}", json={
        "email": "alice.updated@example.com"
    })
    assert update_response.status_code == 200

    # 5. Verify update persisted
    test_db.refresh(db_user)
    assert db_user.email == "alice.updated@example.com"


def test_user_authentication_flow(client, test_db):
    """Test signup → login → access protected endpoint"""
    # 1. Signup
    signup_response = client.post("/auth/signup", json={
        "username": "bob",
        "password": "secret123",
        "email": "bob@example.com"
    })
    assert signup_response.status_code == 201

    # 2. Login
    login_response = client.post("/auth/login", json={
        "username": "bob",
        "password": "secret123"
    })
    assert login_response.status_code == 200
    token = login_response.json()["access_token"]

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

# Install Prism
npm install -g @stoplight/prism-cli

# Start mock server from OpenAPI spec
prism mock openapi.yaml

# Output:
# [1:23:45 PM] › [CLI] …  awaiting  Starting Prism…
# [1:23:45 PM] › [CLI] ℹ  info      GET     http://127.0.0.1:4010/users
# [1:23:45 PM] › [CLI] ℹ  info      POST    http://127.0.0.1:4010/users
# [1:23:45 PM] › [CLI] ▶  start     Prism is listening on http://127.0.0.1:4010

# openapi.yaml
openapi: 3.0.0
info:
  title: User API
  version: 1.0.0

paths:
  /users:
    get:
      summary: List users
      responses:
        '200':
          description: Successful response
          content:
            application/json:
              schema:
                type: array
                items:
                  $ref: '#/components/schemas/User'
              example:
                - id: 1
                  username: "alice"
                  email: "alice@example.com"
                - id: 2
                  username: "bob"
                  email: "bob@example.com"

    post:
      summary: Create user
      requestBody:
        required: true
        content:
          application/json:
            schema:
              $ref: '#/components/schemas/UserCreate'
      responses:
        '201':
          description: User created
          content:
            application/json:
              schema:
                $ref: '#/components/schemas/User'

components:
  schemas:
    User:
      type: object
      properties:
        id:
          type: integer
        username:
          type: string
        email:
          type: string
          format: email

    UserCreate:
      type: object
      required:
        - username
        - email
      properties:
        username:
          type: string
        email:
          type: string
          format: email

# tests/test_with_prism.py
import httpx
import pytest

PRISM_BASE_URL = "http://127.0.0.1:4010"

@pytest.mark.asyncio
async def test_get_users_from_mock():
    """Test frontend code against Prism mock"""
    async with httpx.AsyncClient(base_url=PRISM_BASE_URL) as client:
        response = await client.get("/users")

        assert response.status_code == 200
        users = response.json()
        assert isinstance(users, list)
        assert len(users) > 0
        assert "username" in users[0]
        assert "email" in users[0]


@pytest.mark.asyncio
async def test_create_user_mock():
    """Test POST endpoint against mock"""
    async with httpx.AsyncClient(base_url=PRISM_BASE_URL) as client:
        response = await client.post("/users", json={
            "username": "charlie",
            "email": "charlie@example.com"
        })

        assert response.status_code == 201
        user = response.json()
        assert user["username"] == "charlie"

# Benefits of Prism:
# ✓ Frontend development doesn't wait for backend
# ✓ Consistent with OpenAPI spec (catches schema drift)
# ✓ Dynamic responses based on request data
# ✓ Validates requests against schema automatically

# Install pact-python
pip install pact-python

# tests/test_consumer_pact.py
import pytest
from pact import Consumer, Provider, Like, EachLike

pact = Consumer("UserWebApp").has_pact_with(
    Provider("UserAPI"),
    pact_dir="./pacts"
)

@pytest.fixture(scope="module")
def pact_server():
    """Start Pact mock server"""
    pact.start_service()
    yield pact
    pact.stop_service()

import requests

def test_get_user_contract(pact_server):
    """Consumer expects: GET /users/1 returns user object"""
    expected = {
        "id": 1,
        "username": Like("alice"),
        "email": Like("alice@example.com")
    }

    (pact
     .given("User 1 exists")
     .upon_receiving("a request for user 1")
     .with_request("GET", "/users/1")
     .will_respond_with(200, body=expected))

    with pact:
        # Make request to pact mock server
        response = requests.get(f"{pact.uri}/users/1")
        assert response.status_code == 200
        assert response.json()["username"] == "alice"

    # Pact file generated: pacts/UserWebApp-UserAPI.json

def test_create_user_contract(pact_server):
    """Consumer expects: POST /users creates user"""
    request_body = {
        "username": "bob",
        "email": "bob@example.com"
    }

    expected_response = {
        "id": Like(2),
        "username": "bob",
        "email": "bob@example.com"
    }

    (pact
     .given("User does not exist")
     .upon_receiving("a request to create user")
     .with_request("POST", "/users", body=request_body)
     .will_respond_with(201, body=expected_response))

    with pact:
        response = requests.post(f"{pact.uri}/users", json=request_body)
        assert response.status_code == 201
        assert "id" in response.json()

# tests/test_provider_pact.py
import pytest
from pact import Verifier
from app.main import app
from app.database import Base, engine

@pytest.fixture(scope="module")
def provider_app():
    """Setup provider app with test data"""
    # Create tables
    Base.metadata.create_all(bind=engine)

    # Add test data for "User 1 exists" state
    from app.models import User
    db = SessionLocal()
    user = User(id=1, username="alice", email="alice@example.com")
    db.add(user)
    db.commit()
    db.close()

    yield app

    # Cleanup
    Base.metadata.drop_all(bind=engine)

def test_verify_pact(provider_app):
    """Verify provider meets consumer contract"""
    verifier = Verifier(
        provider="UserAPI",
        provider_base_url="http://localhost:8000"
    )

    # Verify against pact file from consumer
    success, logs = verifier.verify_pacts(
        "./pacts/UserWebApp-UserAPI.json",
        provider_states_setup_url="http://localhost:8000/_pact/provider_states"
    )

    assert success == 0  # 0 = success, 1 = failure

# If verification fails:
# - Provider changed API without updating consumer
# - Breaking change detected
# - Fix provider OR update consumer contract

# Install json-server
npm install -g json-server

# Create db.json
{
  "users": [
    { "id": 1, "username": "alice", "email": "alice@example.com" },
    { "id": 2, "username": "bob", "email": "bob@example.com" }
  ],
  "posts": [
    { "id": 1, "userId": 1, "title": "Hello World", "body": "First post" }
  ]
}

# Start server
json-server --watch db.json --port 3000

# Auto-generated endpoints:
# GET    /users
# GET    /users/1
# POST   /users
# PUT    /users/1
# PATCH  /users/1
# DELETE /users/1
# GET    /users?username=alice  (filtering)
# GET    /posts?_expand=user     (join)

pip install locust

# locustfile.py
import random
from locust import HttpUser, task, between


class APIUser(HttpUser):
    """Simulates realistic user behavior against the API."""
    wait_time = between(1, 3)  # Wait 1-3 seconds between tasks

    def on_start(self):
        """Called when a simulated user starts. Login and get token."""
        response = self.client.post("/auth/login", json={
            "username": "testuser",
            "password": "testpass"
        })
        self.token = response.json().get("access_token", "")
        self.headers = {"Authorization": f"Bearer {self.token}"}

    @task(5)  # Weight: 5x more likely than other tasks
    def list_users(self):
        """Most common operation: list users"""
        self.client.get("/users?limit=20", headers=self.headers)

    @task(3)
    def get_user(self):
        """Get individual user"""
        user_id = random.randint(1, 100)
        self.client.get(f"/users/{user_id}", headers=self.headers)

    @task(1)
    def create_user(self):
        """Least common: create user"""
        self.client.post("/users", json={
            "username": f"user_{random.randint(1000, 9999)}",
            "email": f"user{random.randint(1000, 9999)}@example.com"
        }, headers=self.headers)

# Start Locust web UI
locust -f locustfile.py --host=http://localhost:8000

# Open http://localhost:8089 in browser
# Configure: 100 users, spawn rate 10/s

# Or run headless (for CI/CD)
locust -f locustfile.py \
    --host=http://localhost:8000 \
    --users 100 \
    --spawn-rate 10 \
    --run-time 60s \
    --headless \
    --csv=results

# app/health.py
from fastapi import APIRouter, Response
import asyncio

router = APIRouter(tags=["health"])

# Database and cache clients (injected or imported)
from app.database import SessionLocal
from app.cache import redis_client


@router.get("/health/live")
async def liveness():
    """
    Liveness probe: Is the process alive?

    Returns 200 if the application is running.
    Kubernetes restarts the pod if this fails.
    """
    return {"status": "alive"}


@router.get("/health/ready")
async def readiness():
    """
    Readiness probe: Can the app serve traffic?

    Checks all dependencies (database, cache, etc.).
    Load balancer stops routing traffic if this fails.
    """
    checks = {}

    # Check database
    try:
        db = SessionLocal()
        db.execute("SELECT 1")
        db.close()
        checks["database"] = "ok"
    except Exception as e:
        checks["database"] = f"error: {str(e)}"

    # Check Redis cache
    try:
        redis_client.ping()
        checks["cache"] = "ok"
    except Exception as e:
        checks["cache"] = f"error: {str(e)}"

    # Overall status
    all_ok = all(v == "ok" for v in checks.values())
    status_code = 200 if all_ok else 503

    return Response(
        content={
            "status": "ready" if all_ok else "not_ready",
            "checks": checks
        },
        status_code=status_code
    )

# kubernetes/deployment.yaml
apiVersion: apps/v1
kind: Deployment
spec:
  template:
    spec:
      containers:
        - name: api
          image: myapi:latest
          ports:
            - containerPort: 8000

          # Liveness: restart if process hangs
          livenessProbe:
            httpGet:
              path: /health/live
              port: 8000
            initialDelaySeconds: 10
            periodSeconds: 15
            failureThreshold: 3

          # Readiness: stop traffic if dependencies down
          readinessProbe:
            httpGet:
              path: /health/ready
              port: 8000
            initialDelaySeconds: 5
            periodSeconds: 10
            failureThreshold: 2

          # Startup: give time for initialization
          startupProbe:
            httpGet:
              path: /health/live
              port: 8000
            initialDelaySeconds: 5
            periodSeconds: 5
            failureThreshold: 30  # 30 × 5s = 150s max startup

pip install prometheus-client

# app/metrics.py
import time
from fastapi import FastAPI, Request, Response
from prometheus_client import (
    Counter, Histogram, Gauge,
    generate_latest, CONTENT_TYPE_LATEST
)

# Define metrics
REQUEST_COUNT = Counter(
    "http_requests_total",
    "Total HTTP requests",
    ["method", "endpoint", "status_code"]
)

REQUEST_DURATION = Histogram(
    "http_request_duration_seconds",
    "HTTP request duration in seconds",
    ["method", "endpoint"],
    buckets=[0.01, 0.05, 0.1, 0.25, 0.5, 1.0, 2.5, 5.0]
)

ACTIVE_REQUESTS = Gauge(
    "http_active_requests",
    "Number of active HTTP requests"
)

ERROR_COUNT = Counter(
    "http_errors_total",
    "Total HTTP errors",
    ["method", "endpoint", "error_type"]
)


def setup_metrics(app: FastAPI):
    """Add metrics middleware to FastAPI app."""

    @app.middleware("http")
    async def metrics_middleware(request: Request, call_next):
        # Track active requests
        ACTIVE_REQUESTS.inc()
        start_time = time.time()

        try:
            response = await call_next(request)

            # Record metrics
            duration = time.time() - start_time
            endpoint = request.url.path
            method = request.method

            REQUEST_COUNT.labels(
                method=method,
                endpoint=endpoint,
                status_code=response.status_code
            ).inc()

            REQUEST_DURATION.labels(
                method=method,
                endpoint=endpoint
            ).observe(duration)

            if response.status_code >= 400:
                ERROR_COUNT.labels(
                    method=method,
                    endpoint=endpoint,
                    error_type=f"http_{response.status_code}"
                ).inc()

            return response
        finally:
            ACTIVE_REQUESTS.dec()

    @app.get("/metrics")
    async def metrics():
        """Prometheus scrape endpoint."""
        return Response(
            content=generate_latest(),
            media_type=CONTENT_TYPE_LATEST
        )

pip install opentelemetry-api opentelemetry-sdk \
    opentelemetry-instrumentation-fastapi \
    opentelemetry-exporter-otlp

# app/tracing.py
from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
from opentelemetry.instrumentation.fastapi import FastAPIInstrumentor

def setup_tracing(app):
    """Configure OpenTelemetry tracing for FastAPI."""

    # Create tracer provider
    provider = TracerProvider()

    # Export traces to collector (Jaeger, Zipkin, etc.)
    exporter = OTLPSpanExporter(endpoint="http://localhost:4317")
    provider.add_span_processor(BatchSpanProcessor(exporter))

    # Set global tracer
    trace.set_tracer_provider(provider)

    # Auto-instrument FastAPI
    FastAPIInstrumentor.instrument_app(app)


# Using traces in your code
tracer = trace.get_tracer(__name__)

async def get_user_with_posts(user_id: str):
    """Example: trace a multi-step operation."""

    with tracer.start_as_current_span("get_user_with_posts") as span:
        # Add attributes for debugging
        span.set_attribute("user.id", user_id)

        # Child span: database query
        with tracer.start_as_current_span("db.get_user"):
            user = await database.get_user(user_id)
            span.set_attribute("user.found", user is not None)

        # Child span: fetch posts
        with tracer.start_as_current_span("db.get_posts"):
            posts = await database.get_posts_by_user(user_id)
            span.set_attribute("posts.count", len(posts))

        return {"user": user, "posts": posts}

# Trace output (visible in Jaeger UI):
# get_user_with_posts [45ms]
#   ├── db.get_user [12ms]
#   └── db.get_posts [28ms]

# app/logging_config.py
import logging
import json
import uuid
from fastapi import FastAPI, Request


class JSONFormatter(logging.Formatter):
    """Output logs as structured JSON for log aggregation."""

    def format(self, record):
        log_data = {
            "timestamp": self.formatTime(record),
            "level": record.levelname,
            "message": record.getMessage(),
            "module": record.module,
            "function": record.funcName,
        }

        # Add correlation ID if available
        if hasattr(record, "correlation_id"):
            log_data["correlation_id"] = record.correlation_id

        # Add extra fields
        if hasattr(record, "user_id"):
            log_data["user_id"] = record.user_id

        return json.dumps(log_data)


def setup_logging(app: FastAPI):
    """Configure structured logging with correlation IDs."""

    # Setup JSON formatter
    handler = logging.StreamHandler()
    handler.setFormatter(JSONFormatter())
    logging.root.handlers = [handler]
    logging.root.setLevel(logging.INFO)

    @app.middleware("http")
    async def correlation_id_middleware(request: Request, call_next):
        # Get or generate correlation ID
        correlation_id = request.headers.get(
            "X-Correlation-ID",
            str(uuid.uuid4())
        )

        # Store in request state
        request.state.correlation_id = correlation_id

        # Process request
        response = await call_next(request)

        # Add to response headers
        response.headers["X-Correlation-ID"] = correlation_id
        return response


# Usage in endpoints
logger = logging.getLogger(__name__)

async def create_user(request: Request, user_data: dict):
    correlation_id = request.state.correlation_id

    logger.info(
        "Creating user",
        extra={
            "correlation_id": correlation_id,
            "user_id": user_data["username"]
        }
    )

    # ... create user ...

    logger.info(
        "User created successfully",
        extra={
            "correlation_id": correlation_id,
            "user_id": user_data["username"]
        }
    )

# Log output (structured JSON):
# {"timestamp": "2024-01-15 10:30:00", "level": "INFO",
#  "message": "Creating user", "correlation_id": "abc-123",
#  "user_id": "alice"}
#
# Trace the full request across services
# by searching for correlation_id: "abc-123"

# Service Level Indicators (SLIs) - What you measure
# Service Level Objectives (SLOs) - Your targets
# Service Level Agreements (SLAs) - Your promises to customers

# Example SLOs for a REST API:

# Availability SLO
# SLI: Percentage of successful requests (non-5xx)
# SLO: 99.9% availability (43.8 minutes downtime/month)
availability = successful_requests / total_requests
# Target: availability >= 0.999

# Latency SLO
# SLI: p95 response time
# SLO: 95% of requests complete within 500ms
latency_p95 = percentile(request_durations, 95)
# Target: latency_p95 <= 0.5

# Error Budget
# If SLO is 99.9%, error budget is 0.1%
# = 43.8 minutes of downtime per month
# Use error budget for deployments and experiments
# When budget is exhausted: freeze deployments, fix reliability

# prometheus/alert_rules.yml
groups:
  - name: api_alerts
    rules:
      # High error rate
      - alert: HighErrorRate
        expr: |
          rate(http_errors_total[5m])
          / rate(http_requests_total[5m]) > 0.01
        for: 5m
        labels:
          severity: critical
        annotations:
          summary: "Error rate above 1% for 5 minutes"

      # Slow responses
      - alert: HighLatency
        expr: |
          histogram_quantile(0.95,
            rate(http_request_duration_seconds_bucket[5m])
          ) > 0.5
        for: 10m
        labels:
          severity: warning
        annotations:
          summary: "p95 latency above 500ms for 10 minutes"

      # Service down
      - alert: ServiceDown
        expr: up == 0
        for: 1m
        labels:
          severity: critical
        annotations:
          summary: "API service is down"