Sprint 5: Mindermengen-Genehmigung + automatischer Ablauf, Historie für Kernkette
CI / backend-tests (push) Successful in 47s

- MindermengenGenehmigung-Modell/Service (Prompt 04): reine fachliche Bewertung,
  ändert NIEMALS Ist-Menge/Fehlmenge/Fehlbestand-Status (Leitplanke, U2)
- Automatischer Ablauf bei Start der nächsten Kontrolle desselben Objekts (Karte 08,
  U6) - kein Zeit-Trigger, der Kontrollstart selbst ist der Ablauf-Zeitpunkt
- Genehmigung endet gemeinsam mit dem Fehlbestand, wenn dieser durch Nachfüllung
  erledigt wird (U7), unabhängig davon ob vorher eine neue Kontrolle stattfand
- Historie-Service (app/services/historie.py) + Verdrahtung in die komplette
  Kernkette: kontrolle_gestartet/-abgeschlossen/-abgebrochen/-uebernommen,
  istmenge_erfasst, fehlbestand_entstanden/-erledigt, nachfuellung_erfasst,
  mindermenge_genehmigt/-abgelaufen/-beendet_durch_erledigung (U14)
  Hinweis: Stammdaten-/Vorlagen-/Benutzerverwaltung noch nicht retrofittet -
  Sprint 5 deckt bewusst die im Testkonzept referenzierte Kernkette ab, keine
  flächendeckende Audit-Abdeckung aller CRUD-Endpunkte.
