patrickandClaude Sonnet 5
567ce9d3c3
entferne_dokument verlangt _materialverantwortliche - Mitarbeiter-Login lief mit 403 ins Leere, v2 blieb aktuell, Test verglich falsche ID gegen v1. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01CVgbozhYmuEhiEJHffRXCV
371 lines
15 KiB
Python
371 lines
15 KiB
Python
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
|