patrickandClaude Sonnet 5
9721873088
_synchronisiere_fehlbestand suchte nur nach Fehlbeständen der eigenen
Kontroll-Sitzung. Ein bereits offener Fehlbestand aus einer früheren
Kontrolle für dasselbe Objekt+Material wurde nicht wiedergefunden -
es entstand ein zweiter, paralleler Fehlbestand mit eigener (neuerer)
Sollmenge, der alte blieb mit veralteter Sollmenge offen liegen.
Lookup jetzt über objekt_id+material_id+status=offen statt
kontrolle_id; kontrolle_id am gefundenen Datensatz bleibt unverändert,
damit ein Abbruch der aktuellen Kontrolle ihn nicht fälschlich
mitlöscht. Löst sich die Abweichung über eine andere Kontrolle als
die erzeugende auf, wird der Fehlbestand jetzt sauber als erledigt
markiert (inkl. Mindermenge-Abschluss) statt stillschweigend gelöscht.
Nebenbei: Kommentare mit Kachel-/Datums-Referenzen ("Bug-Fund
2026-09-06", "UI-00X (Epic 20)") aus dem Code dieser Session entfernt.
Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01CVgbozhYmuEhiEJHffRXCV
304 lines
12 KiB
Python
304 lines
12 KiB
Python
from datetime import date, timedelta
|
|
|
|
import pytest
|
|
from sqlalchemy import select
|
|
|
|
from app.models.geraet_instanz import GeraetInstanz
|
|
from app.models.objektposition import Objektposition
|
|
from tests.conftest import auth_header, login
|
|
|
|
|
|
async def _fehlbestand(client, token, objekt, material, istmenge: str, abschliessen=True):
|
|
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),
|
|
)
|
|
if abschliessen:
|
|
await client.post(f"/api/v1/kontrollen/{kontrolle_id}/abschliessen", headers=auth_header(token))
|
|
return put_response.json()["fehlbestand_id"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_offene_stati_werden_gemeinsam_gezaehlt(
|
|
client, objekt_mit_position, mitarbeiter_user, materialverantwortlicher_user
|
|
):
|
|
"""Prompt 12.2: offen + in_bearbeitung + nachgefuellt_teilweise = "offen"."""
|
|
objekt, material = objekt_mit_position
|
|
mitarbeiter_token = await login(client, "mitarbeiter1")
|
|
fehlbestand_id = await _fehlbestand(client, mitarbeiter_token, objekt, material, "6")
|
|
|
|
# Teilnachfüllung -> Status nachgefuellt_teilweise, zählt trotzdem als offen
|
|
await client.post(
|
|
f"/api/v1/fehlbestaende/{fehlbestand_id}/nachfuellungen",
|
|
json={"menge": "1"},
|
|
headers=auth_header(mitarbeiter_token),
|
|
)
|
|
|
|
verantwortlicher_token = await login(client, "materialverantwortlicher1")
|
|
response = await client.get("/api/v1/dashboard/kennzahlen", headers=auth_header(verantwortlicher_token))
|
|
assert response.status_code == 200
|
|
assert response.json()["anzahl_offener_fehlbestaende"] == 1
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_genehmigte_mindermenge_zaehlt_als_offen_und_separat(
|
|
client, objekt_mit_position, mitarbeiter_user, materialverantwortlicher_user
|
|
):
|
|
objekt, material = objekt_mit_position
|
|
mitarbeiter_token = await login(client, "mitarbeiter1")
|
|
fehlbestand_id = await _fehlbestand(client, mitarbeiter_token, objekt, material, "6")
|
|
|
|
verantwortlicher_token = await login(client, "materialverantwortlicher1")
|
|
await client.post(
|
|
f"/api/v1/fehlbestaende/{fehlbestand_id}/mindermenge",
|
|
json={"begruendung": "Lieferengpass"},
|
|
headers=auth_header(verantwortlicher_token),
|
|
)
|
|
|
|
response = await client.get("/api/v1/dashboard/kennzahlen", headers=auth_header(verantwortlicher_token))
|
|
body = response.json()
|
|
assert body["anzahl_genehmigter_mindermengen"] == 1
|
|
assert body["anzahl_offener_fehlbestaende"] == 1 # Genehmigung ist keine Erledigung
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_kuerzlich_erledigt_wird_gezaehlt(
|
|
client, objekt_mit_position, mitarbeiter_user, materialverantwortlicher_user
|
|
):
|
|
objekt, material = objekt_mit_position
|
|
mitarbeiter_token = await login(client, "mitarbeiter1")
|
|
fehlbestand_id = await _fehlbestand(client, mitarbeiter_token, objekt, material, "6")
|
|
await client.post(
|
|
f"/api/v1/fehlbestaende/{fehlbestand_id}/nachfuellungen",
|
|
json={"menge": "4"},
|
|
headers=auth_header(mitarbeiter_token),
|
|
)
|
|
|
|
verantwortlicher_token = await login(client, "materialverantwortlicher1")
|
|
response = await client.get("/api/v1/dashboard/kennzahlen", headers=auth_header(verantwortlicher_token))
|
|
body = response.json()
|
|
assert body["anzahl_kuerzlich_erledigter_fehlbestaende"] == 1
|
|
assert body["anzahl_offener_fehlbestaende"] == 0
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_mitarbeiter_darf_dashboard_nicht_sehen(client, mitarbeiter_user):
|
|
token = await login(client, "mitarbeiter1")
|
|
response = await client.get("/api/v1/dashboard/kennzahlen", headers=auth_header(token))
|
|
assert response.status_code == 403
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_ablaufdaten_liste_sortiert_und_gefiltert(
|
|
client, db_session, objekt_mit_position, materialverantwortlicher_user
|
|
):
|
|
"""Prompt 14: nur bald_ablaufend/abgelaufen, sortiert nach verbleibenden Tagen aufsteigend."""
|
|
from datetime import date, timedelta
|
|
|
|
objekt, material = objekt_mit_position
|
|
verantwortlicher_token = await login(client, "materialverantwortlicher1")
|
|
|
|
positionen = (
|
|
await client.get(f"/api/v1/objekte/{objekt.id}/positionen", headers=auth_header(verantwortlicher_token))
|
|
).json()
|
|
position_id = positionen[0]["id"]
|
|
|
|
# weit in der Zukunft -> "gueltig", darf NICHT in der Liste auftauchen
|
|
await client.patch(
|
|
f"/api/v1/objekte/{objekt.id}/positionen/{position_id}",
|
|
json={"ablaufdatum": (date.today() + timedelta(days=365)).isoformat()},
|
|
headers=auth_header(verantwortlicher_token),
|
|
)
|
|
response = await client.get("/api/v1/dashboard/ablaufdaten", headers=auth_header(verantwortlicher_token))
|
|
assert response.json() == []
|
|
|
|
# bald ablaufend (Standard-Warnzeitraum 30 Tage, kein warnzeitraum_tage auf Material gesetzt)
|
|
await client.patch(
|
|
f"/api/v1/objekte/{objekt.id}/positionen/{position_id}",
|
|
json={"ablaufdatum": (date.today() + timedelta(days=5)).isoformat()},
|
|
headers=auth_header(verantwortlicher_token),
|
|
)
|
|
response = await client.get("/api/v1/dashboard/ablaufdaten", headers=auth_header(verantwortlicher_token))
|
|
body = response.json()
|
|
assert len(body) == 1
|
|
assert body[0]["status"] == "bald_ablaufend"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_abgelaufenes_material_wird_als_abgelaufen_markiert(
|
|
client, objekt_mit_position, materialverantwortlicher_user
|
|
):
|
|
from datetime import date, timedelta
|
|
|
|
objekt, material = objekt_mit_position
|
|
token = await login(client, "materialverantwortlicher1")
|
|
positionen = (await client.get(f"/api/v1/objekte/{objekt.id}/positionen", headers=auth_header(token))).json()
|
|
position_id = positionen[0]["id"]
|
|
|
|
await client.patch(
|
|
f"/api/v1/objekte/{objekt.id}/positionen/{position_id}",
|
|
json={"ablaufdatum": (date.today() - timedelta(days=3)).isoformat()},
|
|
headers=auth_header(token),
|
|
)
|
|
response = await client.get("/api/v1/dashboard/ablaufdaten", headers=auth_header(token))
|
|
body = response.json()
|
|
assert len(body) == 1
|
|
assert body[0]["status"] == "abgelaufen"
|
|
assert body[0]["verbleibende_tage"] == -3
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_objekt_ohne_maengel_und_kontrolliert_ist_einsatzbereit(
|
|
client, objekt_mit_position, mitarbeiter_user, materialverantwortlicher_user
|
|
):
|
|
"""Kontrolliert (Ist=Soll, kein Fehlbestand) und ohne sonstige Mängel -> bereit.
|
|
Unterscheidet sich bewusst von test_nie_kontrolliertes_objekt_ist_unbekannt."""
|
|
objekt, material = objekt_mit_position
|
|
mitarbeiter_token = await login(client, "mitarbeiter1")
|
|
await _fehlbestand(client, mitarbeiter_token, objekt, material, "10") # = Sollmenge, kein Fehlbestand
|
|
|
|
token = await login(client, "materialverantwortlicher1")
|
|
response = await client.get("/api/v1/dashboard/einsatzbereitschaft", headers=auth_header(token))
|
|
body = response.json()
|
|
assert body["einsatzbereit"] == 1
|
|
assert body["eingeschraenkt_einsatzbereit"] == 0
|
|
assert body["unbekannt"] == 0
|
|
assert body["nicht_einsatzbereit"] == 0
|
|
assert body["details"] == []
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_nie_kontrolliertes_objekt_ist_unbekannt(
|
|
client, objekt_mit_position, materialverantwortlicher_user
|
|
):
|
|
"""Nutzer-Fund: ein nie kontrolliertes Objekt darf NICHT als einsatzbereit zählen
|
|
(unterscheidet "nie geprüft" von "geprüft und in Ordnung")."""
|
|
objekt, _material = objekt_mit_position
|
|
token = await login(client, "materialverantwortlicher1")
|
|
response = await client.get("/api/v1/dashboard/einsatzbereitschaft", headers=auth_header(token))
|
|
body = response.json()
|
|
assert body["einsatzbereit"] == 0
|
|
assert body["unbekannt"] == 1
|
|
assert body["nicht_einsatzbereit"] == 0
|
|
assert body["details"][0]["gruende"] == ["noch_nie_kontrolliert"]
|
|
assert body["nie_kontrolliert_gesamt"] == 1
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_offener_fehlbestand_macht_objekt_nicht_einsatzbereit(
|
|
client, objekt_mit_position, mitarbeiter_user, materialverantwortlicher_user
|
|
):
|
|
objekt, material = objekt_mit_position
|
|
mitarbeiter_token = await login(client, "mitarbeiter1")
|
|
await _fehlbestand(client, mitarbeiter_token, objekt, material, "6")
|
|
|
|
verantwortlicher_token = await login(client, "materialverantwortlicher1")
|
|
response = await client.get(
|
|
"/api/v1/dashboard/einsatzbereitschaft", headers=auth_header(verantwortlicher_token)
|
|
)
|
|
body = response.json()
|
|
assert body["einsatzbereit"] == 0
|
|
assert body["nicht_einsatzbereit"] == 1
|
|
assert body["details"][0]["gruende"] == ["fehlbestand_offen", "noch_nie_kontrolliert"]
|
|
assert body["nie_kontrolliert_gesamt"] == 1
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_genehmigte_mindermenge_macht_objekt_eingeschraenkt_einsatzbereit(
|
|
client, objekt_mit_position, mitarbeiter_user, materialverantwortlicher_user
|
|
):
|
|
"""U2/Karte 08: Genehmigung erledigt den Fehlbestand fachlich nicht -
|
|
eingeschränkt einsatzbereit, nicht (voll) einsatzbereit."""
|
|
objekt, material = objekt_mit_position
|
|
mitarbeiter_token = await login(client, "mitarbeiter1")
|
|
fehlbestand_id = await _fehlbestand(client, mitarbeiter_token, objekt, material, "6")
|
|
|
|
verantwortlicher_token = await login(client, "materialverantwortlicher1")
|
|
await client.post(
|
|
f"/api/v1/fehlbestaende/{fehlbestand_id}/mindermenge",
|
|
json={"begruendung": "Lieferengpass"},
|
|
headers=auth_header(verantwortlicher_token),
|
|
)
|
|
|
|
response = await client.get(
|
|
"/api/v1/dashboard/einsatzbereitschaft", headers=auth_header(verantwortlicher_token)
|
|
)
|
|
body = response.json()
|
|
assert body["einsatzbereit"] == 0
|
|
assert body["eingeschraenkt_einsatzbereit"] == 1
|
|
assert body["nicht_einsatzbereit"] == 0
|
|
assert body["details"][0]["gruende"] == ["mindermenge_genehmigt", "noch_nie_kontrolliert"]
|
|
assert body["nie_kontrolliert_gesamt"] == 1
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_ueberfaellige_pruefung_macht_objekt_nicht_einsatzbereit(
|
|
client, db_session, objekt_mit_position, materialverantwortlicher_user
|
|
):
|
|
objekt, material = objekt_mit_position
|
|
result = await db_session.execute(
|
|
select(Objektposition).where(
|
|
Objektposition.objekt_id == objekt.id, Objektposition.material_id == material.id
|
|
)
|
|
)
|
|
position = result.scalar_one()
|
|
db_session.add(
|
|
GeraetInstanz(
|
|
objektposition_id=position.id,
|
|
seriennummer="SN-DASH-1",
|
|
naechste_pruefung=date.today() - timedelta(days=1),
|
|
)
|
|
)
|
|
await db_session.flush()
|
|
|
|
token = await login(client, "materialverantwortlicher1")
|
|
response = await client.get("/api/v1/dashboard/einsatzbereitschaft", headers=auth_header(token))
|
|
body = response.json()
|
|
assert body["nicht_einsatzbereit"] == 1
|
|
assert body["details"][0]["gruende"] == ["pruefung_ueberfaellig", "noch_nie_kontrolliert"]
|
|
assert body["nie_kontrolliert_gesamt"] == 1
|
|
|
|
prueftermine = await client.get("/api/v1/dashboard/prueftermine", headers=auth_header(token))
|
|
prueftermine_body = prueftermine.json()
|
|
assert len(prueftermine_body) == 1
|
|
assert prueftermine_body[0]["status"] == "abgelaufen"
|
|
assert prueftermine_body[0]["seriennummer"] == "SN-DASH-1"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_gueltige_pruefung_taucht_nicht_in_prueftermine_auf(
|
|
client, db_session, objekt_mit_position, materialverantwortlicher_user
|
|
):
|
|
objekt, material = objekt_mit_position
|
|
result = await db_session.execute(
|
|
select(Objektposition).where(
|
|
Objektposition.objekt_id == objekt.id, Objektposition.material_id == material.id
|
|
)
|
|
)
|
|
position = result.scalar_one()
|
|
db_session.add(
|
|
GeraetInstanz(
|
|
objektposition_id=position.id,
|
|
seriennummer="SN-DASH-2",
|
|
naechste_pruefung=date.today() + timedelta(days=365),
|
|
)
|
|
)
|
|
await db_session.flush()
|
|
|
|
token = await login(client, "materialverantwortlicher1")
|
|
response = await client.get("/api/v1/dashboard/prueftermine", headers=auth_header(token))
|
|
assert response.json() == []
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_mitarbeiter_darf_neue_dashboard_endpunkte_nicht_sehen(client, mitarbeiter_user):
|
|
token = await login(client, "mitarbeiter1")
|
|
for pfad in ("/api/v1/dashboard/einsatzbereitschaft", "/api/v1/dashboard/prueftermine"):
|
|
response = await client.get(pfad, headers=auth_header(token))
|
|
assert response.status_code == 403
|