- Tests: U2, U6, U7, vollständige Historie-Kette nach Prompt-13.3-Beispiel (U14)

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01L85hmKbvX7Cqkq47KnQhFt
This commit is contained in:
2026-09-03 23:53:14 +02:00
co-authored by Claude Sonnet 5
parent 18fba28eae
commit 1e9012cba3
13 changed files with 716 additions and 21 deletions
+187
View File
@@ -0,0 +1,187 @@
import pytest
from sqlalchemy import select
from app.models.fehlbestand import Fehlbestand
from app.models.mindermenge import MindermengenGenehmigung
from app.models.objektposition import Objektposition
from tests.conftest import auth_header, login
async def _fehlbestand_ueber_kontrolle(client, token, objekt, material, istmenge: str, abschliessen=False):
start = await client.post(
f"/api/v1/objekte/{objekt.id}/kontrollen", json={"uebernehmen": False}, headers=auth_header(token)
)
kontrolle_id = start.json()["id"]
put_response = await client.put(
f"/api/v1/kontrollen/{kontrolle_id}/positionen/{material.id}",
json={"istmenge": istmenge},
headers=auth_header(token),
)
fehlbestand_id = put_response.json()["fehlbestand_id"]
if abschliessen:
await client.post(f"/api/v1/kontrollen/{kontrolle_id}/abschliessen", headers=auth_header(token))
return kontrolle_id, fehlbestand_id
@pytest.mark.asyncio
async def test_genehmigung_aendert_nichts_am_fehlbestand(
client, db_session, objekt_mit_position, mitarbeiter_user, materialverantwortlicher_user
):
"""U2 - Leitplanke: Genehmigung ändert NIEMALS Ist-Menge/Fehlmenge/Status."""
objekt, material = objekt_mit_position
mitarbeiter_token = await login(client, "mitarbeiter1")
_, fehlbestand_id = await _fehlbestand_ueber_kontrolle(
client, mitarbeiter_token, objekt, material, "8", abschliessen=True
)
vor = await db_session.get(Fehlbestand, fehlbestand_id)
vor_status, vor_istmenge, vor_fehlmenge = vor.status, vor.istmenge, vor.fehlmenge
verantwortlicher_token = await login(client, "materialverantwortlicher1")
response = await client.post(
f"/api/v1/fehlbestaende/{fehlbestand_id}/mindermenge",
json={"begruendung": "Lieferengpass, Ersatz bestellt"},
headers=auth_header(verantwortlicher_token),
)
assert response.status_code == 201
assert response.json()["status"] == "aktiv"
await db_session.refresh(vor)
assert vor.status == vor_status
assert vor.istmenge == vor_istmenge
assert vor.fehlmenge == vor_fehlmenge
objektposition = (
await db_session.execute(
select(Objektposition).where(
Objektposition.objekt_id == objekt.id, Objektposition.material_id == material.id
)
)
).scalar_one()
assert float(objektposition.istmenge) == 0 # unverändert (Kontrolle/Genehmigung ändern nie Ist-Menge)
@pytest.mark.asyncio
async def test_mitarbeiter_darf_nicht_genehmigen(
client, objekt_mit_position, mitarbeiter_user
):
objekt, material = objekt_mit_position
token = await login(client, "mitarbeiter1")
_, fehlbestand_id = await _fehlbestand_ueber_kontrolle(client, token, objekt, material, "8", abschliessen=True)
response = await client.post(
f"/api/v1/fehlbestaende/{fehlbestand_id}/mindermenge",
json={"begruendung": "Versuch"},
headers=auth_header(token),
)
assert response.status_code == 403
@pytest.mark.asyncio
async def test_genehmigung_laeuft_bei_naechster_kontrolle_ab(
client, db_session, objekt_mit_position, mitarbeiter_user, materialverantwortlicher_user
):
"""U6 / Karte 08: Genehmigung gilt automatisch nur bis zur nächsten Kontrolle."""
objekt, material = objekt_mit_position
mitarbeiter_token = await login(client, "mitarbeiter1")
_, fehlbestand_id = await _fehlbestand_ueber_kontrolle(
client, mitarbeiter_token, objekt, material, "8", abschliessen=True
)
verantwortlicher_token = await login(client, "materialverantwortlicher1")
genehmigung_response = await client.post(
f"/api/v1/fehlbestaende/{fehlbestand_id}/mindermenge",
json={"begruendung": "Lieferengpass"},
headers=auth_header(verantwortlicher_token),
)
genehmigung_id = genehmigung_response.json()["id"]
naechste_kontrolle = await client.post(
f"/api/v1/objekte/{objekt.id}/kontrollen", json={"uebernehmen": False}, headers=auth_header(mitarbeiter_token)
)
assert naechste_kontrolle.status_code == 201
neue_kontrolle_id = naechste_kontrolle.json()["id"]
genehmigung = await db_session.get(MindermengenGenehmigung, genehmigung_id)
assert genehmigung.status.value == "abgelaufen"
assert str(genehmigung.beendende_kontrolle_id) == neue_kontrolle_id
@pytest.mark.asyncio
async def test_genehmigung_endet_gemeinsam_mit_erledigung(
client, db_session, objekt_mit_position, mitarbeiter_user, materialverantwortlicher_user
):
"""U7: wird die Abweichung behoben, erledigt sich Fehlbestand UND Genehmigung gemeinsam."""
objekt, material = objekt_mit_position
mitarbeiter_token = await login(client, "mitarbeiter1")
_, fehlbestand_id = await _fehlbestand_ueber_kontrolle(
client, mitarbeiter_token, objekt, material, "8", abschliessen=True
)
verantwortlicher_token = await login(client, "materialverantwortlicher1")
genehmigung_response = await client.post(
f"/api/v1/fehlbestaende/{fehlbestand_id}/mindermenge",
json={"begruendung": "Lieferengpass"},
headers=auth_header(verantwortlicher_token),
)
genehmigung_id = genehmigung_response.json()["id"]
nachfuellung = await client.post(
f"/api/v1/fehlbestaende/{fehlbestand_id}/nachfuellungen",
json={"menge": "2"},
headers=auth_header(mitarbeiter_token),
)
assert nachfuellung.json()["fehlbestand"]["status"] == "erledigt"
genehmigung = await db_session.get(MindermengenGenehmigung, genehmigung_id)
assert genehmigung.status.value == "beendet_durch_erledigung"
@pytest.mark.asyncio
async def test_doppelte_genehmigung_ist_409(
client, objekt_mit_position, mitarbeiter_user, materialverantwortlicher_user
):
objekt, material = objekt_mit_position
mitarbeiter_token = await login(client, "mitarbeiter1")
_, fehlbestand_id = await _fehlbestand_ueber_kontrolle(
client, mitarbeiter_token, objekt, material, "8", abschliessen=True
)
verantwortlicher_token = await login(client, "materialverantwortlicher1")
erste = await client.post(
f"/api/v1/fehlbestaende/{fehlbestand_id}/mindermenge",
json={"begruendung": "Lieferengpass"},
headers=auth_header(verantwortlicher_token),
)
assert erste.status_code == 201
zweite = await client.post(
f"/api/v1/fehlbestaende/{fehlbestand_id}/mindermenge",
json={"begruendung": "Nochmal"},
headers=auth_header(verantwortlicher_token),
)
assert zweite.status_code == 409
@pytest.mark.asyncio
async def test_genehmigung_auf_erledigten_fehlbestand_ist_409(
client, objekt_mit_position, mitarbeiter_user, materialverantwortlicher_user
):
objekt, material = objekt_mit_position
mitarbeiter_token = await login(client, "mitarbeiter1")
_, fehlbestand_id = await _fehlbestand_ueber_kontrolle(
client, mitarbeiter_token, objekt, material, "8", abschliessen=True
)
await client.post(
f"/api/v1/fehlbestaende/{fehlbestand_id}/nachfuellungen",
json={"menge": "2"},
headers=auth_header(mitarbeiter_token),
)
verantwortlicher_token = await login(client, "materialverantwortlicher1")
response = await client.post(
f"/api/v1/fehlbestaende/{fehlbestand_id}/mindermenge",
json={"begruendung": "Zu spät"},
headers=auth_header(verantwortlicher_token),
)
assert response.status_code == 409