M-Bus Gateway
← Tilbage til blog
· FastAPI· testing· pytest· Python· integration tests· async· SQLAlchemy· backend

FastAPI testing mønstre — integration og unit tests i produktion

FastAPI testing mønstre: AsyncClient fixtures, database rollback pr. test, JWT auth helpers, mock external APIs, parametrize og factory functions til multi-tenant tests.

Af M-Bus Gateway

Teststrategier til FastAPI SaaS-applikationer med multi-tenancy, JWT auth og async database.


Basis-fixtures

# server/tests/conftest.py

import pytest
import pytest_asyncio
from httpx import AsyncClient, ASGITransport
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker
from server.src.main import app
from server.src.db import get_session, Base
from server.src.config import get_settings, Settings

TEST_DATABASE_URL = "postgresql+asyncpg://mbus:test@localhost:5432/mbus_test"


@pytest.fixture(scope="session")
def event_loop_policy():
    import asyncio
    return asyncio.DefaultEventLoopPolicy()


@pytest_asyncio.fixture(scope="session")
async def engine():
    """Test database engine — opret alle tabeller én gang pr. test-session."""
    engine = create_async_engine(TEST_DATABASE_URL, echo=False)
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    yield engine
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.drop_all)
    await engine.dispose()


@pytest_asyncio.fixture
async def session(engine):
    """
    Database session med rollback efter hver test.
    Bruger SAVEPOINT for at tillade nested transactions i koden.
    """
    async with engine.connect() as conn:
        await conn.begin()
        session = AsyncSession(bind=conn)
        await conn.begin_nested()  # SAVEPOINT

        yield session

        await session.close()
        await conn.rollback()  # Alt slettes efter testen


@pytest_asyncio.fixture
async def client(session):
    """AsyncClient med session-override."""
    async def override_get_session():
        yield session

    app.dependency_overrides[get_session] = override_get_session
    app.dependency_overrides[get_settings] = lambda: Settings(
        jwt_secret="test-secret-min-32-characters!!",
        database_url=TEST_DATABASE_URL,
        environment="test",
        stripe_secret_key="sk_test_dummy",
        brevo_api_key="",
    )

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

    app.dependency_overrides.clear()

JWT auth helpers

# server/tests/helpers/auth.py

from datetime import datetime, timedelta
from jose import jwt
from uuid import uuid4, UUID

TEST_JWT_SECRET = "test-secret-min-32-characters!!"
TEST_JWT_ALGORITHM = "HS256"


def make_token(
    role: str = "landlord",
    tenant_id: str | None = None,
    user_id: str | None = None,
    expire_minutes: int = 60,
) -> str:
    """Generer JWT token til tests — ingen database-opslag."""
    payload = {
        "sub": user_id or str(uuid4()),
        "role": role,
        "tenant_id": tenant_id or str(uuid4()),
        "exp": datetime.utcnow() + timedelta(minutes=expire_minutes),
    }
    return jwt.encode(payload, TEST_JWT_SECRET, algorithm=TEST_JWT_ALGORITHM)


def auth_headers(role: str = "landlord", tenant_id: str | None = None) -> dict:
    return {"Authorization": f"Bearer {make_token(role=role, tenant_id=tenant_id)}"}


# Fixtures:
@pytest.fixture
def landlord_headers(test_tenant) -> dict:
    return auth_headers(role="landlord", tenant_id=str(test_tenant.id))

@pytest.fixture
def super_admin_headers() -> dict:
    return {"Authorization": f"Bearer {make_token(role='super_admin', tenant_id=None)}"}

@pytest.fixture
def technician_headers(test_tenant) -> dict:
    return auth_headers(role="technician", tenant_id=str(test_tenant.id))

Factory functions til test-data

# server/tests/factories.py

from datetime import date
from decimal import Decimal
from uuid import uuid4
from server.src.db.models import Tenant, Property, Unit, Occupancy, Gateway, Meter


async def make_tenant(session, name: str = "Test Udlejer A/S") -> Tenant:
    t = Tenant(id=uuid4(), name=name, email=f"test+{uuid4().hex[:6]}@example.com")
    session.add(t)
    await session.flush()
    return t


async def make_property(session, tenant_id, **kwargs) -> Property:
    p = Property(
        id=uuid4(),
        tenant_id=tenant_id,
        name=kwargs.get("name", "Testgården 1-3"),
        address=kwargs.get("address", "Testvej 1"),
        zip_code=kwargs.get("zip_code", "2100"),
        city=kwargs.get("city", "København Ø"),
        area_m2=kwargs.get("area_m2", 500.0),
        heat_distribution_method=kwargs.get("heat_distribution_method", "hca"),
    )
    session.add(p)
    await session.flush()
    return p


