FastAPI

FastAPI Integration Testing with TestClient and Real Dependencies

05 Oct 2026 — 9 min read

Unit tests verify individual functions. Integration tests verify that your application works when all the pieces are connected. For FastAPI applications, integration testing means running your endpoints against real databases, real caches, and real external services — not mocks. This is where you find the bugs that only appear in production: connection pool exhaustion, transaction isolation failures, serialization mismatches, and race conditions in background workers.

TestClient vs httpx.AsyncClient

FastAPI's TestClient is built on requests and runs synchronously. It's simpler to use but has a fundamental limitation: it can't test truly async behavior. httpx.AsyncClient is the async alternative that lets you write tests that reflect how real HTTP clients interact with your API.

# Synchronous — TestClient
from fastapi.testclient import TestClient
from app.main import app

def test_sync_endpoint():
    client = TestClient(app)
    response = client.get("/health")
    assert response.status_code == 200
    # Client session doesn't persist between calls

# Asynchronous — httpx.AsyncClient
import pytest
import pytest_asyncio
from httpx import AsyncClient, ASGITransport
from app.main import app

@pytest_asyncio.fixture
async def client():
    async with AsyncClient(
        transport=ASGITransport(app=app),
        base_url="http://testserver"
    ) as c:
        yield c

async def test_async_endpoint(client):
    response = await client.get("/health")
    assert response.status_code == 200

Use TestClient when:

Tests are synchronous (no async def)
You're testing simple request/response behavior
You want simplicity and fast iteration
You're in a codebase that doesn't use async fixtures

Use AsyncClient when:

You need to test concurrent requests
You're testing WebSocket endpoints
Your test fixtures are async (database sessions, etc.)
You're testing streaming responses
You need fine-grained control over connection behavior

# The hybrid approach: AsyncClient for async tests, TestClient for sync
# conftest.py
import pytest
import pytest_asyncio
from fastapi.testclient import TestClient
from httpx import AsyncClient, ASGITransport
from app.main import app

@pytest.fixture
def sync_client():
    with TestClient(app) as client:
        yield client

@pytest_asyncio.fixture
async def async_client():
    async with AsyncClient(
        transport=ASGITransport(app=app),
        base_url="http://testserver",
        timeout=30.0
    ) as client:
        yield client

Testing with Real Redis

Mocking Redis gives you false confidence. The mock will happily return whatever you configure, regardless of whether your actual Redis commands are correct. Real Redis integration tests catch serialization bugs, key naming issues, and TTL logic errors.

# docker-compose.test.yml
version: "3.8"
services:
  redis-test:
    image: redis:7-alpine
    ports:
      - "6380:6379"  # Different port to avoid conflicts
    command: redis-server --appendonly no --save ""  # No persistence for tests

# tests/conftest.py
import pytest
import pytest_asyncio
import redis.asyncio as aioredis
from app.main import app
from app.dependencies import get_redis

TEST_REDIS_URL = "redis://localhost:6380"

@pytest_asyncio.fixture(scope="session")
async def redis_client():
    """Session-scoped Redis client for all tests."""
    client = await aioredis.from_url(TEST_REDIS_URL, decode_responses=True)
    yield client
    await client.aclose()

@pytest_asyncio.fixture(autouse=True)
async def flush_redis(redis_client):
    """Flush Redis before each test for clean state."""
    await redis_client.flushdb()
    yield
    # Optional: flush after too, for debugging

@pytest_asyncio.fixture
async def app_client(redis_client):
    """HTTP client connected to the app with real Redis."""
    app.dependency_overrides[get_redis] = lambda: redis_client

    from httpx import AsyncClient, ASGITransport
    async with AsyncClient(
        transport=ASGITransport(app=app),
        base_url="http://testserver"
    ) as client:
        yield client

    app.dependency_overrides.clear()

# tests/test_caching.py
import asyncio

async def test_cache_hit_served_from_redis(app_client, redis_client):
    """First call populates cache, second call should be faster and from cache."""
    # Cold cache
    response1 = await app_client.get("/products/1")
    assert response1.status_code == 200
    assert response1.headers.get("x-cache") == "MISS"

    # Verify key was set in Redis
    cached = await redis_client.get("product:1")
    assert cached is not None

    # Warm cache
    response2 = await app_client.get("/products/1")
    assert response2.status_code == 200
    assert response2.headers.get("x-cache") == "HIT"
    assert response2.json() == response1.json()

