MABEA/backend/tests/test_historie.py

import pytest
from sqlalchemy import select

from app.models.historie import Historie
from tests.conftest import auth_header, login


@pytest.mark.asyncio
async def test_historie_kette_soll10_ist8_genehmigt_dann_nachgefuellt(
    client, db_session, objekt_mit_position, mitarbeiter_user, materialverantwortlicher_user
):
    """U14/Prompt 13.3 Beispiel-Kette: Soll 10, Ist 8 -> Fehlbestand entstanden ->
    keine Nachfüllung möglich -> Mindermenge genehmigt -> später 2 nachgefüllt ->
    Ist wieder 10 -> Fehlbestand erledigt, Genehmigung endet gemeinsam. Jeder
    Schritt muss als chronologisch korrekter Historieneintrag auffindbar sein."""
    objekt, material = objekt_mit_position
    mitarbeiter_token = await login(client, "mitarbeiter1")

    start = await client.post(
        f"/api/v1/objekte/{objekt.id}/kontrollen", json={"uebernehmen": False}, headers=auth_header(mitarbeiter_token)
    )
    kontrolle_id = start.json()["id"]

    put_response = await client.put(
        f"/api/v1/kontrollen/{kontrolle_id}/positionen/{material.id}",
        json={"istmenge": "8"},
        headers=auth_header(mitarbeiter_token),
    )
    fehlbestand_id = put_response.json()["fehlbestand_id"]
    assert fehlbestand_id is not None

    await client.post(f"/api/v1/kontrollen/{kontrolle_id}/abschliessen", headers=auth_header(mitarbeiter_token))

    verantwortlicher_token = await login(client, "materialverantwortlicher1")
    await client.post(
        f"/api/v1/fehlbestaende/{fehlbestand_id}/mindermenge",
        json={"begruendung": "Lieferengpass, Ersatz bestellt"},
        headers=auth_header(verantwortlicher_token),
    )

    await client.post(
        f"/api/v1/fehlbestaende/{fehlbestand_id}/nachfuellungen",
        json={"menge": "2"},
        headers=auth_header(mitarbeiter_token),
    )

    result = await db_session.execute(select(Historie).order_by(Historie.zeitpunkt))
    ereignisse = [h.ereignistyp for h in result.scalars().all()]

    erwartete_reihenfolge = [
        "kontrolle_gestartet",
        "istmenge_erfasst",
        "fehlbestand_entstanden",
        "kontrolle_abgeschlossen",
        "mindermenge_genehmigt",
        "nachfuellung_erfasst",
        "fehlbestand_erledigt",
        "mindermenge_beendet_durch_erledigung",
    ]
    # Reine Teilfolge-Prüfung (keine anderen Events dazwischen erwartet, aber falls
    # doch, ist die Reihenfolge der genannten Ereignisse das eigentlich Wichtige).
    assert ereignisse == erwartete_reihenfolge


@pytest.mark.asyncio
async def test_historie_endpunkt_liefert_gefiltert(
    client, db_session, objekt_mit_position, mitarbeiter_user, materialverantwortlicher_user
):
    objekt, material = objekt_mit_position
    mitarbeiter_token = await login(client, "mitarbeiter1")

    start = await client.post(
        f"/api/v1/objekte/{objekt.id}/kontrollen", json={"uebernehmen": False}, headers=auth_header(mitarbeiter_token)
    )
    kontrolle_id = start.json()["id"]
    await client.put(
        f"/api/v1/kontrollen/{kontrolle_id}/positionen/{material.id}",
        json={"istmenge": "10"},
        headers=auth_header(mitarbeiter_token),
    )

    verantwortlicher_token = await login(client, "materialverantwortlicher1")
    response = await client.get(
        "/api/v1/historie",
        params={"entitaet_typ": "kontrolle", "entitaet_id": kontrolle_id},
        headers=auth_header(verantwortlicher_token),
    )
    assert response.status_code == 200
    body = response.json()
    assert len(body) >= 1
    assert all(e["entitaet_typ"] == "kontrolle" for e in body)
    assert body[0]["benutzer_name"] == "Test Mitarbeiter"


@pytest.mark.asyncio
async def test_historie_mitarbeiter_ohne_zugriff(client, mitarbeiter_user):
    token = await login(client, "mitarbeiter1")
    response = await client.get("/api/v1/historie", headers=auth_header(token))
    assert response.status_code == 403


@pytest.mark.asyncio
async def test_hash_kette_intakt_im_normalfall(
    client, db_session, objekt_mit_position, mitarbeiter_user, admin_user
):
    objekt, material = objekt_mit_position
    mitarbeiter_token = await login(client, "mitarbeiter1")

    start = await client.post(
        f"/api/v1/objekte/{objekt.id}/kontrollen", json={"uebernehmen": False}, headers=auth_header(mitarbeiter_token)
    )
    kontrolle_id = start.json()["id"]
    await client.put(
        f"/api/v1/kontrollen/{kontrolle_id}/positionen/{material.id}",
        json={"istmenge": "10"},
        headers=auth_header(mitarbeiter_token),
    )

    admin_token = await login(client, "admin1")
    response = await client.get(
        "/api/v1/historie/verifikation",
        params={"entitaet_typ": "kontrolle", "entitaet_id": kontrolle_id},
        headers=auth_header(admin_token),
    )
    assert response.status_code == 200
    body = response.json()
    assert body["intakt"] is True
    assert body["geprueft"] >= 1
    assert body["bruch_bei_eintrag_id"] is None


@pytest.mark.asyncio
async def test_hash_kette_erkennt_manipulation(
    client, db_session, objekt_mit_position, mitarbeiter_user, admin_user
):
    objekt, material = objekt_mit_position
    mitarbeiter_token = await login(client, "mitarbeiter1")

    start = await client.post(
        f"/api/v1/objekte/{objekt.id}/kontrollen", json={"uebernehmen": False}, headers=auth_header(mitarbeiter_token)
    )
    kontrolle_id = start.json()["id"]
    await client.put(
        f"/api/v1/kontrollen/{kontrolle_id}/positionen/{material.id}",
        json={"istmenge": "10"},
        headers=auth_header(mitarbeiter_token),
    )

    erster_eintrag = (
        await db_session.execute(
            select(Historie)
            .where(Historie.entitaet_typ == "kontrolle", Historie.entitaet_id == str(kontrolle_id))
            .order_by(Historie.zeitpunkt)
        )
    ).scalars().first()
    erster_eintrag.begruendung = "nachträglich manipuliert"
    await db_session.commit()

    admin_token = await login(client, "admin1")
    response = await client.get(
        "/api/v1/historie/verifikation",
        params={"entitaet_typ": "kontrolle", "entitaet_id": kontrolle_id},
        headers=auth_header(admin_token),
    )
    body = response.json()
    assert body["intakt"] is False
    assert body["bruch_bei_eintrag_id"] == str(erster_eintrag.id)


@pytest.mark.asyncio
async def test_hash_kette_verifikation_nur_admin(client, materialverantwortlicher_user):
    token = await login(client, "materialverantwortlicher1")
    response = await client.get(
        "/api/v1/historie/verifikation",
        params={"entitaet_typ": "kontrolle", "entitaet_id": "irgendwas"},
        headers=auth_header(token),
    )
    assert response.status_code == 403