· 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.