MABEA/backend/tests/conftest.py

import pytest
import pytest_asyncio
from httpx import ASGITransport, AsyncClient
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine

import app.models  # noqa: F401  (registriert ALLE ORM-Modelle an Base.metadata, auch
# solche, die noch von keinem API-Endpunkt importiert werden - sonst schlägt die
# FK-Auflösung für andere Modelle fehl, sobald z. B. Objekt.vorlage_id auf eine noch
# nicht importierte Tabelle zeigt. Fund im dritten echten CI-Lauf.)
from app.core.app_settings import settings
from app.core.security import hash_password
from app.core.rate_limit import limiter
from app.db.session import get_db
from app.main import app

# Login-Rate-Limit (5/Minute, Karte owasp-Review) würde bei vielen Tests, die
# login() mehrfach aufrufen, den Testlauf selbst blockieren - hier bewusst
# global für die Testsuite deaktiviert, Produktivverhalten bleibt unberührt.
limiter.enabled = False
from app.models.auth import Benutzer, BenutzerRolle, RolleTyp
from app.models.objekt import Objekt
from app.models.stammdaten import Bereich, Material, MaterialTyp, Objekttyp, Standort


@pytest_asyncio.fixture
async def db_session():
    """Jeder Test läuft in einer äußeren Transaktion, die am Ende zurückgerollt wird.

    Die Engine wird bewusst PRO TEST neu erzeugt (nicht als Modul-Global): asyncpg-
    Verbindungen sind an die Event-Loop gebunden, in der sie entstanden – ein
    session-weit wiederverwendeter Pool kollidierte mit pytest-asyncios (Default)
    function-scoped Loop pro Test ("attached to a different loop" / "another
    operation is in progress", Fund im zweiten echten CI-Lauf). Mit einer frischen
    Engine pro Test entsteht der Pool immer auf der Loop, die auch den Test ausführt.

    Die Session ist zusätzlich per `join_transaction_mode="create_savepoint"` an die
    äußere Connection gebunden: ein `commit()` innerhalb des Tests (z. B. durch
    `get_db`, das jetzt selbst committet) schließt nur eine SAVEPOINT ab, nicht die
    äußere Transaktion – das äußere `connection.rollback()` verwirft am Ende trotzdem
    alles (SQLAlchemy 2.0 "Joining a Session into an External Transaction").
    """
    engine = create_async_engine(settings.database_url, pool_pre_ping=True)
    try:
        async with engine.connect() as connection:
            await connection.begin()
            session = AsyncSession(
                bind=connection, expire_on_commit=False, join_transaction_mode="create_savepoint"
            )
            try:
                yield session
            finally:
                await session.close()
                await connection.rollback()
    finally:
        await engine.dispose()


@pytest_asyncio.fixture
async def client(db_session):
    async def _get_db_override():
        yield db_session

    app.dependency_overrides[get_db] = _get_db_override
    transport = ASGITransport(app=app)
    async with AsyncClient(transport=transport, base_url="http://test") as ac:
        yield ac
    app.dependency_overrides.clear()


@pytest_asyncio.fixture
async def mitarbeiter_user(db_session):
    benutzer = Benutzer(
        name="Test Mitarbeiter",
        login="mitarbeiter1",
        passwort_hash=hash_password("test-passwort-123"),
        aktiv=True,
    )
    db_session.add(benutzer)
    await db_session.flush()
    db_session.add(BenutzerRolle(benutzer_id=benutzer.id, rolle=RolleTyp.mitarbeiter))
    await db_session.flush()
    return benutzer


@pytest_asyncio.fixture
async def mitarbeiter2_user(db_session):
    benutzer = Benutzer(
        name="Test Mitarbeiter 2",
        login="mitarbeiter2",
        passwort_hash=hash_password("test-passwort-123"),
        aktiv=True,
    )
    db_session.add(benutzer)
    await db_session.flush()
    db_session.add(BenutzerRolle(benutzer_id=benutzer.id, rolle=RolleTyp.mitarbeiter))
    await db_session.flush()
    return benutzer


@pytest_asyncio.fixture
async def materialverantwortlicher_user(db_session):
    benutzer = Benutzer(
        name="Test Materialverantwortlicher",
        login="materialverantwortlicher1",
        email="mv1@example.org",
        passwort_hash=hash_password("test-passwort-123"),
        aktiv=True,
    )
    db_session.add(benutzer)
    await db_session.flush()
    db_session.add(BenutzerRolle(benutzer_id=benutzer.id, rolle=RolleTyp.materialverantwortlicher))
    await db_session.flush()
    return benutzer


