MABEA/backend/tests/test_dokument.py

import pytest

from tests.conftest import auth_header, login


@pytest.mark.asyncio
async def test_dokument_hochladen_und_liste(client, objekt_mit_position, mitarbeiter_user):
    objekt, _material = objekt_mit_position
    token = await login(client, "mitarbeiter1")

    hochgeladen = await client.post(
        "/api/v1/dokumente",
        data={"entitaet_typ": "objekt", "entitaet_id": str(objekt.id), "dokumenttyp": "zulassungsdokument", "beschreibung": "Zulassung"},
        files={"datei": ("zulassung.pdf", b"%PDF-1.4 fake", "application/pdf")},
        headers=auth_header(token),
    )
    assert hochgeladen.status_code == 201
    body = hochgeladen.json()
    assert body["dateiname"] == "zulassung.pdf"
    assert body["groesse_bytes"] > 0

    liste = await client.get(
        f"/api/v1/dokumente?entitaet_typ=objekt&entitaet_id={objekt.id}", headers=auth_header(token)
    )
    assert len(liste.json()) == 1


@pytest.mark.asyncio
async def test_download_liefert_inhalt(client, objekt_mit_position, mitarbeiter_user):
    objekt, _material = objekt_mit_position
    token = await login(client, "mitarbeiter1")
    hochgeladen = await client.post(
        "/api/v1/dokumente",
        data={"entitaet_typ": "objekt", "entitaet_id": str(objekt.id), "dokumenttyp": "sonstiges"},
        files={"datei": ("foto.png", b"\x89PNG fake", "image/png")},
        headers=auth_header(token),
    )
    dokument_id = hochgeladen.json()["id"]

    download = await client.get(f"/api/v1/dokumente/{dokument_id}/download", headers=auth_header(token))
    assert download.status_code == 200
    assert download.content == b"\x89PNG fake"


@pytest.mark.asyncio
async def test_unerlaubter_dateityp_wird_abgelehnt(client, objekt_mit_position, mitarbeiter_user):
    objekt, _material = objekt_mit_position
    token = await login(client, "mitarbeiter1")

    response = await client.post(
        "/api/v1/dokumente",
        data={"entitaet_typ": "objekt", "entitaet_id": str(objekt.id), "dokumenttyp": "sonstiges"},
        files={"datei": ("boese.html", b"<script>alert(1)</script>", "text/html")},
        headers=auth_header(token),
    )
    assert response.status_code == 415


@pytest.mark.asyncio
async def test_materialverantwortlicher_kann_dokument_loeschen(
    client, objekt_mit_position, mitarbeiter_user, materialverantwortlicher_user
):
    objekt, _material = objekt_mit_position
    mitarbeiter_token = await login(client, "mitarbeiter1")
    hochgeladen = await client.post(
        "/api/v1/dokumente",
        data={"entitaet_typ": "objekt", "entitaet_id": str(objekt.id), "dokumenttyp": "sonstiges"},
        files={"datei": ("bericht.pdf", b"%PDF-1.4 x", "application/pdf")},
        headers=auth_header(mitarbeiter_token),
    )
    dokument_id = hochgeladen.json()["id"]

    verantwortlicher_token = await login(client, "materialverantwortlicher1")
    geloescht = await client.delete(f"/api/v1/dokumente/{dokument_id}", headers=auth_header(verantwortlicher_token))
    assert geloescht.status_code == 204

    liste = await client.get(
        f"/api/v1/dokumente?entitaet_typ=objekt&entitaet_id={objekt.id}", headers=auth_header(verantwortlicher_token)
    )
    assert liste.json() == []


@pytest.mark.asyncio
async def test_mitarbeiter_darf_dokument_nicht_loeschen(client, objekt_mit_position, mitarbeiter_user):
    objekt, _material = objekt_mit_position
    token = await login(client, "mitarbeiter1")
    hochgeladen = await client.post(
        "/api/v1/dokumente",
        data={"entitaet_typ": "objekt", "entitaet_id": str(objekt.id), "dokumenttyp": "sonstiges"},
        files={"datei": ("x.pdf", b"%PDF-1.4 x", "application/pdf")},
        headers=auth_header(token),
    )
    dokument_id = hochgeladen.json()["id"]

    response = await client.delete(f"/api/v1/dokumente/{dokument_id}", headers=auth_header(token))
    assert response.status_code == 403


