patrickandClaude Sonnet 5
8fd507cc5d
CI / backend-tests (push) Successful in 38s
- Kontrolle/Kontrollposition-Modelle: Kontrolle ändert NIEMALS Objektposition.istmenge (Vier-Kernbegriffe-Tabelle, Prompt 02.9) - reine Erfassung/Snapshot - Objekt-Sperre (Prompt 02.8): Kontrolle.status='in_bearbeitung' + benutzer_id IST die Sperre, kein separates Sperr-Modell. Zweiter Zugriff -> 409 mit wer/seit; Übernahme bricht alte Kontrolle ab (Datensatz bleibt, Prompt 16.6-Analogie) und protokolliert das Ereignis in einer neuen, minimalen Historie-Tabelle (volle Ausbaustufe Sprint 5) - Automatische Fehlbestand-Erzeugung bei Ist < Soll (U1), Überbestand erzeugt ausdrücklich KEINEN Fehlbestand (Prompt 02.4); Korrektur vor Abschluss möglich (von derselben Kontrolle erzeugter Fehlbestand ist bis zum Abschluss nicht final) - Abschluss verweigert bei unbestätigten Positionen, liefert deren IDs (U11) - Abbruch verwirft Kontrollpositionen UND von dieser Kontrolle erzeugte Fehlbestände, Kontrolle selbst bleibt als Datensatz mit Status "abgebrochen" erhalten (U12) - Tests: U1, U11, U12, Sperre/409, Übernahme, Fremdzugriff/403, parallele Objekte Mitarbeiter-UI (PWA-Frontend) ist bewusst noch nicht Teil dieses Commits - eigenes Techstack-Setup, wird als nächster Schritt separat angegangen. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01L85hmKbvX7Cqkq47KnQhFt
239 lines
8.5 KiB
Python
239 lines
8.5 KiB
Python
import pytest
|
|
from sqlalchemy import select
|
|
|
|
from app.models.fehlbestand import Fehlbestand
|
|
from app.models.kontrolle import Kontrolle, KontrollStatus, Kontrollposition
|
|
from tests.conftest import auth_header, login
|
|
|
|
|
|
async def _starte(client, token, objekt_id, uebernehmen=False):
|
|
return await client.post(
|
|
f"/api/v1/objekte/{objekt_id}/kontrollen",
|
|
json={"uebernehmen": uebernehmen},
|
|
headers=auth_header(token),
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_fehlbestand_entsteht_bei_abweichung(client, db_session, objekt_mit_position, mitarbeiter_user):
|
|
"""U1: Fehlbestand entsteht automatisch bei Ist < Soll."""
|
|
objekt, material = objekt_mit_position
|
|
token = await login(client, "mitarbeiter1")
|
|
start = await _starte(client, token, objekt.id)
|
|
assert start.status_code == 201
|
|
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(token),
|
|
)
|
|
assert put_response.status_code == 200
|
|
assert put_response.json()["abweichung"] is True
|
|
|
|
result = await db_session.execute(select(Fehlbestand).where(Fehlbestand.kontrolle_id == kontrolle_id))
|
|
fehlbestand = result.scalar_one()
|
|
assert float(fehlbestand.fehlmenge) == 2
|
|
assert fehlbestand.status.value == "offen"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_ueberbestand_erzeugt_keinen_fehlbestand(client, db_session, objekt_mit_position, mitarbeiter_user):
|
|
objekt, material = objekt_mit_position
|
|
token = await login(client, "mitarbeiter1")
|
|
start = await _starte(client, token, objekt.id)
|
|
kontrolle_id = start.json()["id"]
|
|
|
|
put_response = await client.put(
|
|
f"/api/v1/kontrollen/{kontrolle_id}/positionen/{material.id}",
|
|
json={"istmenge": "12"},
|
|
headers=auth_header(token),
|
|
)
|
|
assert put_response.status_code == 200
|
|
assert put_response.json()["abweichung"] is True # protokolliert...
|
|
|
|
result = await db_session.execute(select(Fehlbestand).where(Fehlbestand.kontrolle_id == kontrolle_id))
|
|
assert result.scalar_one_or_none() is None # ...aber KEIN Fehlbestand (Prompt 02.4)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_abschluss_mit_fehlender_position_ist_409(client, objekt_mit_position, mitarbeiter_user):
|
|
"""U11: Abschluss mit unbestätigten Positionen -> Fehler mit Liste."""
|
|
objekt, material = objekt_mit_position
|
|
token = await login(client, "mitarbeiter1")
|
|
start = await _starte(client, token, objekt.id)
|
|
kontrolle_id = start.json()["id"]
|
|
|
|
response = await client.post(
|
|
f"/api/v1/kontrollen/{kontrolle_id}/abschliessen", headers=auth_header(token)
|
|
)
|
|
assert response.status_code == 409
|
|
assert material.id in response.json()["detail"]["fehlende_material_ids"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_abschluss_erfolgreich_wenn_vollstaendig(client, objekt_mit_position, mitarbeiter_user):
|
|
objekt, material = objekt_mit_position
|
|
token = await login(client, "mitarbeiter1")
|
|
start = await _starte(client, token, objekt.id)
|
|
kontrolle_id = start.json()["id"]
|
|
|
|
await client.put(
|
|
f"/api/v1/kontrollen/{kontrolle_id}/positionen/{material.id}",
|
|
json={"istmenge": "10"},
|
|
headers=auth_header(token),
|
|
)
|
|
response = await client.post(
|
|
f"/api/v1/kontrollen/{kontrolle_id}/abschliessen", headers=auth_header(token)
|
|
)
|
|
assert response.status_code == 200
|
|
assert response.json()["status"] == "abgeschlossen"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_abbruch_verwirft_fehlbestand_und_positionen(
|
|
client, db_session, objekt_mit_position, mitarbeiter_user
|
|
):
|
|
"""U12: abgebrochene Kontrolle - keine Ist-Übernahme, kein Fehlbestand, Objekt
|
|
bleibt unkontrolliert (Kontrolle-Datensatz selbst bleibt aber erhalten)."""
|
|
objekt, material = objekt_mit_position
|
|
token = await login(client, "mitarbeiter1")
|
|
start = await _starte(client, token, objekt.id)
|
|
kontrolle_id = start.json()["id"]
|
|
|
|
await client.put(
|
|
f"/api/v1/kontrollen/{kontrolle_id}/positionen/{material.id}",
|
|
json={"istmenge": "8"},
|
|
headers=auth_header(token),
|
|
)
|
|
response = await client.post(
|
|
f"/api/v1/kontrollen/{kontrolle_id}/abbrechen",
|
|
json={"grund": "Einsatzalarmierung"},
|
|
headers=auth_header(token),
|
|
)
|
|
assert response.status_code == 200
|
|
assert response.json()["status"] == "abgebrochen"
|
|
|
|
fb_result = await db_session.execute(select(Fehlbestand).where(Fehlbestand.kontrolle_id == kontrolle_id))
|
|
assert fb_result.scalar_one_or_none() is None
|
|
|
|
pos_result = await db_session.execute(
|
|
select(Kontrollposition).where(Kontrollposition.kontrolle_id == kontrolle_id)
|
|
)
|
|
assert pos_result.scalar_one_or_none() is None
|
|
|
|
kontrolle = await db_session.get(Kontrolle, kontrolle_id)
|
|
assert kontrolle is not None # Datensatz bleibt erhalten (Prompt 16.6)
|
|
assert kontrolle.status == KontrollStatus.abgebrochen
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_objekt_sperre_zweiter_zugriff_ist_409(
|
|
client, objekt_mit_position, mitarbeiter_user, mitarbeiter2_user
|
|
):
|
|
objekt, _material = objekt_mit_position
|
|
token1 = await login(client, "mitarbeiter1")
|
|
token2 = await login(client, "mitarbeiter2")
|
|
|
|
erster = await _starte(client, token1, objekt.id)
|
|
assert erster.status_code == 201
|
|
|
|
zweiter = await _starte(client, token2, objekt.id)
|
|
assert zweiter.status_code == 409
|
|
assert zweiter.json()["detail"]["benutzer"] == "Test Mitarbeiter"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_uebernahme_beendet_alte_kontrolle_und_verwirft_ihren_fehlbestand(
|
|
client, db_session, objekt_mit_position, mitarbeiter_user, mitarbeiter2_user
|
|
):
|
|
objekt, material = objekt_mit_position
|
|
token1 = await login(client, "mitarbeiter1")
|
|
token2 = await login(client, "mitarbeiter2")
|
|
|
|
erster = await _starte(client, token1, objekt.id)
|
|
alte_kontrolle_id = erster.json()["id"]
|
|
await client.put(
|
|
f"/api/v1/kontrollen/{alte_kontrolle_id}/positionen/{material.id}",
|
|
json={"istmenge": "5"},
|
|
headers=auth_header(token1),
|
|
)
|
|
|
|
uebernahme = await _starte(client, token2, objekt.id, uebernehmen=True)
|
|
assert uebernahme.status_code == 201
|
|
neue_kontrolle_id = uebernahme.json()["id"]
|
|
assert neue_kontrolle_id != alte_kontrolle_id
|
|
|
|
alte_kontrolle = await db_session.get(Kontrolle, alte_kontrolle_id)
|
|
assert alte_kontrolle.status == KontrollStatus.abgebrochen
|
|
|
|
fb_result = await db_session.execute(
|
|
select(Fehlbestand).where(Fehlbestand.kontrolle_id == alte_kontrolle_id)
|
|
)
|
|
assert fb_result.scalar_one_or_none() is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_fremder_benutzer_kann_position_nicht_erfassen(
|
|
client, objekt_mit_position, mitarbeiter_user, mitarbeiter2_user
|
|
):
|
|
objekt, material = objekt_mit_position
|
|
token1 = await login(client, "mitarbeiter1")
|
|
token2 = await login(client, "mitarbeiter2")
|
|
|
|
start = await _starte(client, token1, objekt.id)
|
|
kontrolle_id = start.json()["id"]
|
|
|
|
response = await client.put(
|
|
f"/api/v1/kontrollen/{kontrolle_id}/positionen/{material.id}",
|
|
json={"istmenge": "5"},
|
|
headers=auth_header(token2),
|
|
)
|
|
assert response.status_code == 403
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_verschiedene_objekte_parallel_kontrollierbar(
|
|
client,
|
|
db_session,
|
|
hauptserver_id,
|
|
objekt_mit_position,
|
|
standort_factory,
|
|
objekttyp_factory,
|
|
material_factory,
|
|
mitarbeiter_user,
|
|
mitarbeiter2_user,
|
|
):
|
|
"""Sperre gilt pro Objekt, nicht global (Prompt 02.8 Klarstellung)."""
|
|
from app.schemas.vorlage import VorlagenpositionCreate
|
|
from app.services.objekte import objekt_aus_vorlage
|
|
from app.services.vorlagen import erstelle_vorlage
|
|
|
|
objekt_a, _material_a = objekt_mit_position
|
|
token1 = await login(client, "mitarbeiter1")
|
|
token2 = await login(client, "mitarbeiter2")
|
|
|
|
objekttyp_b = await objekttyp_factory("Objekttyp-B")
|
|
standort_b = await standort_factory("Standort-B")
|
|
material_b = await material_factory("Verbandspäckchen")
|
|
vorlage_b = await erstelle_vorlage(
|
|
db_session,
|
|
objekttyp_id=objekttyp_b.id,
|
|
name="Vorlage-B",
|
|
positionen=[VorlagenpositionCreate(material_id=material_b.id, sollmenge=5)],
|
|
)
|
|
objekt_b = await objekt_aus_vorlage(
|
|
db_session,
|
|
code="OB-01",
|
|
name="Objekt B",
|
|
standort_id=standort_b.id,
|
|
vorlage=vorlage_b,
|
|
zustaendiger_server_id=hauptserver_id,
|
|
)
|
|
|
|
erster = await _starte(client, token1, objekt_a.id)
|
|
assert erster.status_code == 201
|
|
|
|
zweiter = await _starte(client, token2, objekt_b.id)
|
|
assert zweiter.status_code == 201
|