# Testing FastAPI Services: Unit, Integration, and Contract Tests

The test suite was green. All 47 tests passing, CI clean, PR merged. Two hours later someone flagged that the staging database had 3,000 rows of test order data in the `orders` table.

A test that was supposed to mock the database wasn't. The mock was patched at the wrong path — the module where the function was defined, not where it was imported and used. Real DB calls went through. Real rows got written. Green tests, real side effects.

This is the most common testing mistake in FastAPI services, and it's entirely avoidable.

* * *

## Project Setup

```plaintext
tests/
├── conftest.py          # shared fixtures: app, client, db, mocks
├── unit/
│   └── test_models.py   # Pydantic validation, pure logic
├── integration/
│   └── test_routes.py   # full request → response against real dependencies
└── contract/
    └── test_contracts.py # schema compatibility between producer and consumer
```

One `conftest.py` at the root. Fixtures flow down — integration tests use the same client fixture as unit tests, just with different dependency overrides.

* * *

## The Mock Path Problem

The bug from the story above:

```python
# services/orders.py
from db import get_pool  # defined here

async def create_order(payload): ...

# ❌ Wrong — patches where it's defined, not where it's used
@patch("db.get_pool")
async def test_create_order(mock_pool): ...

# ✅ Correct — patch where the function is imported and called
@patch("services.orders.get_pool")
async def test_create_order(mock_pool): ...
```

Always patch at the import site, not the definition site. If `services/orders.py` imports `get_pool` from `db`, patch `services.orders.get_pool`.

FastAPI's dependency injection makes this cleaner — use `app.dependency_overrides` instead of `@patch` wherever possible:

```python
# conftest.py
import pytest
from httpx import AsyncClient, ASGITransport
from main import app
from db import get_db

async def override_get_db():
    # return a test session or mock
    yield MockDB()

@pytest.fixture
def client():
    app.dependency_overrides[get_db] = override_get_db
    yield  # tests run here
    app.dependency_overrides.clear()  # always clean up

@pytest.fixture
async def async_client(client):
    async with AsyncClient(
        transport=ASGITransport(app=app),
        base_url="http://test"
    ) as ac:
        yield ac
```

`dependency_overrides` swaps the real dependency for the test version at the FastAPI layer — no patching required, no wrong-path bugs.

* * *

## Unit Tests — Pure Logic Only

Unit tests should have zero I/O. No DB, no HTTP, no Redis. Test Pydantic models, business logic, and utility functions:

```python
# tests/unit/test_models.py
import pytest
from models.commerce import OrderPlacedEvent

def test_order_event_requires_amount():
    with pytest.raises(ValueError):
        OrderPlacedEvent(order_id="ord_1", user_id="usr_1", payload={})
        # amount is required — missing field raises

def test_order_event_legacy_field_accepted():
    event = OrderPlacedEvent(
        order_id="ord_1",
        user_id="usr_1",
        payload={"total_amount": 99.99}  # old field name
    )
    assert event.payload["amount"] == 99.99  # transition validator fired
```

Fast, no fixtures, no setup. These should run in milliseconds.

* * *

## Integration Tests — Full Request Stack

Integration tests hit the actual FastAPI routes with real request/response cycles, but with controlled dependencies:

```python
# tests/integration/test_routes.py
import pytest
from httpx import AsyncClient

@pytest.mark.anyio
async def test_ingest_valid_event(async_client: AsyncClient, mock_publisher):
    response = await async_client.post("/events/ingest", json={
        "event_type": "order.placed",
        "user_id": "usr_8821",
        "order_id": "ord_77234",
        "payload": {"amount": 89.99, "currency": "USD"}
    })

    assert response.status_code == 202
    assert "event_id" in response.json()
    mock_publisher.assert_called_once()

@pytest.mark.anyio
async def test_ingest_unknown_event_type(async_client: AsyncClient):
    response = await async_client.post("/events/ingest", json={
        "event_type": "nonexistent.event",
        "user_id": "usr_8821",
        "payload": {}
    })

    assert response.status_code == 422

@pytest.mark.anyio
async def test_health_check_degraded_when_db_down(async_client: AsyncClient, broken_db):
    response = await async_client.get("/health")
    assert response.status_code == 503
```

The `mock_publisher` and `broken_db` fixtures live in `conftest.py` and swap real dependencies via `dependency_overrides`. The routes, validation, exception handlers, and response shapes all get exercised — just not the real infrastructure.

Add `anyio` mode to `pytest.ini` or `pyproject.toml`:

```toml
[tool.pytest.ini_options]
anyio_mode = "auto"
```

* * *

## Contract Tests — Schema Compatibility

Contract tests answer one question: if the producer changes its schema, does the consumer still work?

```python
# tests/contract/test_contracts.py
from models.commerce import OrderPlacedEvent   # producer model
from consumer.models import IncomingOrderEvent  # consumer model
import json

def test_producer_output_satisfies_consumer_contract():
    # Simulate what the producer publishes
    producer_event = OrderPlacedEvent(
        order_id="ord_77234",
        user_id="usr_8821",
        payload={"amount": 89.99, "currency": "USD"}
    )

    # Serialize as the producer would (to Pub/Sub)
    published = json.loads(producer_event.model_dump_json())

    # Deserialize as the consumer would
    try:
        IncomingOrderEvent(**published)
    except Exception as e:
        pytest.fail(f"Consumer cannot parse producer output: {e}")
```

This test runs in CI on every PR. If a producer field rename would break the consumer, this catches it before merge — not after 3,000 rows of bad data.

For teams using a shared models package (covered in S01E02), this test is even simpler — the same class is imported on both sides, and the contract is enforced by the package version.

* * *

## CI Configuration

```yaml
# .github/workflows/test.yml
name: Test

on: [push, pull_request]

jobs:
  test:
    runs-on: ubuntu-latest

    services:
      postgres:
        image: postgres:15
        env:
          POSTGRES_PASSWORD: test
          POSTGRES_DB: pulsecart_test
        options: >-
          --health-cmd pg_isready
          --health-interval 10s
          --health-timeout 5s
          --health-retries 5

    steps:
      - uses: actions/checkout@v4

      - uses: actions/setup-python@v5
        with:
          python-version: "3.12"

      - run: pip install -r requirements.txt

      - run: pytest tests/ -v --tb=short
        env:
          DATABASE_URL: postgresql://postgres:test@localhost/pulsecart_test
          ENVIRONMENT: test
```

Real Postgres in CI via GitHub Actions services — no mocking the DB layer in integration tests, just a throwaway test database that gets created fresh on every run. Unit and contract tests run against it too, but don't touch it.

* * *

## The Three Rules

**Mock at the import site, not the definition site.** Or better — use `dependency_overrides` and skip `@patch` entirely.

**Unit tests have zero I/O.** If a unit test needs a database or an HTTP call, it's an integration test.

**Contract tests run in CI on every PR.** Schema mismatches are a merge-time problem, not a production problem.

**Next: S01E04 — FastAPI Background Tasks vs Pub/Sub vs Cloud Tasks: when each one breaks.**