patrickandClaude Sonnet 5
fb676a4a0b
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
74 lines
2.7 KiB
Python
74 lines
2.7 KiB
Python
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}
|