@pytest_asyncio.fixture
async def admin_user(db_session):
    benutzer = Benutzer(
        name="Test Administration",
        login="admin1",
        passwort_hash=hash_password("test-passwort-123"),
        aktiv=True,
    )
    db_session.add(benutzer)
    await db_session.flush()
    db_session.add(BenutzerRolle(benutzer_id=benutzer.id, rolle=RolleTyp.administration))
    await db_session.flush()
    return benutzer


async def login(client, username: str, password: str = "test-passwort-123") -> str:
    response = await client.post(
        "/api/v1/auth/login", data={"username": username, "password": password}
    )
    assert response.status_code == 200, response.text
    return response.json()["access_token"]


def auth_header(token: str) -> dict:
    return {"Authorization": f"Bearer {token}"}


@pytest_asyncio.fixture
async def hauptserver_id(db_session):
    # Seed-Migration 0002 legt genau einen Datensatz mit typ='haupt' an (Sprintplan E6).
    from sqlalchemy import select

    from app.models.auth import KnotenTyp, Systemknoten

    result = await db_session.execute(
        select(Systemknoten.id).where(Systemknoten.typ == KnotenTyp.haupt)
    )
    return result.scalar_one()


@pytest_asyncio.fixture
async def standort_factory(db_session):
    async def _make(name: str) -> Standort:
        standort = Standort(name=name)
        db_session.add(standort)
        await db_session.flush()
        return standort

    return _make


@pytest_asyncio.fixture
async def objekt_factory(db_session, hauptserver_id):
    async def _make(*, name: str, code: str, standort: Standort) -> Objekt:
        bereich = Bereich(name=f"Bereich-{code}")
        db_session.add(bereich)
        await db_session.flush()
        objekttyp = Objekttyp(bereich_id=bereich.id, name=f"Typ-{code}")
        db_session.add(objekttyp)
        await db_session.flush()
        objekt = Objekt(
            code=code,
            name=name,
            objekttyp_id=objekttyp.id,
            standort_id=standort.id,
            zustaendiger_server_id=hauptserver_id,
        )
        db_session.add(objekt)
        await db_session.flush()
        return objekt

    return _make


@pytest_asyncio.fixture
async def material_factory(db_session):
    async def _make(name: str) -> Material:
        material = Material(name=name, einheit="Stück", materialtyp=MaterialTyp.standard)
        db_session.add(material)
        await db_session.flush()
        return material

    return _make


@pytest_asyncio.fixture
async def objekttyp_factory(db_session):
    async def _make(name: str) -> Objekttyp:
        bereich = Bereich(name=f"Bereich-{name}")
        db_session.add(bereich)
        await db_session.flush()
        objekttyp = Objekttyp(bereich_id=bereich.id, name=name)
        db_session.add(objekttyp)
        await db_session.flush()
        return objekttyp

    return _make


@pytest_asyncio.fixture
async def objekt_mit_position(db_session, hauptserver_id):
    """Fertiges Objekt mit einer Vorlagen-Position (Sollmenge 10) für Kontroll-Tests."""
    from app.services.objekte import objekt_aus_vorlage
    from app.services.vorlagen import erstelle_vorlage
    from app.schemas.vorlage import VorlagenpositionCreate

    bereich = Bereich(name="Bereich-KT")
    db_session.add(bereich)
    await db_session.flush()
    objekttyp = Objekttyp(bereich_id=bereich.id, name="Rucksack-KT")
    db_session.add(objekttyp)
    await db_session.flush()
    material = Material(name="Kompresse", einheit="Stück", materialtyp=MaterialTyp.standard)
    db_session.add(material)
    await db_session.flush()
    standort = Standort(name="Wache-KT")
    db_session.add(standort)
    await db_session.flush()

    vorlage = await erstelle_vorlage(
        db_session,
        objekttyp_id=objekttyp.id,
        name="Vorlage-KT",
        positionen=[VorlagenpositionCreate(material_id=material.id, sollmenge=10)],
    )
    objekt = await objekt_aus_vorlage(
        db_session,
        code="KT-01",
        name="Kontroll-Test-Objekt",
        standort_id=standort.id,
        vorlage=vorlage,
        zustaendiger_server_id=hauptserver_id,
    )
    return objekt, material