patrickandClaude Sonnet 5
2dd7e574e6
Neue Tabellen hersteller/modell (Migration 0030), CRUD-Endpunkte (/hersteller, /modelle) nach bestehendem Kategorie-Muster, inkl. 409 bei Duplikaten (IntegrityError abgefangen wie in personal.py). geraet_instanz bekommt optionales modell_id-Feld - verknüpft eine konkrete Geräteinstanz (mit Seriennummer, IDENT-004) mit ihrem Hersteller/Modell. Ergänzt Material.hersteller (bleibt unverändert als Freitext) um eine normalisierte Variante gezielt für Geräte mit Seriennummer - keine Datenmigration bestehender Freitext-Werte (Scope dieser Kachel). Bewusst kein neues Admin-UI-Screen für die Pflege (Scope-Grenze der Kachel) - Hersteller/Modell aktuell nur über die API verwaltbar. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01CVgbozhYmuEhiEJHffRXCV
416 lines
15 KiB
Python
416 lines
15 KiB
Python
import pytest
|
|
from sqlalchemy import select
|
|
|
|
from app.models.stammdaten import Fach
|
|
from app.models.vorlage import Vorlagenposition
|
|
from tests.conftest import auth_header, login
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_admin_can_create_bereich(client, admin_user):
|
|
token = await login(client, "admin1")
|
|
response = await client.post(
|
|
"/api/v1/bereiche", json={"name": "Rettungsdienst"}, headers=auth_header(token)
|
|
)
|
|
assert response.status_code == 201
|
|
assert response.json()["name"] == "Rettungsdienst"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_mitarbeiter_cannot_create_bereich(client, mitarbeiter_user):
|
|
token = await login(client, "mitarbeiter1")
|
|
response = await client.post(
|
|
"/api/v1/bereiche", json={"name": "Rettungsdienst"}, headers=auth_header(token)
|
|
)
|
|
assert response.status_code == 403
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_mitarbeiter_can_list_bereiche(client, mitarbeiter_user, admin_user, db_session):
|
|
from app.models.stammdaten import Bereich
|
|
|
|
db_session.add(Bereich(name="KatS"))
|
|
await db_session.flush()
|
|
|
|
token = await login(client, "mitarbeiter1")
|
|
response = await client.get("/api/v1/bereiche", headers=auth_header(token))
|
|
assert response.status_code == 200
|
|
assert any(b["name"] == "KatS" for b in response.json())
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_material_lifecycle(client, admin_user):
|
|
token = await login(client, "admin1")
|
|
create_response = await client.post(
|
|
"/api/v1/materialien",
|
|
json={"name": "Kompresse steril", "einheit": "Stück", "materialtyp": "standard"},
|
|
headers=auth_header(token),
|
|
)
|
|
assert create_response.status_code == 201
|
|
material_id = create_response.json()["id"]
|
|
|
|
patch_response = await client.patch(
|
|
f"/api/v1/materialien/{material_id}",
|
|
json={"aktiv": False},
|
|
headers=auth_header(token),
|
|
)
|
|
assert patch_response.status_code == 200
|
|
assert patch_response.json()["aktiv"] is False
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_material_not_found(client, admin_user):
|
|
token = await login(client, "admin1")
|
|
response = await client.get("/api/v1/materialien/999999", headers=auth_header(token))
|
|
assert response.status_code == 404
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_material_zusammenfuehren_ohne_konflikt(client, db_session, admin_user, objekt_mit_position):
|
|
"""Import-Dublette: quelle_id verschwindet, Referenzen zeigen auf ziel_material_id."""
|
|
from app.models.fehlbestand import Fehlbestand, FehlbestandStatus
|
|
from datetime import datetime, timezone
|
|
|
|
objekt, material_ziel = objekt_mit_position
|
|
token = await login(client, "admin1")
|
|
|
|
quelle = await client.post(
|
|
"/api/v1/materialien",
|
|
json={"name": "Kompresse (Dublette)", "einheit": "Stück", "materialtyp": "standard"},
|
|
headers=auth_header(token),
|
|
)
|
|
quelle_id = quelle.json()["id"]
|
|
|
|
db_session.add(
|
|
Fehlbestand(
|
|
erzeugt_von_server_id=objekt.zustaendiger_server_id,
|
|
objekt_id=objekt.id,
|
|
material_id=quelle_id,
|
|
standort_id=objekt.standort_id,
|
|
sollmenge=5,
|
|
istmenge=0,
|
|
fehlmenge=5,
|
|
entstanden_am=datetime.now(timezone.utc),
|
|
festgestellt_von=admin_user.id,
|
|
status=FehlbestandStatus.offen,
|
|
)
|
|
)
|
|
await db_session.flush()
|
|
|
|
verwendung = await client.get(f"/api/v1/materialien/{quelle_id}/verwendung", headers=auth_header(token))
|
|
assert verwendung.status_code == 200
|
|
assert verwendung.json()["fehlbestand"] == 1
|
|
|
|
ergebnis = await client.post(
|
|
f"/api/v1/materialien/{quelle_id}/ersetzen",
|
|
json={"ziel_material_id": material_ziel.id},
|
|
headers=auth_header(token),
|
|
)
|
|
assert ergebnis.status_code == 200
|
|
assert ergebnis.json()["erfolgreich"] is True
|
|
assert ergebnis.json()["konflikte"] == []
|
|
|
|
geloescht = await client.get(f"/api/v1/materialien/{quelle_id}", headers=auth_header(token))
|
|
assert geloescht.status_code == 404
|
|
|
|
fb = await db_session.execute(
|
|
select(Fehlbestand).where(Fehlbestand.objekt_id == objekt.id, Fehlbestand.sollmenge == 5)
|
|
)
|
|
assert fb.scalar_one().material_id == material_ziel.id
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_material_zusammenfuehren_mit_konflikt_wird_abgelehnt(
|
|
client, db_session, admin_user, objekt_mit_position, material_factory
|
|
):
|
|
"""Beide Materialien haben eine Position am selben Objekt - Merge würde die
|
|
UNIQUE(objekt_id, material_id)-Regel verletzen, muss also abgelehnt werden."""
|
|
from app.models.objektposition import Objektposition
|
|
|
|
objekt, material_ziel = objekt_mit_position
|
|
token = await login(client, "admin1")
|
|
material_quelle = await material_factory("Kompresse (Konflikt)")
|
|
|
|
db_session.add(
|
|
Objektposition(objekt_id=objekt.id, material_id=material_quelle.id, sollmenge_vorlage=1, istmenge=0)
|
|
)
|
|
await db_session.flush()
|
|
|
|
ergebnis = await client.post(
|
|
f"/api/v1/materialien/{material_quelle.id}/ersetzen",
|
|
json={"ziel_material_id": material_ziel.id},
|
|
headers=auth_header(token),
|
|
)
|
|
assert ergebnis.status_code == 200
|
|
assert ergebnis.json()["erfolgreich"] is False
|
|
assert len(ergebnis.json()["konflikte"]) == 1
|
|
|
|
weiterhin_da = await client.get(f"/api/v1/materialien/{material_quelle.id}", headers=auth_header(token))
|
|
assert weiterhin_da.status_code == 200
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_admin_can_update_bereich(client, admin_user):
|
|
token = await login(client, "admin1")
|
|
erstellt = await client.post("/api/v1/bereiche", json={"name": "Alt"}, headers=auth_header(token))
|
|
bereich_id = erstellt.json()["id"]
|
|
|
|
response = await client.patch(
|
|
f"/api/v1/bereiche/{bereich_id}", json={"name": "Neu"}, headers=auth_header(token)
|
|
)
|
|
assert response.status_code == 200
|
|
assert response.json()["name"] == "Neu"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_admin_can_update_kategorie(client, admin_user):
|
|
token = await login(client, "admin1")
|
|
bereich = await client.post("/api/v1/bereiche", json={"name": "B"}, headers=auth_header(token))
|
|
erstellt = await client.post(
|
|
"/api/v1/kategorien",
|
|
json={"bereich_id": bereich.json()["id"], "name": "Alt"},
|
|
headers=auth_header(token),
|
|
)
|
|
kategorie_id = erstellt.json()["id"]
|
|
|
|
response = await client.patch(
|
|
f"/api/v1/kategorien/{kategorie_id}", json={"name": "Neu"}, headers=auth_header(token)
|
|
)
|
|
assert response.status_code == 200
|
|
assert response.json()["name"] == "Neu"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_admin_can_update_standort(client, admin_user):
|
|
token = await login(client, "admin1")
|
|
erstellt = await client.post("/api/v1/standorte", json={"name": "Alt"}, headers=auth_header(token))
|
|
standort_id = erstellt.json()["id"]
|
|
|
|
response = await client.patch(
|
|
f"/api/v1/standorte/{standort_id}", json={"name": "Neu"}, headers=auth_header(token)
|
|
)
|
|
assert response.status_code == 200
|
|
assert response.json()["name"] == "Neu"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_admin_can_update_objekttyp(client, admin_user):
|
|
token = await login(client, "admin1")
|
|
bereich = await client.post("/api/v1/bereiche", json={"name": "B"}, headers=auth_header(token))
|
|
erstellt = await client.post(
|
|
"/api/v1/objekttypen",
|
|
json={"bereich_id": bereich.json()["id"], "name": "Alt"},
|
|
headers=auth_header(token),
|
|
)
|
|
objekttyp_id = erstellt.json()["id"]
|
|
|
|
response = await client.patch(
|
|
f"/api/v1/objekttypen/{objekttyp_id}", json={"name": "Neu"}, headers=auth_header(token)
|
|
)
|
|
assert response.status_code == 200
|
|
assert response.json()["name"] == "Neu"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_fach_lifecycle(client, admin_user):
|
|
"""Nutzer-Vorgabe: feste Fächer-Liste je Objekttyp."""
|
|
token = await login(client, "admin1")
|
|
bereich = await client.post("/api/v1/bereiche", json={"name": "B"}, headers=auth_header(token))
|
|
objekttyp = await client.post(
|
|
"/api/v1/objekttypen",
|
|
json={"bereich_id": bereich.json()["id"], "name": "Rucksack"},
|
|
headers=auth_header(token),
|
|
)
|
|
objekttyp_id = objekttyp.json()["id"]
|
|
|
|
erstellt = await client.post(
|
|
"/api/v1/faecher",
|
|
json={"objekttyp_id": objekttyp_id, "name": "Fach 1", "sortierung": 1},
|
|
headers=auth_header(token),
|
|
)
|
|
assert erstellt.status_code == 201
|
|
fach_id = erstellt.json()["id"]
|
|
|
|
liste = await client.get("/api/v1/faecher", headers=auth_header(token))
|
|
assert any(f["id"] == fach_id for f in liste.json())
|
|
|
|
patch_response = await client.patch(
|
|
f"/api/v1/faecher/{fach_id}", json={"name": "Fach 1 links"}, headers=auth_header(token)
|
|
)
|
|
assert patch_response.status_code == 200
|
|
assert patch_response.json()["name"] == "Fach 1 links"
|
|
|
|
delete_response = await client.delete(f"/api/v1/faecher/{fach_id}", headers=auth_header(token))
|
|
assert delete_response.status_code == 204
|
|
|
|
liste_nach_delete = await client.get("/api/v1/faecher", headers=auth_header(token))
|
|
assert not any(f["id"] == fach_id for f in liste_nach_delete.json())
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_fach_duplikat_case_insensitiv_ist_409(client, admin_user):
|
|
token = await login(client, "admin1")
|
|
bereich = await client.post("/api/v1/bereiche", json={"name": "B-Dup"}, headers=auth_header(token))
|
|
objekttyp = await client.post(
|
|
"/api/v1/objekttypen",
|
|
json={"bereich_id": bereich.json()["id"], "name": "Typ-Dup"},
|
|
headers=auth_header(token),
|
|
)
|
|
objekttyp_id = objekttyp.json()["id"]
|
|
|
|
erster = await client.post(
|
|
"/api/v1/faecher", json={"objekttyp_id": objekttyp_id, "name": "Hauptfach"}, headers=auth_header(token)
|
|
)
|
|
assert erster.status_code == 201
|
|
|
|
zweiter = await client.post(
|
|
"/api/v1/faecher", json={"objekttyp_id": objekttyp_id, "name": "hauptfach"}, headers=auth_header(token)
|
|
)
|
|
assert zweiter.status_code == 409
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_fach_verwendung_und_ersetzen_fuehrt_zusammen(
|
|
client, db_session, admin_user, hauptserver_id, standort_factory
|
|
):
|
|
from app.schemas.vorlage import VorlagenpositionCreate
|
|
from app.services.objekte import objekt_aus_vorlage
|
|
from app.services.vorlagen import erstelle_vorlage
|
|
|
|
token = await login(client, "admin1")
|
|
bereich = await client.post("/api/v1/bereiche", json={"name": "B-Merge"}, headers=auth_header(token))
|
|
objekttyp = await client.post(
|
|
"/api/v1/objekttypen",
|
|
json={"bereich_id": bereich.json()["id"], "name": "Typ-Merge"},
|
|
headers=auth_header(token),
|
|
)
|
|
objekttyp_id = objekttyp.json()["id"]
|
|
material = await client.post(
|
|
"/api/v1/materialien",
|
|
json={"name": "Material-Merge", "einheit": "Stück", "materialtyp": "standard"},
|
|
headers=auth_header(token),
|
|
)
|
|
material_id = material.json()["id"]
|
|
|
|
vorlage = await erstelle_vorlage(
|
|
db_session,
|
|
objekttyp_id=objekttyp_id,
|
|
name="Vorlage-Merge",
|
|
positionen=[VorlagenpositionCreate(material_id=material_id, sollmenge=1, fach="Reizverschlussfach gross")],
|
|
)
|
|
standort = await standort_factory("Standort-Merge")
|
|
objekt = await objekt_aus_vorlage(
|
|
db_session,
|
|
code="MERGE-01",
|
|
name="Objekt-Merge",
|
|
standort_id=standort.id,
|
|
vorlage=vorlage,
|
|
zustaendiger_server_id=hauptserver_id,
|
|
)
|
|
|
|
# erstelle_vorlage() legt für den Freitext-Fachnamen der Vorlagenposition
|
|
# bereits automatisch ein Fach an (services/vorlagen.py::_erstelle_faecher)
|
|
# - das ist hier schon das spätere "Ziel", ein erneutes Anlegen desselben
|
|
# Namens würde am case-insensitiven Duplikat-Schutz scheitern (409).
|
|
ziel = (
|
|
await db_session.execute(
|
|
select(Fach).where(Fach.objekttyp_id == objekttyp_id, Fach.name == "Reizverschlussfach gross")
|
|
)
|
|
).scalar_one()
|
|
|
|
quell_fach = await client.post(
|
|
"/api/v1/faecher",
|
|
json={"objekttyp_id": objekttyp_id, "name": "reizverschlussfach gross tmp"},
|
|
headers=auth_header(token),
|
|
)
|
|
# Feld manuell auf den echten (klein geschriebenen) Namen zurücksetzen, um
|
|
# die reale Dublette nachzustellen (Anlegen selbst blockt Duplikate ab).
|
|
quell = (await db_session.execute(select(Fach).where(Fach.id == quell_fach.json()["id"]))).scalar_one()
|
|
quell.name = "reizverschlussfach gross"
|
|
await db_session.flush()
|
|
|
|
verwendung = await client.get(f"/api/v1/faecher/{quell.id}/objekte", headers=auth_header(token))
|
|
assert verwendung.status_code == 200
|
|
assert any(o["id"] == objekt.id for o in verwendung.json())
|
|
|
|
ersetzt = await client.post(
|
|
f"/api/v1/faecher/{quell.id}/ersetzen",
|
|
json={"neuer_name": "Reizverschlussfach Gross"},
|
|
headers=auth_header(token),
|
|
)
|
|
assert ersetzt.status_code == 200
|
|
assert ersetzt.json()["id"] == ziel.id
|
|
|
|
liste = await client.get("/api/v1/faecher", headers=auth_header(token))
|
|
verbleibende = [f for f in liste.json() if f["objekttyp_id"] == objekttyp_id]
|
|
assert len(verbleibende) == 1
|
|
|
|
position = await db_session.execute(
|
|
select(Vorlagenposition).where(Vorlagenposition.vorlage_id == vorlage.id)
|
|
)
|
|
assert position.scalar_one().fach == "Reizverschlussfach Gross"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_mitarbeiter_cannot_create_fach(client, mitarbeiter_user, admin_user, db_session):
|
|
from app.models.stammdaten import Bereich, Objekttyp
|
|
|
|
bereich = Bereich(name="B2")
|
|
db_session.add(bereich)
|
|
await db_session.flush()
|
|
objekttyp = Objekttyp(bereich_id=bereich.id, name="Rucksack2")
|
|
db_session.add(objekttyp)
|
|
await db_session.flush()
|
|
|
|
token = await login(client, "mitarbeiter1")
|
|
response = await client.post(
|
|
"/api/v1/faecher",
|
|
json={"objekttyp_id": objekttyp.id, "name": "Fach X"},
|
|
headers=auth_header(token),
|
|
)
|
|
assert response.status_code == 403
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_hersteller_modell_lifecycle(client, admin_user, mitarbeiter_user):
|
|
"""IDENT-005: Hersteller/Modell-Stammdaten anlegen, per hersteller_id filtern."""
|
|
admin_token = await login(client, "admin1")
|
|
|
|
hersteller = await client.post(
|
|
"/api/v1/hersteller", json={"name": "Weinmann"}, headers=auth_header(admin_token)
|
|
)
|
|
assert hersteller.status_code == 201
|
|
hersteller_id = hersteller.json()["id"]
|
|
|
|
modell = await client.post(
|
|
"/api/v1/modelle",
|
|
json={"hersteller_id": hersteller_id, "name": "MEDUMAT Standard2"},
|
|
headers=auth_header(admin_token),
|
|
)
|
|
assert modell.status_code == 201
|
|
assert modell.json()["hersteller_id"] == hersteller_id
|
|
|
|
anderer_hersteller = await client.post(
|
|
"/api/v1/hersteller", json={"name": "Weinmann"}, headers=auth_header(admin_token)
|
|
)
|
|
assert anderer_hersteller.status_code == 409
|
|
|
|
unbekannter_hersteller = await client.post(
|
|
"/api/v1/modelle",
|
|
json={"hersteller_id": 999999, "name": "Irgendwas"},
|
|
headers=auth_header(admin_token),
|
|
)
|
|
assert unbekannter_hersteller.status_code == 404
|
|
|
|
mitarbeiter_token = await login(client, "mitarbeiter1")
|
|
verweigert = await client.post(
|
|
"/api/v1/hersteller", json={"name": "Dräger"}, headers=auth_header(mitarbeiter_token)
|
|
)
|
|
assert verweigert.status_code == 403
|
|
|
|
liste = await client.get(
|
|
f"/api/v1/modelle?hersteller_id={hersteller_id}", headers=auth_header(mitarbeiter_token)
|
|
)
|
|
assert len(liste.json()) == 1
|
|
assert liste.json()[0]["name"] == "MEDUMAT Standard2"
|