MABEA/backend/tests/test_custody.py

import io

import pytest
from PIL import Image

from tests.conftest import auth_header, login


def _signatur_png_bytes() -> bytes:
    bild = Image.new("RGB", (100, 40), color="white")
    puffer = io.BytesIO()
    bild.save(puffer, format="PNG")
    return puffer.getvalue()


async def _geraet_instanz(client, db_session, objekt_mit_position, admin_token):
    objekt, material = objekt_mit_position
    positionen = await client.get(f"/api/v1/objekte/{objekt.id}/positionen", headers=auth_header(admin_token))
    position_id = positionen.json()[0]["id"]
    erstellt = await client.post(
        f"/api/v1/objektpositionen/{position_id}/geraete",
        json={"seriennummer": "SN-CUSTODY-1"},
        headers=auth_header(admin_token),
    )
    return erstellt.json()["id"]


async def _person(db_session):
    from app.models.personal import Person

    person = Person(name="Max Mustermann")
    db_session.add(person)
    await db_session.flush()
    await db_session.commit()
    return person


@pytest.mark.asyncio
async def test_formlose_ausgabe_sofort_abgeschlossen(
    client, db_session, objekt_mit_position, mitarbeiter_user, admin_user
):
    admin_token = await login(client, "admin1")
    token = await login(client, "mitarbeiter1")
    geraet_id = await _geraet_instanz(client, db_session, objekt_mit_position, admin_token)
    person = await _person(db_session)

    response = await client.post(
        f"/api/v1/geraete/{geraet_id}/ausgeben",
        json={"person_id": person.id, "mit_protokoll": False},
        headers=auth_header(token),
    )
    assert response.status_code == 201
    body = response.json()
    assert body["signier_url"] is None
    assert body["custody"]["ausgabe_abgeschlossen_am"] is not None


@pytest.mark.asyncio
async def test_doppelausgabe_abgelehnt(client, db_session, objekt_mit_position, mitarbeiter_user, admin_user):
    admin_token = await login(client, "admin1")
    token = await login(client, "mitarbeiter1")
    geraet_id = await _geraet_instanz(client, db_session, objekt_mit_position, admin_token)
    person_a = await _person(db_session)
    person_b = await _person(db_session)

    erste = await client.post(
        f"/api/v1/geraete/{geraet_id}/ausgeben",
        json={"person_id": person_a.id, "mit_protokoll": False},
        headers=auth_header(token),
    )
    assert erste.status_code == 201

    zweite = await client.post(
        f"/api/v1/geraete/{geraet_id}/ausgeben",
        json={"person_id": person_b.id, "mit_protokoll": False},
        headers=auth_header(token),
    )
    assert zweite.status_code == 409


@pytest.mark.asyncio
async def test_ausgabe_mit_protokoll_wartet_auf_signatur(
    client, db_session, objekt_mit_position, mitarbeiter_user, admin_user
):
    admin_token = await login(client, "admin1")
    token = await login(client, "mitarbeiter1")
    geraet_id = await _geraet_instanz(client, db_session, objekt_mit_position, admin_token)
    person = await _person(db_session)

    ausgabe = await client.post(
        f"/api/v1/geraete/{geraet_id}/ausgeben",
        json={"person_id": person.id, "mit_protokoll": True},
        headers=auth_header(token),
    )
    assert ausgabe.status_code == 201
    body = ausgabe.json()
    assert body["signier_url"] is not None
    assert body["custody"]["ausgabe_abgeschlossen_am"] is None

    custody_id = body["custody"]["id"]
    status_antwort = await client.get(f"/api/v1/custody/{custody_id}/status", headers=auth_header(token))
    assert status_antwort.json()["abgeschlossen"] is False

    signier_token = body["signier_url"].rsplit("/", 1)[-1]

    seite = await client.get(f"/api/v1/signieren/{signier_token}")
    assert seite.status_code == 200
    daten = seite.json()
    assert daten["vorgang"] == "ausgabe"
    assert daten["person_name"] == "Max Mustermann"

    absenden = await client.post(
        f"/api/v1/signieren/{signier_token}",
        files={"signatur": ("sig.png", _signatur_png_bytes(), "image/png")},
    )
    assert absenden.status_code == 204

    status_danach = await client.get(f"/api/v1/custody/{custody_id}/status", headers=auth_header(token))
    assert status_danach.json()["abgeschlossen"] is True


@pytest.mark.asyncio
async def test_token_nur_einmal_verwendbar(client, db_session, objekt_mit_position, mitarbeiter_user, admin_user):
    admin_token = await login(client, "admin1")
    token = await login(client, "mitarbeiter1")
    geraet_id = await _geraet_instanz(client, db_session, objekt_mit_position, admin_token)
    person = await _person(db_session)

    ausgabe = await client.post(
        f"/api/v1/geraete/{geraet_id}/ausgeben",
        json={"person_id": person.id, "mit_protokoll": True},
        headers=auth_header(token),
    )
    signier_token = ausgabe.json()["signier_url"].rsplit("/", 1)[-1]

    erste = await client.post(
        f"/api/v1/signieren/{signier_token}",
        files={"signatur": ("sig.png", _signatur_png_bytes(), "image/png")},
    )
    assert erste.status_code == 204

    zweite = await client.post(
        f"/api/v1/signieren/{signier_token}",
        files={"signatur": ("sig.png", _signatur_png_bytes(), "image/png")},
    )
    assert zweite.status_code == 404