async def test_cache_expires_after_ttl(app_client, redis_client):
    await app_client.get("/products/1")  # populate cache

    # Get the TTL
    ttl = await redis_client.ttl("product:1")
    assert 0 < ttl <= 300  # Should expire within 5 minutes

async def test_cache_invalidated_on_update(app_client, redis_client):
    # Populate cache
    await app_client.get("/products/1")
    assert await redis_client.exists("product:1")

    # Update the product
    response = await app_client.patch(
        "/products/1",
        json={"price": 99.99}
    )
    assert response.status_code == 200

    # Cache should be invalidated
    assert not await redis_client.exists("product:1")

async def test_rate_limiting_persists_across_requests(app_client, redis_client):
    """Rate limit counters should accumulate in real Redis."""
    for i in range(10):
        response = await app_client.get(
            "/api/limited",
            headers={"X-Client-ID": "test-client"}
        )
        if i < 5:
            assert response.status_code == 200
        else:
            assert response.status_code == 429

    # Verify counter in Redis
    count = await redis_client.get("ratelimit:test-client")
    assert int(count) > 5

Docker Compose for Test Infrastructure

For integration tests that need multiple services, Docker Compose provides a reproducible environment that matches production:

# docker-compose.test.yml
version: "3.8"

services:
  postgres-test:
    image: postgres:15-alpine
    environment:
      POSTGRES_DB: testdb
      POSTGRES_USER: postgres
      POSTGRES_PASSWORD: testpassword
    ports:
      - "5433:5432"
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U postgres"]
      interval: 5s
      timeout: 3s
      retries: 10

  redis-test:
    image: redis:7-alpine
    ports:
      - "6380:6379"

  minio-test:
    image: minio/minio:latest
    command: server /data --console-address ":9001"
    environment:
      MINIO_ROOT_USER: minioadmin
      MINIO_ROOT_PASSWORD: minioadmin
    ports:
      - "9000:9000"
    healthcheck:
      test: ["CMD", "curl", "-f", "http://localhost:9000/minio/health/live"]
      interval: 5s
      retries: 10

# Makefile
test-integration:
    docker-compose -f docker-compose.test.yml up -d
    sleep 5  # Wait for services
    pytest tests/integration/ -v
    docker-compose -f docker-compose.test.yml down

Or use pytest-docker for automatic lifecycle management:

# conftest.py (with pytest-docker)
import pytest

@pytest.fixture(scope="session")
def docker_compose_file():
    return "docker-compose.test.yml"

@pytest.fixture(scope="session")
def docker_services_project_name():
    return "myapp_test"

@pytest.fixture(scope="session")
def postgres_url(docker_ip, docker_services):
    port = docker_services.port_for("postgres-test", 5432)
    docker_services.wait_until_responsive(
        timeout=30,
        pause=1,
        check=lambda: is_responsive(f"postgresql://postgres:testpassword@{docker_ip}:{port}/testdb")
    )
    return f"postgresql+asyncpg://postgres:testpassword@{docker_ip}:{port}/testdb"

Testing Webhooks

Webhooks are one of the hardest things to test because your application is the server, not the client. You need to simulate receiving a webhook payload:

# app/routers/webhooks.py
import hmac
import hashlib
from fastapi import APIRouter, Request, HTTPException, Header
from app.services import OrderService

router = APIRouter()

def verify_webhook_signature(payload: bytes, signature: str, secret: str) -> bool:
    expected = hmac.new(secret.encode(), payload, hashlib.sha256).hexdigest()
    return hmac.compare_digest(f"sha256={expected}", signature)

@router.post("/webhooks/stripe")
async def handle_stripe_webhook(
    request: Request,
    stripe_signature: str = Header(..., alias="stripe-signature"),
    order_service: OrderService = Depends(get_order_service),
):
    payload = await request.body()
    if not verify_webhook_signature(payload, stripe_signature, STRIPE_WEBHOOK_SECRET):
        raise HTTPException(status_code=400, detail="Invalid webhook signature")

    event = parse_stripe_event(payload)

    if event["type"] == "payment_intent.succeeded":
        await order_service.fulfill_order(event["data"]["object"]["metadata"]["order_id"])
    elif event["type"] == "payment_intent.payment_failed":
        await order_service.mark_payment_failed(event["data"]["object"]["metadata"]["order_id"])

    return {"status": "received"}

