FastAPI test-strategier — unit, integration og E2E tests

Robuste tests til FastAPI-applikationer kræver tre lag: unit tests med mocked dependencies, integration tests med rigtig database og E2E tests i Docker. Her er de konkrete mønstre.

---

Test-pyramid for IoT platform

E2E tests (5%):
  → Docker Compose stack (TimescaleDB, Redis, Mosquitto)
  → Fuld HTTP-request til respons
  → Fokus: Happy path for hvert brugerflow
  → Langsom — kør kun på main-merge

Integration tests (30%):
  → Rigtig test-database (PostgreSQL service i CI)
  → Rollback efter hvert test
  → Mock: Stripe, Brevo, MQTT
  → Kør på PR

Unit tests (65%):
  → Mocked dependencies
  → Ingen I/O
  → Hurtig (< 1s)
  → Kør lokalt + CI

---

pytest-asyncio opsætning

# server/tests/conftest.py

import pytest
import pytest_asyncio
from httpx import AsyncClient, ASGITransport
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker
import fakeredis.aioredis

# pytest.ini konfiguration:
# [pytest]
# asyncio_mode = auto


@pytest_asyncio.fixture(scope="session")
async def db_engine():
    """Session-scoped engine — database oprettes én gang pr. test-session."""
    engine = create_async_engine(
        settings.test_database_url,
        echo=False,
    )
    async with engine.begin() as conn:
        await conn.run_sync(SQLModel.metadata.create_all)

    yield engine

    async with engine.begin() as conn:
        await conn.run_sync(SQLModel.metadata.drop_all)
    await engine.dispose()


@pytest_asyncio.fixture
async def session(db_engine):
    """
    Function-scoped session med rollback.
    Hvert test starter med ren database.
    """
    async with db_engine.begin() as conn:
        session = AsyncSession(bind=conn)
        yield session
        await session.close()
        await conn.rollback()    # Annullér alle ændringer fra testen


@pytest_asyncio.fixture
async def redis():
    """Fake Redis — ingen rigtig Redis-server nødvendig."""
    client = fakeredis.aioredis.FakeRedis()
    yield client
    await client.aclose()

---

AsyncClient med dependency override

# server/tests/conftest.py

from fastapi.testclient import TestClient
from server.src.main import app
from server.src.db import get_session
from server.src.auth.deps import get_current_user


def make_jwt_token(user_id: str, tenant_id: str, role: str) -> str:
    """Generer JWT-token til test."""
    return jwt.encode(
        {
            "sub": user_id,
            "tenant_id": tenant_id,
            "role": role,
            "exp": int(time.time()) + 3600,
        },
        settings.jwt_secret.get_secret_value(),
        algorithm="HS256",
    )


@pytest_asyncio.fixture
async def client(session, redis):
    """AsyncClient med overridden dependencies."""

    async def override_get_session():
        yield session

    async def override_get_redis():
        return redis

    app.dependency_overrides[get_session] = override_get_session
    app.dependency_overrides[get_redis] = override_get_redis

    async with AsyncClient(
        transport=ASGITransport(app=app),
        base_url="http://test",
    ) as c:
        yield c

    app.dependency_overrides.clear()


@pytest.fixture
def landlord_headers(tenant_fixture):
    """Headers med JWT for landlord-bruger."""
    token = make_jwt_token(
        user_id=str(tenant_fixture.owner_user_id),
        tenant_id=str(tenant_fixture.id),
        role="landlord",
    )
    return {"Authorization": f"Bearer {token}"}


@pytest.fixture
def admin_headers():
    """Headers med JWT for super_admin."""
    token = make_jwt_token(
        user_id="admin-user-id",
        tenant_id=None,
        role="super_admin",
    )
    return {"Authorization": f"Bearer {token}"}

---

Database fixtures med factory

# server/tests/factories.py

import uuid
from datetime import date, datetime
from decimal import Decimal


async def create_tenant(session: AsyncSession, **kwargs) -> Tenant:
    """Factory til oprettelse af test-tenant."""
    defaults = {
        "id": uuid.uuid4(),
        "name": f"Test Tenant {uuid.uuid4().hex[:6]}",
        "created_at": datetime.utcnow(),
        "updated_at": datetime.utcnow(),
    }
    defaults.update(kwargs)
    tenant = Tenant(**defaults)
    session.add(tenant)
    await session.flush()
    return tenant


