Python FastAPI testing — mønstre til IoT og SaaS

FastAPI-tests kræver async HTTP-klient, JWT-mocking og databaseisolering. Her er mønstrene der virker i produktion.

---

ASGITransport — in-process testning

# 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 create_app
from server.src.db.engine import get_db


# Brug SQLite i tests (ingen Docker krævet)
TEST_DATABASE_URL = "sqlite+aiosqlite:///:memory:"


@pytest_asyncio.fixture(scope="session")
async def test_engine():
    engine = create_async_engine(TEST_DATABASE_URL, echo=False)
    async with engine.begin() as conn:
        from server.src.db.models import SQLModel
        await conn.run_sync(SQLModel.metadata.create_all)
    yield engine
    await engine.dispose()


@pytest_asyncio.fixture
async def db_session(test_engine):
    """
    Isoleret DB-session pr. test.
    Alle ændringer rulles tilbage — ingen data lækkager mellem tests.
    """
    async_session = async_sessionmaker(test_engine, expire_on_commit=False)

    async with test_engine.connect() as conn:
        await conn.begin()
        async with AsyncSession(bind=conn) as session:
            yield session
            await conn.rollback()


@pytest_asyncio.fixture
async def client(db_session):
    """
    FastAPI TestClient via ASGITransport.
    Kører in-process — ingen netværksforbindelse.
    """
    app = create_app()

    # Override DB dependency til isoleret test-session
    async def override_get_db():
        yield db_session

    app.dependency_overrides[get_db] = override_get_db

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

    app.dependency_overrides.clear()

---

JWT dependency override

# server/tests/conftest.py
from server.src.auth.deps import get_current_user
from server.src.db.models import User


def make_jwt_override(
    user_id: str = "test-user-id",
    tenant_id: str = "test-tenant-id",
    role: str = "landlord",
):
    """
    Returner en dependency override der simulerer en authentikeret bruger.
    Bruges til at teste RBAC-logik uden rigtig JWT.
    """
    async def override():
        return User(
            id=user_id,
            tenant_id=tenant_id,
            role=role,
            email="test@example.com",
            name="Test User",
        )
    return override


@pytest.fixture
def landlord_client(client, app):
    """Client med landlord-JWT."""
    app.dependency_overrides[get_current_user] = make_jwt_override(role="landlord")
    yield client
    app.dependency_overrides[get_current_user] = make_jwt_override(role="landlord")


@pytest.fixture
def super_admin_client(client, app):
    """Client med super_admin-JWT."""
    app.dependency_overrides[get_current_user] = make_jwt_override(role="super_admin")
    yield client

---

Test-eksempler med parametrize

# server/tests/test_portfolio_kpi.py
import pytest
from decimal import Decimal
from unittest.mock import AsyncMock, patch


@pytest.mark.asyncio
@pytest.mark.parametrize("role,expected_status", [
    ("landlord", 200),
    ("technician", 200),
    ("installer", 403),
    ("tenant", 403),
])
async def test_portfolio_kpi_rbac(
    client,
    app,
    role: str,
    expected_status: int,
):
    """RBAC: Kun landlord og technician må se portfolio KPI."""
    from server.src.auth.deps import get_current_user
    app.dependency_overrides[get_current_user] = make_jwt_override(role=role)

    response = await client.get("/api/v1/portfolio/kpi")
    assert response.status_code == expected_status


@pytest.mark.asyncio
@pytest.mark.parametrize("property_count,unit_count,expected_tier", [
    (1, 10, "small"),
    (5, 100, "medium"),
    (20, 500, "large"),
])
async def test_portfolio_kpi_tiers(
    client,
    db_session,
    property_count: int,
    unit_count: int,
    expected_tier: str,
):
    """KPI tier-beregning baseret på porteføljestørrelse."""
    # Seed testdata
    await _seed_properties(db_session, property_count, unit_count)

    response = await client.get("/api/v1/portfolio/kpi")
    assert response.status_code == 200
    data = response.json()
    assert data["tier"] == expected_tier

---

respx mock til httpx-kald

# server/tests/test_dmi_sync.py
import pytest
import respx
import httpx
from server.src.workers.tasks.dmi import sync_degree_days_for_zip