# tests/test_webhooks.py
import hmac
import hashlib
import json
import pytest
from fastapi.testclient import TestClient
from unittest.mock import AsyncMock, MagicMock
from app.main import app
from app.dependencies import get_order_service

WEBHOOK_SECRET = "test_webhook_secret"

def make_stripe_signature(payload: bytes, secret: str = WEBHOOK_SECRET) -> str:
    sig = hmac.new(secret.encode(), payload, hashlib.sha256).hexdigest()
    return f"sha256={sig}"

def stripe_event(event_type: str, order_id: str = "order-123") -> dict:
    return {
        "type": event_type,
        "data": {
            "object": {
                "id": "pi_test_123",
                "amount": 5000,
                "metadata": {"order_id": order_id}
            }
        }
    }

@pytest.fixture
def mock_order_service():
    service = MagicMock()
    service.fulfill_order = AsyncMock()
    service.mark_payment_failed = AsyncMock()
    return service

@pytest.fixture
def client(mock_order_service):
    app.dependency_overrides[get_order_service] = lambda: mock_order_service
    yield TestClient(app)
    app.dependency_overrides.clear()

class TestStripeWebhook:
    def test_payment_succeeded_fulfills_order(self, client, mock_order_service):
        payload = json.dumps(stripe_event("payment_intent.succeeded")).encode()
        sig = make_stripe_signature(payload)

        response = client.post(
            "/webhooks/stripe",
            content=payload,
            headers={
                "stripe-signature": sig,
                "content-type": "application/json"
            }
        )
        assert response.status_code == 200
        mock_order_service.fulfill_order.assert_called_once_with("order-123")

    def test_payment_failed_marks_order_failed(self, client, mock_order_service):
        payload = json.dumps(stripe_event("payment_intent.payment_failed")).encode()
        sig = make_stripe_signature(payload)

        response = client.post(
            "/webhooks/stripe",
            content=payload,
            headers={"stripe-signature": sig, "content-type": "application/json"}
        )
        assert response.status_code == 200
        mock_order_service.mark_payment_failed.assert_called_once_with("order-123")

    def test_invalid_signature_rejected(self, client, mock_order_service):
        payload = json.dumps(stripe_event("payment_intent.succeeded")).encode()

        response = client.post(
            "/webhooks/stripe",
            content=payload,
            headers={
                "stripe-signature": "sha256=invalidsignature",
                "content-type": "application/json"
            }
        )
        assert response.status_code == 400
        mock_order_service.fulfill_order.assert_not_called()

    def test_missing_signature_rejected(self, client):
        payload = json.dumps(stripe_event("payment_intent.succeeded")).encode()

        response = client.post(
            "/webhooks/stripe",
            content=payload,
            headers={"content-type": "application/json"}
        )
        assert response.status_code == 422  # Missing required header

    def test_tampered_payload_rejected(self, client):
        original = json.dumps(stripe_event("payment_intent.succeeded")).encode()
        sig = make_stripe_signature(original)

        # Tamper with the payload after signing
        tampered = original.replace(b"order-123", b"order-999")

        response = client.post(
            "/webhooks/stripe",
            content=tampered,
            headers={"stripe-signature": sig, "content-type": "application/json"}
        )
        assert response.status_code == 400

Testing File Uploads

File upload endpoints need tests for valid files, oversized files, wrong MIME types, and empty files:

# app/routers/uploads.py
from fastapi import APIRouter, File, UploadFile, HTTPException
from app.services import FileStorageService

router = APIRouter()

MAX_FILE_SIZE = 10 * 1024 * 1024  # 10MB
ALLOWED_TYPES = {"image/jpeg", "image/png", "image/webp", "application/pdf"}

@router.post("/documents/upload")
async def upload_document(
    file: UploadFile = File(...),
    storage: FileStorageService = Depends(get_storage),
):
    if file.content_type not in ALLOWED_TYPES:
        raise HTTPException(
            status_code=415,
            detail=f"Unsupported file type: {file.content_type}"
        )

    content = await file.read()
    if len(content) > MAX_FILE_SIZE:
        raise HTTPException(
            status_code=413,
            detail=f"File too large. Max size: {MAX_FILE_SIZE} bytes"
        )
    if len(content) == 0:
        raise HTTPException(status_code=400, detail="File is empty")

    url = await storage.store(
        filename=file.filename,
        content=content,
        content_type=file.content_type
    )
    return {"url": url, "size": len(content), "filename": file.filename}

