pytest-asyncio til FastAPI — async tests og fixtures

Asynkrone FastAPI-applikationer kræver asynkrone tests. pytest-asyncio giver fuld async/await support i test-suiten.

---

Opsætning

# pyproject.toml
[tool.pytest.ini_options]
asyncio_mode = "auto"           # auto: ingen @pytest.mark.asyncio nødvendig
testpaths = ["server/tests"]

[tool.uv]
dev-dependencies = [
    "pytest>=8.0",
    "pytest-asyncio>=0.23",
    "httpx>=0.27",
    "anyio>=4.3",
]

# server/tests/conftest.py

import pytest
from httpx import AsyncClient, ASGITransport
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker
from sqlmodel import SQLModel

from src.main import app
from src.core.database import get_session
from src.core.redis import get_redis

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

@pytest.fixture(scope="session")
async def engine():
    """Session-scoped engine — én forbindelse til hele test-kørslen."""
    engine = create_async_engine(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.fixture
async def session(engine):
    """Per-test session med automatisk rollback — ingen data lækker."""
    async with engine.connect() as conn:
        await conn.begin()
        async_session = AsyncSession(bind=conn, expire_on_commit=False)
        yield async_session
        await conn.rollback()

@pytest.fixture
async def client(session):
    """AsyncClient med injiceret test-session."""
    async def override_session():
        yield session

    app.dependency_overrides[get_session] = override_session
    async with AsyncClient(
        transport=ASGITransport(app=app),
        base_url="http://test",
    ) as client:
        yield client
    app.dependency_overrides.clear()

---

JWT-autentificering i tests

# server/tests/conftest.py (fortsættelse)

from datetime import datetime, timezone, timedelta
from jose import jwt
from src.core.config import settings

def make_token(
    tenant_id: str = "550e8400-e29b-41d4-a716-446655440000",
    role: str = "landlord",
    user_id: str = "660e8400-e29b-41d4-a716-446655440000",
) -> str:
    payload = {
        "sub": user_id,
        "tenant_id": tenant_id,
        "role": role,
        "exp": datetime.now(timezone.utc) + timedelta(hours=1),
    }
    return jwt.encode(payload, settings.JWT_SECRET_KEY, algorithm="HS256")

@pytest.fixture
def landlord_headers() -> dict:
    return {"Authorization": f"Bearer {make_token(role='landlord')}"}

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

---

Test af endpoints

# server/tests/test_properties.py

import pytest
from uuid import uuid4

pytestmark = pytest.mark.asyncio

async def test_create_property(client, session, landlord_headers):
    payload = {
        "name": "Testejendommen",
        "address": "Testvej 1",
        "zip_code": "2100",
        "city": "København Ø",
        "country": "DK",
        "heat_distribution_method": "hca",
    }
    response = await client.post("/api/v1/properties", json=payload, headers=landlord_headers)
    assert response.status_code == 201
    data = response.json()
    assert data["name"] == "Testejendommen"
    assert "id" in data

async def test_get_property_not_found(client, landlord_headers):
    response = await client.get(f"/api/v1/properties/{uuid4()}", headers=landlord_headers)
    assert response.status_code == 404

async def test_list_properties_empty(client, session, landlord_headers):
    response = await client.get("/api/v1/properties", headers=landlord_headers)
    assert response.status_code == 200
    assert response.json()["items"] == []

async def test_tenant_isolation(client, session):
    """Lejer A kan ikke se lejer B's ejendom."""
    headers_a = {"Authorization": f"Bearer {make_token(tenant_id='aaaaaaaa-0000-0000-0000-000000000000')}"}
    headers_b = {"Authorization": f"Bearer {make_token(tenant_id='bbbbbbbb-0000-0000-0000-000000000000')}"}

    # Opret ejendom som tenant A
    create_r = await client.post("/api/v1/properties", json={"name": "A's ejendom", "address": "A-vej 1", "zip_code": "1000", "city": "KBH", "country": "DK"}, headers=headers_a)
    prop_id = create_r.json()["id"]

    # Forsøg at hente som tenant B
    get_r = await client.get(f"/api/v1/properties/{prop_id}", headers=headers_b)
    assert get_r.status_code == 404

---

Mock af Celery tasks

# server/tests/test_settlements.py

from unittest.mock import AsyncMock, patch

async def test_generate_settlement_triggers_celery(client, session, landlord_headers):
    """Generer afregning → Celery task dispatches."""
    with patch("src.settlements.router.generate_pdf_task.delay") as mock_delay:
        response = await client.post(
            "/api/v1/settlements/generate",
            json={"property_id": "...", "period_start": "2025-06-01", "period_end": "2026-05-31"},
            headers=landlord_headers,
        )
        assert response.status_code == 202
        mock_delay.assert_called_once()

async def test_send_settlement_email(client, session, landlord_headers):
    """Send afregning → email task dispatches."""
    with patch("src.settlements.router.send_settlement_email_task.delay") as mock_email:
        settlement_id = "..."
        response = await client.post(
            f"/api/v1/settlements/{settlement_id}/send",
            headers=landlord_headers,
        )
        assert response.status_code == 200
        mock_email.assert_called_once_with(str(settlement_id))

---

Mock af Redis

# server/tests/conftest.py

from unittest.mock import AsyncMock
import fakeredis.aioredis

@pytest.fixture
async def redis():
    """Fake Redis — ingen ekstern afhængighed."""
    redis = fakeredis.aioredis.FakeRedis()
    yield redis
    await redis.aclose()

@pytest.fixture
async def client_with_redis(session, redis):
    async def override_session():
        yield session

    async def override_redis():
        return redis

    app.dependency_overrides[get_session] = override_session
    app.dependency_overrides[get_redis] = override_redis
    async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as c:
        yield c
    app.dependency_overrides.clear()

# Test med Redis cache:
async def test_portfolio_kpi_cached(client_with_redis, redis, landlord_headers):
    # Første kald → cache miss (DB)
    r1 = await client_with_redis.get("/api/v1/portfolio/kpi", headers=landlord_headers)
    assert r1.status_code == 200

    # Andet kald → cache hit (Redis)
    r2 = await client_with_redis.get("/api/v1/portfolio/kpi", headers=landlord_headers)
    assert r2.status_code == 200
    assert r1.json() == r2.json()

---

Database fixtures med testdata

# server/tests/factories.py

from datetime import date
from uuid import uuid4
from sqlmodel import select
from src.db.models import Property, Unit, Occupancy, Meter, MeterInstallation

TENANT_ID = "550e8400-e29b-41d4-a716-446655440000"

async def create_property(session, **kwargs) -> Property:
    prop = Property(
        id=uuid4(),
        tenant_id=TENANT_ID,
        name=kwargs.get("name", "Testejendom"),
        address=kwargs.get("address", "Testvej 1"),
        zip_code="2100",
        city="KBH",
        country="DK",
    )
    session.add(prop)
    await session.flush()
    return prop

async def create_unit(session, property_id, **kwargs) -> Unit:
    unit = Unit(
        id=uuid4(),
        tenant_id=TENANT_ID,
        property_id=property_id,
        unit_number=kwargs.get("unit_number", "1"),
        floor=kwargs.get("floor", 0),
        area_m2=kwargs.get("area_m2", 75.0),
    )
    session.add(unit)
    await session.flush()
    return unit

async def create_occupancy(session, unit_id, **kwargs) -> Occupancy:
    occ = Occupancy(
        id=uuid4(),
        tenant_id=TENANT_ID,
        unit_id=unit_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 Testersen"),
        tenant_email=kwargs.get("tenant_email", "test@example.com"),
    )
    session.add(occ)
    await session.flush()
    return occ

---

CI integration

# .github/workflows/test-server.yml (uddrag)

- name: Start TimescaleDB
  run: |
    docker run -d \
      --name timescaledb-test \
      -e POSTGRES_PASSWORD=mbus \
      -e POSTGRES_USER=mbus \
      -e POSTGRES_DB=mbus_test \
      -p 5433:5432 \
      timescale/timescaledb:latest-pg16
    sleep 5

- name: Run tests
  env:
    TEST_DATABASE_URL: postgresql+asyncpg://mbus:mbus@localhost:5433/mbus_test
    JWT_SECRET_KEY: test-secret-key-not-for-production
  run: |
    cd server
    uv run pytest --tb=short -q

---

Konklusion

asyncio_mode = "auto" eliminerer gentagne @pytest.mark.asyncio dekoratorer. Rollback-fixture sikrer test-isolation uden truncation. ASGITransport + AsyncClient tester det fulde middleware-stack inkl. JWT og rate limiting. fakeredis erstatter Redis i tests uden ekstern afhængighed.

Se FastAPI multi-tenant guide eller SQLAlchemy async patterns.