patrickandClaude Sonnet 5
e949b5e9ac
historie-Tabelle (append-only Audit-Trail über Kontrolle/Fehlbestand/ Nachfüllung/Mindermenge) war bisher nur intern befüllt, ohne API/UI. Neu: GET /api/v1/historie (gefiltert nach Typ/ID/Benutzer/Zeitraum, paginiert, Benutzername aufgelöst), Admin-Portal-Tab "Änderungslog" mit Typ-Filter. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01CVgbozhYmuEhiEJHffRXCV
100 lines
3.8 KiB
Python
100 lines
3.8 KiB
Python
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
|