# tests/test_file_uploads.py
import pytest
import io
from fastapi.testclient import TestClient
from unittest.mock import AsyncMock, MagicMock
from PIL import Image
from app.main import app
from app.dependencies import get_storage

def make_storage_mock(url="https://cdn.example.com/test-file.jpg"):
    storage = MagicMock()
    storage.store = AsyncMock(return_value=url)
    return storage

def make_image_bytes(format="JPEG", size=(100, 100)) -> bytes:
    """Create a real in-memory image for testing."""
    img = Image.new("RGB", size, color=(73, 109, 137))
    buf = io.BytesIO()
    img.save(buf, format=format)
    return buf.getvalue()

@pytest.fixture
def client():
    storage = make_storage_mock()
    app.dependency_overrides[get_storage] = lambda: storage
    yield TestClient(app), storage
    app.dependency_overrides.clear()

class TestFileUpload:
    def test_valid_jpeg_upload_succeeds(self, client):
        test_client, storage = client
        image_bytes = make_image_bytes("JPEG")

        response = test_client.post(
            "/documents/upload",
            files={"file": ("photo.jpg", image_bytes, "image/jpeg")}
        )
        assert response.status_code == 200
        body = response.json()
        assert "url" in body
        assert body["filename"] == "photo.jpg"
        assert body["size"] == len(image_bytes)
        storage.store.assert_called_once()

    def test_valid_png_upload_succeeds(self, client):
        test_client, _ = client
        png_bytes = make_image_bytes("PNG")

        response = test_client.post(
            "/documents/upload",
            files={"file": ("image.png", png_bytes, "image/png")}
        )
        assert response.status_code == 200

    def test_unsupported_type_rejected(self, client):
        test_client, storage = client

        response = test_client.post(
            "/documents/upload",
            files={"file": ("malware.exe", b"MZ\x90\x00", "application/octet-stream")}
        )
        assert response.status_code == 415
        assert "Unsupported file type" in response.json()["detail"]
        storage.store.assert_not_called()

    def test_oversized_file_rejected(self, client):
        test_client, storage = client
        # 11MB of data
        large_content = b"x" * (11 * 1024 * 1024)

        response = test_client.post(
            "/documents/upload",
            files={"file": ("large.jpg", large_content, "image/jpeg")}
        )
        assert response.status_code == 413
        storage.store.assert_not_called()

    def test_empty_file_rejected(self, client):
        test_client, storage = client

        response = test_client.post(
            "/documents/upload",
            files={"file": ("empty.jpg", b"", "image/jpeg")}
        )
        assert response.status_code == 400
        assert "empty" in response.json()["detail"].lower()

    def test_no_file_returns_422(self, client):
        test_client, _ = client
        response = test_client.post("/documents/upload")
        assert response.status_code == 422

    def test_multiple_file_upload(self, client):
        test_client, storage = client

        images = [make_image_bytes("JPEG") for _ in range(3)]
        files = [
            ("files", (f"photo{i}.jpg", img, "image/jpeg"))
            for i, img in enumerate(images)
        ]

        response = test_client.post("/documents/upload-batch", files=files)
        assert response.status_code == 200
        assert len(response.json()["uploaded"]) == 3
        assert storage.store.call_count == 3

End-to-End API Flow Tests

E2E tests follow a complete user journey through multiple endpoints:

# tests/integration/test_order_flow.py
import pytest
import pytest_asyncio
from httpx import AsyncClient, ASGITransport
from app.main import app

@pytest_asyncio.fixture
async def client():
    async with AsyncClient(
        transport=ASGITransport(app=app),
        base_url="http://testserver"
    ) as c:
        yield c

async def get_auth_token(client, email, password):
    response = await client.post("/auth/token", data={
        "username": email,
        "password": password,
        "grant_type": "password"
    })
    assert response.status_code == 200, f"Auth failed: {response.json()}"
    return response.json()["access_token"]

