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, materialverantwortlicher_user
):
    """DOC-002: feste Kategorie statt Freitext - Dokumente pro Typ auffindbar.
    Verantwortlicher-Login, nicht Mitarbeiter, da "rechnung" einer der beiden
    Testtypen ist und Mitarbeiter die per DOC-005 nicht sehen (eigener Test)."""
    objekt, _material = objekt_mit_position
    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(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


@pytest.mark.asyncio
async def test_dokument_ersetzen_legt_neue_version_an_alte_bleibt_erhalten(
    client, objekt_mit_position, mitarbeiter_user
):
    """DOC-003: alte Version bleibt in der Versionshistorie, Liste zeigt nur
    die neue als aktuell."""
    objekt, _material = objekt_mit_position
    token = await login(client, "mitarbeiter1")

    v1 = await client.post(
        "/api/v1/dokumente",
        data={
            "entitaet_typ": "objekt",
            "entitaet_id": str(objekt.id),
            "dokumenttyp": "pruefprotokoll",
            "beschreibung": "Erste Fassung",
        },
        files={"datei": ("protokoll_v1.pdf", b"%PDF-1.4 v1", "application/pdf")},
        headers=auth_header(token),
    )
    v1_id = v1.json()["id"]

    v2 = await client.post(
        f"/api/v1/dokumente/{v1_id}/ersetzen",
        data={"beschreibung": "Zweite Fassung"},
        files={"datei": ("protokoll_v2.pdf", b"%PDF-1.4 v2", "application/pdf")},
        headers=auth_header(token),
    )
    assert v2.status_code == 201
    v2_body = v2.json()
    assert v2_body["dateiname"] == "protokoll_v2.pdf"
    assert v2_body["dokumenttyp"] == "pruefprotokoll"
    assert v2_body["vorgaenger_id"] == v1_id

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

    historie = await client.get(f"/api/v1/dokumente/{v2_body['id']}/versionen", headers=auth_header(token))
    dateinamen = [d["dateiname"] for d in historie.json()]
    assert dateinamen == ["protokoll_v2.pdf", "protokoll_v1.pdf"]

    alte_datei_noch_da = await client.get(f"/api/v1/dokumente/{v1_id}/download", headers=auth_header(token))
    assert alte_datei_noch_da.status_code == 200
    assert alte_datei_noch_da.content == b"%PDF-1.4 v1"


@pytest.mark.asyncio
async def test_loeschen_der_aktuellen_version_gibt_vorgaenger_frei(
    client, objekt_mit_position, mitarbeiter_user, materialverantwortlicher_user
):
    """Löscht man die aktuelle Version, wird die vorherige wieder als aktuell
    gelistet (nichts zeigt mehr per vorgaenger_id auf sie)."""
    objekt, _material = objekt_mit_position
    token = await login(client, "mitarbeiter1")

    v1 = await client.post(
        "/api/v1/dokumente",
        data={"entitaet_typ": "objekt", "entitaet_id": str(objekt.id), "dokumenttyp": "sonstiges"},
        files={"datei": ("v1.pdf", b"%PDF-1.4 v1", "application/pdf")},
        headers=auth_header(token),
    )
    v1_id = v1.json()["id"]
    v2 = await client.post(
        f"/api/v1/dokumente/{v1_id}/ersetzen",
        data={},
        files={"datei": ("v2.pdf", b"%PDF-1.4 v2", "application/pdf")},
        headers=auth_header(token),
    )
    v2_id = v2.json()["id"]

    # Löschen erfordert Materialverantwortliche+ (_materialverantwortliche) -
    # Mitarbeiter dürfen selbst hochgeladene Dokumente nicht löschen.
    verantwortlicher_token = await login(client, "materialverantwortlicher1")
    geloescht = await client.delete(
        f"/api/v1/dokumente/{v2_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(token)
    )
    body = liste.json()
    assert len(body) == 1
    assert body[0]["id"] == v1_id


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

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


@pytest.mark.asyncio
async def test_original_kennzeichnung_default_false_und_explizit_setzbar(
    client, objekt_mit_position, mitarbeiter_user
):
    """DOC-004: ohne Angabe gilt eine Datei als Kopie (False), kann aber
    explizit als Original markiert werden - auch beim Ersetzen unabhängig
    vom Vorgänger neu wählbar."""
    objekt, _material = objekt_mit_position
    token = await login(client, "mitarbeiter1")

    ohne_angabe = await client.post(
        "/api/v1/dokumente",
        data={"entitaet_typ": "objekt", "entitaet_id": str(objekt.id), "dokumenttyp": "zulassungsdokument"},
        files={"datei": ("kopie.pdf", b"%PDF-1.4 kopie", "application/pdf")},
        headers=auth_header(token),
    )
    assert ohne_angabe.json()["ist_original"] is False

    original = await client.post(
        "/api/v1/dokumente",
        data={
            "entitaet_typ": "objekt",
            "entitaet_id": str(objekt.id),
            "dokumenttyp": "zulassungsdokument",
            "ist_original": "true",
        },
        files={"datei": ("original.pdf", b"%PDF-1.4 original", "application/pdf")},
        headers=auth_header(token),
    )
    original_body = original.json()
    assert original_body["ist_original"] is True

    ersetzt = await client.post(
        f"/api/v1/dokumente/{original_body['id']}/ersetzen",
        data={"ist_original": "false"},
        files={"datei": ("nachtraeglich_kopie.pdf", b"%PDF-1.4 v2", "application/pdf")},
        headers=auth_header(token),
    )
    assert ersetzt.json()["ist_original"] is False