async def create_property(
    session: AsyncSession,
    tenant: Tenant,
    **kwargs,
) -> Property:
    defaults = {
        "id": uuid.uuid4(),
        "tenant_id": tenant.id,
        "name": "Test Ejendom",
        "address": "Testgade 1",
        "zip_code": "2200",
        "city": "København N",
        "total_area_m2": Decimal("500"),
        "heat_distribution_method": "hca",
        "created_at": datetime.utcnow(),
        "updated_at": datetime.utcnow(),
    }
    defaults.update(kwargs)
    prop = Property(**defaults)
    session.add(prop)
    await session.flush()
    return prop


async def create_reading(
    session: AsyncSession,
    meter_installation: MeterInstallation,
    **kwargs,
) -> Reading:
    defaults = {
        "id": uuid.uuid4(),
        "tenant_id": meter_installation.tenant_id,
        "meter_installation_id": meter_installation.id,
        "timestamp": datetime.utcnow(),
        "value": Decimal("1234.5"),
        "unit": "kWh",
        "received_at": datetime.utcnow(),
    }
    defaults.update(kwargs)
    reading = Reading(**defaults)
    session.add(reading)
    await session.flush()
    return reading

---

Integration test eksempler

# server/tests/test_readings_api.py

@pytest.mark.asyncio
async def test_list_readings_tenant_isolation(client, session, landlord_headers):
    """Landlord må kun se egne aflæsninger — ikke andres."""
    tenant_a = await create_tenant(session)
    tenant_b = await create_tenant(session)

    prop_a = await create_property(session, tenant_a)
    # ... opret meter_installation og readings for begge tenants

    # Landlord A kan se sine readings:
    resp = await client.get(
        "/api/v1/readings",
        headers=make_headers(tenant_a),
    )
    assert resp.status_code == 200
    data = resp.json()
    assert all(r["tenant_id"] == str(tenant_a.id) for r in data["items"])

    # Landlord A kan IKKE se tenant B's readings:
    assert not any(r["tenant_id"] == str(tenant_b.id) for r in data["items"])


@pytest.mark.asyncio
async def test_settlement_generation_requires_annual_input(client, session, landlord_headers):
    """Afregning fejler uden AnnualInput."""
    tenant = await create_tenant(session)
    prop = await create_property(session, tenant)

    resp = await client.post(
        f"/api/v1/properties/{prop.id}/settlements",
        json={"period_start": "2025-06-01", "period_end": "2026-05-31"},
        headers=make_headers(tenant),
    )
    assert resp.status_code == 422
    assert "annual_input" in resp.json()["message"].lower()


@pytest.mark.asyncio
async def test_rate_limit_api_key(client, redis, session):
    """API-nøgle overskrider rate limit — 429."""
    api_key = await create_api_key(session, rate_limit=10)

    for _ in range(10):
        resp = await client.get(
            "/api/v1/readings",
            headers={"X-API-Key": api_key.key},
        )
        assert resp.status_code == 200

    # 11. kald: 429
    resp = await client.get(
        "/api/v1/readings",
        headers={"X-API-Key": api_key.key},
    )
    assert resp.status_code == 429
    assert "Retry-After" in resp.headers

---

Celery eager mode i tests

# server/tests/conftest.py

from celery import current_app as celery_app


@pytest.fixture(autouse=True)
def celery_eager():
    """
    Kør Celery tasks synkront i test-miljø.
    CELERY_TASK_ALWAYS_EAGER=True: Task.delay() kører med det samme.
    """
    celery_app.config_from_object({
        "task_always_eager": True,
        "task_eager_propagates": True,    # Propagér exceptions
    })
    yield
    celery_app.config_from_object(settings.celery)    # Genopret


# Test med Celery task:
@pytest.mark.asyncio
async def test_settlement_pdf_generated(client, session, landlord_headers):
    """PDF-generering via Celery eager mode."""
    # ... setup data ...

    resp = await client.post(
        f"/api/v1/settlements/{settlement.id}/generate",
        headers=landlord_headers,
    )
    assert resp.status_code == 200

    # I eager mode er PDF allerede genereret:
    refreshed = await session.get(Settlement, settlement.id)
    assert refreshed.pdf_url is not None

---

Konklusion

FastAPI-tests kræver tre komplementære lag. Unit tests med dependency_overrides og fakeredis er hurtige og isolerede. Integration tests med rollback-fixture og PostgreSQL service-container verificerer database-logik. E2E tests i Docker Compose verificerer hele stakken. make_jwt_token() + fixture-factories reducerer boilerplate. Celery eager mode eliminerer behov for workers i tests. pytest-asyncio med asyncio_mode=auto fjerner manuel @pytest.mark.asyncio decorering.

Se pytest-asyncio guide eller FastAPI dependency injection guide.