async def test_complete_order_flow(client):
    """
    Complete happy-path order flow:
    1. Register a new user
    2. Browse products
    3. Add items to cart
    4. Checkout
    5. Verify order confirmation
    6. Verify inventory updated
    """
    # Step 1: Register
    reg_response = await client.post("/users/register", json={
        "email": "buyer@example.com",
        "password": "securepass123",
        "name": "Test Buyer"
    })
    assert reg_response.status_code == 201
    user_id = reg_response.json()["id"]

    # Step 2: Authenticate
    token = await get_auth_token(client, "buyer@example.com", "securepass123")
    headers = {"Authorization": f"Bearer {token}"}

    # Step 3: Browse products (unauthenticated works too)
    products_response = await client.get("/products?category=electronics")
    assert products_response.status_code == 200
    products = products_response.json()["items"]
    assert len(products) > 0
    product_id = products[0]["id"]
    initial_stock = products[0]["stock"]

    # Step 4: Add to cart
    cart_response = await client.post(
        "/cart/items",
        json={"product_id": product_id, "quantity": 2},
        headers=headers
    )
    assert cart_response.status_code == 200
    cart_id = cart_response.json()["cart_id"]

    # Verify cart contents
    cart = await client.get(f"/cart/{cart_id}", headers=headers)
    assert cart.json()["item_count"] == 2
    assert cart.json()["total"] > 0

    # Step 5: Checkout
    order_response = await client.post(
        f"/cart/{cart_id}/checkout",
        json={
            "shipping_address": {
                "line1": "123 Test St",
                "city": "Testville",
                "country": "US",
                "postal_code": "12345"
            },
            "payment_method_id": "pm_card_visa"  # Stripe test token
        },
        headers=headers
    )
    assert order_response.status_code == 201
    order_id = order_response.json()["id"]
    assert order_response.json()["status"] == "confirmed"

    # Step 6: Verify order is retrievable
    order = await client.get(f"/orders/{order_id}", headers=headers)
    assert order.status_code == 200
    assert order.json()["user_id"] == user_id
    assert order.json()["status"] == "confirmed"

    # Step 7: Verify inventory was decremented
    product = await client.get(f"/products/{product_id}")
    updated_stock = product.json()["stock"]
    assert updated_stock == initial_stock - 2

async def test_order_flow_fails_gracefully_on_payment_failure(client):
    """Verify cart is preserved when payment fails."""
    token = await get_auth_token(client, "buyer@example.com", "securepass123")
    headers = {"Authorization": f"Bearer {token}"}

    # Add item to cart
    cart_response = await client.post(
        "/cart/items",
        json={"product_id": 1, "quantity": 1},
        headers=headers
    )
    cart_id = cart_response.json()["cart_id"]

    # Attempt checkout with failing payment method
    order_response = await client.post(
        f"/cart/{cart_id}/checkout",
        json={
            "shipping_address": {"line1": "123 St", "city": "City", "country": "US", "postal_code": "00000"},
            "payment_method_id": "pm_card_chargeDeclined"  # Stripe test card that declines
        },
        headers=headers
    )
    assert order_response.status_code == 402  # Payment Required
    assert "payment" in order_response.json()["detail"].lower()

    # Cart should still be intact
    cart = await client.get(f"/cart/{cart_id}", headers=headers)
    assert cart.status_code == 200
    assert cart.json()["item_count"] == 1

Running Integration Tests in CI

Separate integration tests from unit tests using markers:

# conftest.py
import pytest

def pytest_configure(config):
    config.addinivalue_line(
        "markers", "integration: marks tests as integration tests (requires external services)"
    )

# Run only unit tests (fast, no infrastructure needed)
pytest -m "not integration" tests/

# Run only integration tests
pytest -m integration tests/integration/

# Run everything
pytest tests/

# .github/workflows/test.yml
jobs:
  unit-tests:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - run: pip install -e ".[test]"
      - run: pytest -m "not integration" --cov=app

  integration-tests:
    runs-on: ubuntu-latest
    services:
      postgres:
        image: postgres:15
        env:
          POSTGRES_PASSWORD: testpassword
          POSTGRES_DB: testdb
        options: >-
          --health-cmd pg_isready
          --health-interval 5s
          --health-retries 10
      redis:
        image: redis:7
        options: --health-cmd "redis-cli ping"
    steps:
      - uses: actions/checkout@v4
      - run: pip install -e ".[test]"
      - run: pytest -m integration --cov=app
        env:
          DATABASE_URL: postgresql+asyncpg://postgres:testpassword@localhost/testdb
          REDIS_URL: redis://localhost:6379

The split between unit and integration tests lets you run fast feedback in feature branches and full coverage in main. The integration tests are slower but catch the class of bugs that mocks can never find — and those are exactly the bugs that reach production.