Karte 12: Eskalationslogik für lange offene Fehlbestände

Zwei Stufen (Erinnerung an Materialverantwortliche, Eskalation an
Leitungsverantwortliche/Administration) mit Tracking-Zeitstempel je Stufe
gegen Mehrfachversand. Zeitschwellen in eskalation_konfiguration (DB,
Singleton), admin-editierbar über GET/PUT /eskalation/konfiguration - bewusst
nicht fix im Code. Prüflauf per POST /eskalation/pruefen angestoßen,
Scheduling (Cron/systemd-Timer) bleibt Betriebsaufgabe.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01CVgbozhYmuEhiEJHffRXCV

backend/tests/test_eskalation.py

@@ -0,0 +1,73 @@
+from datetime import datetime, timedelta, timezone
+
+import pytest
+
+from app.models.fehlbestand import Fehlbestand
+from app.services.eskalation import pruefe_offene_fehlbestaende
+from tests.conftest import auth_header, login
+
+
+async def _fehlbestand_erzeugen(client, token, objekt, material, istmenge: str):
+    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),
+    )
+    return put_response.json()["fehlbestand_id"]
+
+
+@pytest.mark.asyncio
+async def test_konfiguration_lesen_und_aendern(client, admin_user):
+    token = await login(client, "admin1")
+    lesen = await client.get("/api/v1/eskalation/konfiguration", headers=auth_header(token))
+    assert lesen.status_code == 200
+    assert lesen.json() == {"erinnerung_tage": 3, "leitung_tage": 7}
+
+    aendern = await client.put(
+        "/api/v1/eskalation/konfiguration",
+        json={"erinnerung_tage": 1, "leitung_tage": 2},
+        headers=auth_header(token),
+    )
+    assert aendern.status_code == 200
+    assert aendern.json() == {"erinnerung_tage": 1, "leitung_tage": 2}
+
+
+@pytest.mark.asyncio
+async def test_mitarbeiter_darf_konfiguration_nicht_lesen(client, mitarbeiter_user):
+    token = await login(client, "mitarbeiter1")
+    response = await client.get("/api/v1/eskalation/konfiguration", headers=auth_header(token))
+    assert response.status_code == 403
+
+
+@pytest.mark.asyncio
+async def test_pruefung_setzt_stufen_nur_einmal(
+    client, db_session, objekt_mit_position, mitarbeiter_user, admin_user
+):
+    objekt, material = objekt_mit_position
+    token = await login(client, "mitarbeiter1")
+    fehlbestand_id = await _fehlbestand_erzeugen(client, token, objekt, material, "3")
+
+    fehlbestand = await db_session.get(Fehlbestand, fehlbestand_id)
+    fehlbestand.entstanden_am = datetime.now(timezone.utc) - timedelta(days=10)
+    await db_session.flush()
+
+    admin_token = await login(client, "admin1")
+    await client.put(
+        "/api/v1/eskalation/konfiguration",
+        json={"erinnerung_tage": 3, "leitung_tage": 7},
+        headers=auth_header(admin_token),
+    )
+
+    ergebnis = await pruefe_offene_fehlbestaende(db_session)
+    assert ergebnis == {"erinnerungen": 1, "eskalationen": 1}
+
+    await db_session.refresh(fehlbestand)
+    assert fehlbestand.erinnerung_gesendet_am is not None
+    assert fehlbestand.eskalation_gesendet_am is not None
+
+    wiederholung = await pruefe_offene_fehlbestaende(db_session)
+    assert wiederholung == {"erinnerungen": 0, "eskalationen": 0}