async def make_unit(session, property_id, tenant_id, **kwargs) -> Unit:
    u = Unit(
        id=uuid4(),
        property_id=property_id,
        tenant_id=tenant_id,
        label=kwargs.get("label", f"Lejlighed {uuid4().hex[:3]}"),
        floor=kwargs.get("floor", 1),
        area_m2=kwargs.get("area_m2", 75.0),
    )
    session.add(u)
    await session.flush()
    return u


async def make_occupancy(session, unit_id, tenant_id, **kwargs) -> Occupancy:
    o = Occupancy(
        id=uuid4(),
        unit_id=unit_id,
        tenant_id=tenant_id,
        start_date=kwargs.get("start_date", date(2025, 6, 1)),
        end_date=kwargs.get("end_date", None),
        tenant_name=kwargs.get("tenant_name", "Test Lejer"),
        tenant_email=kwargs.get("tenant_email", "lejer@example.com"),
        monthly_heat_advance=kwargs.get("monthly_heat_advance", Decimal("1200")),
    )
    session.add(o)
    await session.flush()
    return o

Parametriserede tests

# server/tests/test_properties.py

import pytest
from httpx import AsyncClient


@pytest.mark.parametrize("role,expected_status", [
    ("landlord", 200),
    ("technician", 403),
    ("super_admin", 200),
])
async def test_list_properties_by_role(
    client: AsyncClient,
    test_property,
    role: str,
    expected_status: int,
):
    headers = auth_headers(role=role, tenant_id=str(test_property.tenant_id))
    resp = await client.get("/api/v1/properties", headers=headers)
    assert resp.status_code == expected_status


@pytest.mark.parametrize("field,value,expected_status", [
    ("name", "", 422),
    ("name", "A" * 201, 422),
    ("zip_code", "12345", 422),   # Dansk postnummer: 4 cifre
    ("zip_code", "2100", 200),
    ("area_m2", -1.0, 422),
    ("area_m2", 50.0, 200),
])
async def test_create_property_validation(
    client: AsyncClient,
    landlord_headers: dict,
    field: str,
    value,
    expected_status: int,
):
    payload = {
        "name": "Testgården",
        "address": "Testvej 1",
        "zip_code": "2100",
        "city": "København",
        "area_m2": 300.0,
    }
    payload[field] = value
    resp = await client.post("/api/v1/properties", json=payload, headers=landlord_headers)
    assert resp.status_code == expected_status

Mock externe API'er

# server/tests/test_dmi.py

import pytest
from unittest.mock import AsyncMock, patch


@pytest.fixture
def mock_dmi_response():
    return {
        "features": [
            {
                "properties": {
                    "from": "2026-01-01T00:00:00Z",
                    "to": "2026-01-31T23:59:59Z",
                    "value": 280.5,  # Graddage
                }
            }
        ]
    }


async def test_sync_degree_days(client, mock_dmi_response, landlord_headers):
    with patch(
        "server.src.dmi.client.DMIClient.fetch_degree_days",
        new_callable=AsyncMock,
        return_value=mock_dmi_response,
    ) as mock_fetch:
        resp = await client.post(
            "/api/v1/dmi/sync",
            json={"zip_code": "2100", "year": 2026},
            headers=landlord_headers,
        )
        assert resp.status_code == 200
        mock_fetch.assert_called_once_with("2100", 2026)


async def test_stripe_webhook_invoice_paid(client):
    payload = {
        "type": "invoice.paid",
        "data": {"object": {"customer": "cus_test123", "status": "paid"}},
    }
    with patch("stripe.Webhook.construct_event", return_value=payload):
        resp = await client.post(
            "/api/v1/stripe/webhook",
            content=b'{"type":"invoice.paid"}',
            headers={"stripe-signature": "dummy"},
        )
        assert resp.status_code == 200

Konklusion

Effektive FastAPI tests bruger SAVEPOINT-baseret rollback pr. test (ingen database-rens overhead), JWT helpers der genererer tokens uden database-opslag, og factory functions der opretter test-data med minimalt boilerplate. Parametriserede tests dækker validerings- og RBAC-scenarier effektivt. Mock externe API'er (DMI, Stripe, Brevo) for deterministiske tests.

Se pytest-asyncio FastAPI testing guide eller FastAPI dependency injection guide.