patrickandClaude Sonnet 5
5f568c365c
Anlegen prüft jetzt case-insensitiv auf Duplikate (409 statt "Fach A"
neben "fach a"). Neuer Endpunkt GET /faecher/{id}/objekte zeigt, welche
Objekte ein Fach über ihre Vorlage tatsächlich nutzen. POST
/faecher/{id}/ersetzen benennt um und zieht alle Vorlagenpositionen
mit; landet der neue Name auf einem bereits bestehenden Fach, werden
beide zusammengeführt. Frontend: Fach anklicken öffnet Verwendung +
Ersetzen-Eingabe direkt in der Fächer-Verwaltung.
Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01CVgbozhYmuEhiEJHffRXCV
283 lines
10 KiB
Python
283 lines
10 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_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,
|
|
)
|
|
|
|
ziel_fach = await client.post(
|
|
"/api/v1/faecher",
|
|
json={"objekttyp_id": objekttyp_id, "name": "Reizverschlussfach Gross"},
|
|
headers=auth_header(token),
|
|
)
|
|
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_fach.json()["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
|