@pytest.mark.asyncio
async def test_dmi_sync_success():
    """Mock DMI API — kræver ikke rigtig netværksadgang."""
    mock_response = {
        "timeSeriesData": [
            {"time": "2026-05-01T00:00:00Z", "value": 12.5},
            {"time": "2026-05-02T00:00:00Z", "value": 8.3},
        ]
    }

    with respx.mock:
        respx.get("https://dmigw.govcloud.dk/v2/climateData/").mock(
            return_value=httpx.Response(200, json=mock_response)
        )

        result = await sync_degree_days_for_zip("2100", "2026-05-01", "2026-05-31")

    assert result["days_synced"] == 2
    assert result["zip_code"] == "2100"


@pytest.mark.asyncio
async def test_dmi_sync_api_error():
    """DMI API nede → graceful fejl, ikke exception."""
    with respx.mock:
        respx.get("https://dmigw.govcloud.dk/v2/climateData/").mock(
            return_value=httpx.Response(503, text="Service Unavailable")
        )

        result = await sync_degree_days_for_zip("2100", "2026-05-01", "2026-05-31")

    assert result["status"] == "error"
    assert result["days_synced"] == 0


@pytest.mark.asyncio
async def test_dmi_sync_timeout():
    """DMI timeout → retry og til sidst fejl."""
    with respx.mock:
        respx.get("https://dmigw.govcloud.dk/v2/climateData/").mock(
            side_effect=httpx.TimeoutException("Connection timed out")
        )

        result = await sync_degree_days_for_zip("2100", "2026-05-01", "2026-05-31")

    assert result["status"] == "timeout"

---

Database fixtures — seed og isolation

# server/tests/factories.py
from datetime import date, datetime, timedelta
from decimal import Decimal
import uuid
from server.src.db.models import Property, Unit, Meter, MeterInstallation, Reading


async def create_test_property(
    db,
    tenant_id: str = "test-tenant",
    name: str = "Testejendom",
    zip_code: str = "2100",
) -> Property:
    prop = Property(
        id=str(uuid.uuid4()),
        tenant_id=tenant_id,
        name=name,
        address=f"Testvej 1, {zip_code} København",
        zip_code=zip_code,
        country="DK",
        created_at=datetime.utcnow(),
        updated_at=datetime.utcnow(),
    )
    db.add(prop)
    await db.flush()
    return prop


async def create_test_reading(
    db,
    meter_installation_id: str,
    value_kwh: float = 100.0,
    days_ago: int = 1,
) -> Reading:
    reading = Reading(
        id=str(uuid.uuid4()),
        meter_installation_id=meter_installation_id,
        timestamp=datetime.utcnow() - timedelta(days=days_ago),
        value_kwh=Decimal(str(value_kwh)),
        received_at=datetime.utcnow(),
    )
    db.add(reading)
    await db.flush()
    return reading


# I test:
# async def test_something(db_session):
#     prop = await create_test_property(db_session, tenant_id="t1")
#     reading = await create_test_reading(db_session, mi_id, value_kwh=234.5)
#     await db_session.commit()

---

Pytest konfiguration

# pytest.ini
[pytest]
asyncio_mode = auto
testpaths = server/tests
python_files = test_*.py
python_classes = Test*
python_functions = test_*

# Markers
markers =
    slow: Tests der tager >1 sekund (kræver --run-slow)
    integration: Integration tests (kræver ekstern service)

# Ignorer warnings fra tredjepartsbiblioteker
filterwarnings =
    ignore::DeprecationWarning:sqlalchemy
    ignore::DeprecationWarning:pydantic

# Kør unit tests hurtigt (CI baseline):
pytest server/tests/ -x --timeout=30 -q

# Kør med coverage:
pytest server/tests/ --cov=server/src --cov-report=term-missing -q

# Kør kun hurtige tests (ekskludér slow):
pytest server/tests/ -m "not slow" -q

---

Konklusion

ASGITransport + AsyncClient giver in-process FastAPI-tests uden netværk — korrekt isolering, hurtig eksekvering. db_session fixture med rollback sikrer at tests aldrig påvirker hinanden. JWT dependency_overrides giver fuld RBAC-kontrol i tests uden rigtig token-generering. parametrize er det rette valg til RBAC-matrixtests (rolle × endpoint × forventet status). respx mock til httpx-kald (DMI, Stripe, Brevo) eliminerer ekstern afhængighed i tests.

Se Python httpx async guide eller FastAPI lifespan guide.