@pytest.mark.asyncio
async def test_identische_datei_am_gleichen_objekt_wird_abgelehnt(client, objekt_mit_position, mitarbeiter_user):
    """DOC-001: gleiche Bytes, gleiche Entität -> 409 statt zweitem Dokument."""
    objekt, _material = objekt_mit_position
    token = await login(client, "mitarbeiter1")
    inhalt = b"%PDF-1.4 identischer inhalt"

    erster = await client.post(
        "/api/v1/dokumente",
        data={"entitaet_typ": "objekt", "entitaet_id": str(objekt.id), "dokumenttyp": "sonstiges"},
        files={"datei": ("original.pdf", inhalt, "application/pdf")},
        headers=auth_header(token),
    )
    assert erster.status_code == 201

    zweiter = await client.post(
        "/api/v1/dokumente",
        data={"entitaet_typ": "objekt", "entitaet_id": str(objekt.id), "dokumenttyp": "sonstiges"},
        files={"datei": ("kopie.pdf", inhalt, "application/pdf")},
        headers=auth_header(token),
    )
    assert zweiter.status_code == 409
    assert "original.pdf" in zweiter.json()["detail"]

    liste = await client.get(
        f"/api/v1/dokumente?entitaet_typ=objekt&entitaet_id={objekt.id}", headers=auth_header(token)
    )
    assert len(liste.json()) == 1


@pytest.mark.asyncio
async def test_identische_datei_an_anderer_entitaet_ist_kein_duplikat(client, objekt_mit_position, mitarbeiter_user):
    """Gleiche Datei an ZWEI verschiedenen Objekten ist kein Duplikat - Hash-
    Vergleich ist bewusst pro Entität, nicht global."""
    objekt, _material = objekt_mit_position
    token = await login(client, "mitarbeiter1")
    inhalt = b"%PDF-1.4 gleiche datei anderes objekt"

    await client.post(
        "/api/v1/dokumente",
        data={"entitaet_typ": "objekt", "entitaet_id": str(objekt.id), "dokumenttyp": "sonstiges"},
        files={"datei": ("anleitung.pdf", inhalt, "application/pdf")},
        headers=auth_header(token),
    )
    andere_entitaet = await client.post(
        "/api/v1/dokumente",
        data={"entitaet_typ": "mangel", "entitaet_id": "irgendein-mangel-id", "dokumenttyp": "sonstiges"},
        files={"datei": ("anleitung.pdf", inhalt, "application/pdf")},
        headers=auth_header(token),
    )
    assert andere_entitaet.status_code == 201


@pytest.mark.asyncio
async def test_liste_kann_nach_dokumenttyp_gefiltert_werden(client, objekt_mit_position, mitarbeiter_user):
    """DOC-002: feste Kategorie statt Freitext - Dokumente pro Typ auffindbar."""
    objekt, _material = objekt_mit_position
    token = await login(client, "mitarbeiter1")

    await client.post(
        "/api/v1/dokumente",
        data={"entitaet_typ": "objekt", "entitaet_id": str(objekt.id), "dokumenttyp": "rechnung"},
        files={"datei": ("rechnung.pdf", b"%PDF-1.4 r", "application/pdf")},
        headers=auth_header(token),
    )
    await client.post(
        "/api/v1/dokumente",
        data={"entitaet_typ": "objekt", "entitaet_id": str(objekt.id), "dokumenttyp": "pruefprotokoll"},
        files={"datei": ("pruefung.pdf", b"%PDF-1.4 p", "application/pdf")},
        headers=auth_header(token),
    )

    alle = await client.get(
        f"/api/v1/dokumente?entitaet_typ=objekt&entitaet_id={objekt.id}", headers=auth_header(token)
    )
    assert len(alle.json()) == 2

    nur_rechnungen = await client.get(
        f"/api/v1/dokumente?entitaet_typ=objekt&entitaet_id={objekt.id}&dokumenttyp=rechnung",
        headers=auth_header(token),
    )
    body = nur_rechnungen.json()
    assert len(body) == 1
    assert body[0]["dateiname"] == "rechnung.pdf"
    assert body[0]["dokumenttyp"] == "rechnung"