@pytest.mark.asyncio
async def test_ungueltiger_token_generische_fehlermeldung(client):
    response = await client.get("/api/v1/signieren/nicht-existent")
    assert response.status_code == 404


@pytest.mark.asyncio
async def test_ungueltige_signatur_abgelehnt(client, db_session, objekt_mit_position, mitarbeiter_user, admin_user):
    admin_token = await login(client, "admin1")
    token = await login(client, "mitarbeiter1")
    geraet_id = await _geraet_instanz(client, db_session, objekt_mit_position, admin_token)
    person = await _person(db_session)

    ausgabe = await client.post(
        f"/api/v1/geraete/{geraet_id}/ausgeben",
        json={"person_id": person.id, "mit_protokoll": True},
        headers=auth_header(token),
    )
    signier_token = ausgabe.json()["signier_url"].rsplit("/", 1)[-1]

    response = await client.post(
        f"/api/v1/signieren/{signier_token}",
        files={"signatur": ("sig.txt", b"kein-bild", "text/plain")},
    )
    assert response.status_code == 422

    # Token bleibt nach fehlgeschlagener Signatur nutzbar (kein Verbrennen bei
    # bloß ungültiger Datei).
    zweiter_versuch = await client.post(
        f"/api/v1/signieren/{signier_token}",
        files={"signatur": ("sig.png", _signatur_png_bytes(), "image/png")},
    )
    assert zweiter_versuch.status_code == 204


async def _person_mit_benutzer(db_session):
    """PERS-008-Test: Person MIT verknüpftem Benutzerkonto, damit eine
    Qualifikationsprüfung überhaupt möglich ist."""
    from app.core.security import hash_password
    from app.models.auth import Benutzer
    from app.models.personal import Person

    person = Person(name="Praktikant Mit Konto")
    db_session.add(person)
    await db_session.flush()
    benutzer = Benutzer(
        name=person.name,
        login="praktikant_pers008",
        passwort_hash=hash_password("test-passwort-123"),
        aktiv=True,
        person_id=person.id,
    )
    db_session.add(benutzer)
    await db_session.flush()
    await db_session.commit()
    return person


@pytest.mark.asyncio
async def test_pers008_warnung_bei_fehlender_einweisung(
    client, db_session, objekt_mit_position, mitarbeiter_user, admin_user
):
    from app.models.personal import ObjekttypQualifikationsanforderung, Qualifikationskategorie, Qualifikationstyp

    objekt, material = objekt_mit_position
    admin_token = await login(client, "admin1")
    token = await login(client, "mitarbeiter1")
    geraet_id = await _geraet_instanz(client, db_session, objekt_mit_position, admin_token)

    qualifikationstyp = Qualifikationstyp(
        name="Einweisung Testgerät", kategorie=Qualifikationskategorie.berechtigung
    )
    db_session.add(qualifikationstyp)
    await db_session.flush()
    db_session.add(
        ObjekttypQualifikationsanforderung(objekttyp_id=objekt.objekttyp_id, qualifikationstyp_id=qualifikationstyp.id)
    )
    await db_session.commit()

    person = await _person_mit_benutzer(db_session)

    ausgabe = await client.post(
        f"/api/v1/geraete/{geraet_id}/ausgeben",
        json={"person_id": person.id, "mit_protokoll": False},
        headers=auth_header(token),
    )
    assert ausgabe.status_code == 201
    body = ausgabe.json()
    # Warnung, keine Blockade - Ausgabe ist trotzdem erfolgt.
    assert body["fehlende_qualifikationen"] == ["Einweisung Testgerät"]
    assert body["custody"]["ausgabe_abgeschlossen_am"] is not None


@pytest.mark.asyncio
async def test_pers008_keine_warnung_ohne_anforderung(
    client, db_session, objekt_mit_position, mitarbeiter_user, admin_user
):
    admin_token = await login(client, "admin1")
    token = await login(client, "mitarbeiter1")
    geraet_id = await _geraet_instanz(client, db_session, objekt_mit_position, admin_token)
    person = await _person(db_session)

    ausgabe = await client.post(
        f"/api/v1/geraete/{geraet_id}/ausgeben",
        json={"person_id": person.id, "mit_protokoll": False},
        headers=auth_header(token),
    )
    assert ausgabe.status_code == 201
    assert ausgabe.json()["fehlende_qualifikationen"] == []


@pytest.mark.asyncio
async def test_ruecknahme_formlos_und_mit_protokoll(
    client, db_session, objekt_mit_position, mitarbeiter_user, admin_user
):
    admin_token = await login(client, "admin1")
    token = await login(client, "mitarbeiter1")
    geraet_id = await _geraet_instanz(client, db_session, objekt_mit_position, admin_token)
    person = await _person(db_session)

    await client.post(
        f"/api/v1/geraete/{geraet_id}/ausgeben",
        json={"person_id": person.id, "mit_protokoll": False},
        headers=auth_header(token),
    )
    ruecknahme = await client.post(f"/api/v1/geraete/{geraet_id}/zurueckgeben", headers=auth_header(token))
    assert ruecknahme.status_code == 200
    assert ruecknahme.json()["custody"]["rueckgabe_abgeschlossen_am"] is not None
    assert ruecknahme.json()["signier_url"] is None

    ohne_offene_ausgabe = await client.post(f"/api/v1/geraete/{geraet_id}/zurueckgeben", headers=auth_header(token))
    assert ohne_offene_ausgabe.status_code == 409