@pytest.mark.asyncio
async def test_unbekannter_dokumenttyp_wird_abgelehnt(client, objekt_mit_position, mitarbeiter_user):
    objekt, _material = objekt_mit_position
    token = await login(client, "mitarbeiter1")

    response = await client.post(
        "/api/v1/dokumente",
        data={"entitaet_typ": "objekt", "entitaet_id": str(objekt.id), "dokumenttyp": "erfunden"},
        files={"datei": ("x.pdf", b"%PDF-1.4 x", "application/pdf")},
        headers=auth_header(token),
    )
    assert response.status_code == 422


@pytest.mark.asyncio
async def test_mitarbeiter_sieht_rechnungen_nicht_in_der_liste(
    client, objekt_mit_position, mitarbeiter_user, materialverantwortlicher_user
):
    """DOC-005: Rechnungen sind für Mitarbeiter unsichtbar, für Materialverantwortliche sichtbar."""
    objekt, _material = objekt_mit_position
    verantwortlicher_token = await login(client, "materialverantwortlicher1")
    await client.post(
        "/api/v1/dokumente",
        data={"entitaet_typ": "objekt", "entitaet_id": str(objekt.id), "dokumenttyp": "rechnung"},
        files={"datei": ("rechnung.pdf", b"%PDF-1.4 r", "application/pdf")},
        headers=auth_header(verantwortlicher_token),
    )
    await client.post(
        "/api/v1/dokumente",
        data={"entitaet_typ": "objekt", "entitaet_id": str(objekt.id), "dokumenttyp": "pruefprotokoll"},
        files={"datei": ("pruefung.pdf", b"%PDF-1.4 p", "application/pdf")},
        headers=auth_header(verantwortlicher_token),
    )

    mitarbeiter_token = await login(client, "mitarbeiter1")
    als_mitarbeiter = await client.get(
        f"/api/v1/dokumente?entitaet_typ=objekt&entitaet_id={objekt.id}", headers=auth_header(mitarbeiter_token)
    )
    dateinamen = [d["dateiname"] for d in als_mitarbeiter.json()]
    assert "rechnung.pdf" not in dateinamen
    assert "pruefung.pdf" in dateinamen

    als_verantwortlicher = await client.get(
        f"/api/v1/dokumente?entitaet_typ=objekt&entitaet_id={objekt.id}", headers=auth_header(verantwortlicher_token)
    )
    assert len(als_verantwortlicher.json()) == 2


@pytest.mark.asyncio
async def test_mitarbeiter_kann_rechnung_nicht_direkt_herunterladen(
    client, objekt_mit_position, mitarbeiter_user, materialverantwortlicher_user
):
    objekt, _material = objekt_mit_position
    verantwortlicher_token = await login(client, "materialverantwortlicher1")
    hochgeladen = await client.post(
        "/api/v1/dokumente",
        data={"entitaet_typ": "objekt", "entitaet_id": str(objekt.id), "dokumenttyp": "rechnung"},
        files={"datei": ("rechnung.pdf", b"%PDF-1.4 r", "application/pdf")},
        headers=auth_header(verantwortlicher_token),
    )
    dokument_id = hochgeladen.json()["id"]

    mitarbeiter_token = await login(client, "mitarbeiter1")
    verweigert = await client.get(
        f"/api/v1/dokumente/{dokument_id}/download", headers=auth_header(mitarbeiter_token)
    )
    assert verweigert.status_code == 403

    erlaubt = await client.get(
        f"/api/v1/dokumente/{dokument_id}/download", headers=auth_header(verantwortlicher_token)
    )
    assert erlaubt.status_code == 200