Neue Stammdaten-Tabelle nummernschema (Migration 0032): Präfix + Stellenzahl einmal je Objekttyp festlegen statt bei jeder Objekt-Anlage manuell einzutippen (Fehlerquelle - Tippfehler, uneinheitliche Schreibweise, falsches Präfix für den Typ). GET /objekte/naechster-code akzeptiert jetzt objekttyp_id als Alternative zu praefix; ohne konfiguriertes Schema bleibt der bisherige manuelle Präfix-Weg nutzbar (Rückwärtskompatibilität). Admin-UI: Nummernschema-Pflege direkt in der Objekttyp-Verwaltung (ObjekttypSection). ObjektAnlegenFormular versucht zuerst das Schema des gewählten Objekttyps, fällt bei 404 auf manuelles Präfix zurück. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01CVgbozhYmuEhiEJHffRXCV
703 lines
25 KiB
Python
703 lines
25 KiB
Python
import pytest
|
|
|
|
from app.services.vorlagen import erstelle_vorlage
|
|
from app.schemas.vorlage import VorlagenpositionCreate
|
|
from tests.conftest import auth_header, login
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_letzte_kontrolle_am_wird_erst_nach_abschluss_gesetzt(
|
|
client, objekt_mit_position, mitarbeiter_user
|
|
):
|
|
"""Nutzer-Frage: wann war die letzte KOMPLETTE Kontrolle - vorher (None/
|
|
"noch nie"), nach starte_kontrolle allein noch nicht (erst status=abgeschlossen
|
|
zählt), erst nach /abschliessen gesetzt."""
|
|
objekt, material = objekt_mit_position
|
|
token = await login(client, "mitarbeiter1")
|
|
|
|
vor_kontrolle = await client.get(f"/api/v1/objekte/{objekt.id}", headers=auth_header(token))
|
|
assert vor_kontrolle.json()["letzte_kontrolle_am"] is None
|
|
|
|
start = await client.post(
|
|
f"/api/v1/objekte/{objekt.id}/kontrollen", json={"uebernehmen": False}, headers=auth_header(token)
|
|
)
|
|
kontrolle_id = start.json()["id"]
|
|
await client.put(
|
|
f"/api/v1/kontrollen/{kontrolle_id}/positionen/{material.id}",
|
|
json={"istmenge": "10"},
|
|
headers=auth_header(token),
|
|
)
|
|
|
|
waehrend = await client.get(f"/api/v1/objekte/{objekt.id}", headers=auth_header(token))
|
|
assert waehrend.json()["letzte_kontrolle_am"] is None # noch nicht abgeschlossen
|
|
|
|
await client.post(f"/api/v1/kontrollen/{kontrolle_id}/abschliessen", headers=auth_header(token))
|
|
|
|
nach_abschluss = await client.get(f"/api/v1/objekte/{objekt.id}", headers=auth_header(token))
|
|
assert nach_abschluss.json()["letzte_kontrolle_am"] is not None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_objekt_aus_vorlage_startet_unkontrolliert(
|
|
client, db_session, objekttyp_factory, material_factory, standort_factory, admin_user
|
|
):
|
|
objekttyp = await objekttyp_factory("Rucksack Handball")
|
|
material = await material_factory("Kompresse")
|
|
standort = await standort_factory("Wache A")
|
|
vorlage = await erstelle_vorlage(
|
|
db_session,
|
|
objekttyp_id=objekttyp.id,
|
|
name="Rucksack Handball Standard",
|
|
positionen=[VorlagenpositionCreate(material_id=material.id, sollmenge=10)],
|
|
)
|
|
|
|
token = await login(client, "admin1")
|
|
response = await client.post(
|
|
"/api/v1/objekte",
|
|
json={
|
|
"code": "RH-01",
|
|
"name": "Rucksack Handball 1",
|
|
"objekttyp_id": objekttyp.id,
|
|
"vorlage_id": vorlage.id,
|
|
"standort_id": standort.id,
|
|
},
|
|
headers=auth_header(token),
|
|
)
|
|
assert response.status_code == 201
|
|
objekt_id = response.json()["id"]
|
|
|
|
positionen_response = await client.get(
|
|
f"/api/v1/objekte/{objekt_id}/positionen", headers=auth_header(token)
|
|
)
|
|
assert positionen_response.status_code == 200
|
|
positionen = positionen_response.json()
|
|
assert len(positionen) == 1
|
|
assert float(positionen[0]["istmenge"]) == 0
|
|
assert float(positionen[0]["sollmenge_effektiv"]) == 10
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_duplizieren_kopiert_struktur_nicht_zustand(
|
|
client, db_session, objekttyp_factory, material_factory, standort_factory, admin_user
|
|
):
|
|
"""U9: Struktur/Sollmengen ja, Ist-Menge/SN/Ablauf/Charge nein, neuer eindeutiger Code."""
|
|
objekttyp = await objekttyp_factory("Rucksack Handball")
|
|
material = await material_factory("Kompresse")
|
|
standort = await standort_factory("Wache A")
|
|
vorlage = await erstelle_vorlage(
|
|
db_session,
|
|
objekttyp_id=objekttyp.id,
|
|
name="Rucksack Handball Standard",
|
|
positionen=[VorlagenpositionCreate(material_id=material.id, sollmenge=10)],
|
|
)
|
|
|
|
token = await login(client, "admin1")
|
|
create_response = await client.post(
|
|
"/api/v1/objekte",
|
|
json={
|
|
"code": "RH-02",
|
|
"name": "Rucksack Handball 2",
|
|
"objekttyp_id": objekttyp.id,
|
|
"vorlage_id": vorlage.id,
|
|
"standort_id": standort.id,
|
|
},
|
|
headers=auth_header(token),
|
|
)
|
|
objekt_id = create_response.json()["id"]
|
|
|
|
positionen = (
|
|
await client.get(f"/api/v1/objekte/{objekt_id}/positionen", headers=auth_header(token))
|
|
).json()
|
|
position_id = positionen[0]["id"]
|
|
patch_response = await client.patch(
|
|
f"/api/v1/objekte/{objekt_id}/positionen/{position_id}",
|
|
json={"istmenge": "8", "seriennummer": "SN-123", "sollmenge_override": "12"},
|
|
headers=auth_header(token),
|
|
)
|
|
assert patch_response.status_code == 200
|
|
|
|
duplicate_response = await client.post(
|
|
f"/api/v1/objekte/{objekt_id}/duplizieren",
|
|
json={"neuer_code": "RH-03", "neuer_name": "Rucksack Handball 3", "neuer_standort_id": standort.id},
|
|
headers=auth_header(token),
|
|
)
|
|
assert duplicate_response.status_code == 201
|
|
neues_objekt_id = duplicate_response.json()["id"]
|
|
|
|
neue_positionen = (
|
|
await client.get(f"/api/v1/objekte/{neues_objekt_id}/positionen", headers=auth_header(token))
|
|
).json()
|
|
assert len(neue_positionen) == 1
|
|
kopie = neue_positionen[0]
|
|
assert float(kopie["sollmenge_override"]) == 12 # Struktur/Override kopiert
|
|
assert float(kopie["istmenge"]) == 0 # Zustand NICHT kopiert
|
|
assert kopie["seriennummer"] is None # Zustand NICHT kopiert
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_duplizieren_mit_vergebenem_code_ist_409(
|
|
client, db_session, objekttyp_factory, standort_factory, admin_user
|
|
):
|
|
objekttyp = await objekttyp_factory("Fahrzeug")
|
|
standort = await standort_factory("Wache B")
|
|
vorlage = await erstelle_vorlage(
|
|
db_session, objekttyp_id=objekttyp.id, name="Fahrzeug Standard", positionen=[]
|
|
)
|
|
|
|
token = await login(client, "admin1")
|
|
erst = await client.post(
|
|
"/api/v1/objekte",
|
|
json={
|
|
"code": "FZ-01",
|
|
"name": "Fahrzeug 1",
|
|
"objekttyp_id": objekttyp.id,
|
|
"vorlage_id": vorlage.id,
|
|
"standort_id": standort.id,
|
|
},
|
|
headers=auth_header(token),
|
|
)
|
|
objekt_id = erst.json()["id"]
|
|
|
|
# zweites Objekt mit Code, den wir gleich für die Duplizierung kollidieren lassen
|
|
await client.post(
|
|
"/api/v1/objekte",
|
|
json={
|
|
"code": "FZ-02",
|
|
"name": "Fahrzeug 2",
|
|
"objekttyp_id": objekttyp.id,
|
|
"vorlage_id": vorlage.id,
|
|
"standort_id": standort.id,
|
|
},
|
|
headers=auth_header(token),
|
|
)
|
|
|
|
konflikt_response = await client.post(
|
|
f"/api/v1/objekte/{objekt_id}/duplizieren",
|
|
json={"neuer_code": "FZ-02", "neuer_name": "Fahrzeug Kopie", "neuer_standort_id": standort.id},
|
|
headers=auth_header(token),
|
|
)
|
|
assert konflikt_response.status_code == 409
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_mitarbeiter_kann_objekte_nicht_anlegen(
|
|
client, objekttyp_factory, standort_factory, mitarbeiter_user
|
|
):
|
|
objekttyp = await objekttyp_factory("Rucksack")
|
|
standort = await standort_factory("Wache A")
|
|
token = await login(client, "mitarbeiter1")
|
|
response = await client.post(
|
|
"/api/v1/objekte",
|
|
json={
|
|
"code": "X-01",
|
|
"name": "X",
|
|
"objekttyp_id": objekttyp.id,
|
|
"vorlage_id": 999999,
|
|
"standort_id": standort.id,
|
|
},
|
|
headers=auth_header(token),
|
|
)
|
|
assert response.status_code == 403
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_objektposition_code_lookup(
|
|
client, db_session, objekt_mit_position, materialverantwortlicher_user, mitarbeiter_user
|
|
):
|
|
objekt, material = objekt_mit_position
|
|
mv_token = await login(client, "materialverantwortlicher1")
|
|
|
|
positionen = await client.get(f"/api/v1/objekte/{objekt.id}/positionen", headers=auth_header(mv_token))
|
|
position_id = positionen.json()[0]["id"]
|
|
|
|
patch_response = await client.patch(
|
|
f"/api/v1/objekte/{objekt.id}/positionen/{position_id}",
|
|
json={"code": "GR-000123"},
|
|
headers=auth_header(mv_token),
|
|
)
|
|
assert patch_response.status_code == 200
|
|
assert patch_response.json()["code"] == "GR-000123"
|
|
|
|
token = await login(client, "mitarbeiter1")
|
|
lookup_response = await client.get("/api/v1/objektpositionen/code/GR-000123", headers=auth_header(token))
|
|
assert lookup_response.status_code == 200
|
|
body = lookup_response.json()
|
|
assert body["id"] == position_id
|
|
assert body["material_id"] == material.id
|
|
|
|
missing_response = await client.get("/api/v1/objektpositionen/code/UNBEKANNT", headers=auth_header(token))
|
|
assert missing_response.status_code == 404
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_naechster_objekt_code_ohne_bestand(client, admin_user):
|
|
token = await login(client, "admin1")
|
|
response = await client.get("/api/v1/objekte/naechster-code?praefix=RK", headers=auth_header(token))
|
|
assert response.status_code == 200
|
|
assert response.json()["code"] == "RK-00001"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_naechster_objekt_code_zaehlt_hoch(
|
|
client, db_session, objekttyp_factory, standort_factory, admin_user
|
|
):
|
|
objekttyp = await objekttyp_factory("Rucksack Vorschlag")
|
|
standort = await standort_factory("Wache-Vorschlag")
|
|
vorlage = await erstelle_vorlage(
|
|
db_session, objekttyp_id=objekttyp.id, name="Vorschlag Standard", positionen=[]
|
|
)
|
|
token = await login(client, "admin1")
|
|
await client.post(
|
|
"/api/v1/objekte",
|
|
json={
|
|
"code": "RK-00017",
|
|
"name": "Rucksack 17",
|
|
"objekttyp_id": objekttyp.id,
|
|
"vorlage_id": vorlage.id,
|
|
"standort_id": standort.id,
|
|
},
|
|
headers=auth_header(token),
|
|
)
|
|
response = await client.get("/api/v1/objekte/naechster-code?praefix=RK", headers=auth_header(token))
|
|
assert response.json()["code"] == "RK-00018"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_nummernschema_liefert_code_ohne_manuelles_praefix(
|
|
client, objekttyp_factory, admin_user
|
|
):
|
|
"""IDENT-003: Präfix + Stellenzahl kommen aus dem konfigurierten Schema,
|
|
nicht mehr manuell eingegeben."""
|
|
objekttyp = await objekttyp_factory("Typ-Nummernschema")
|
|
token = await login(client, "admin1")
|
|
|
|
schema = await client.post(
|
|
"/api/v1/nummernschemata",
|
|
json={"objekttyp_id": objekttyp.id, "praefix": "SEG", "stellenzahl": 3},
|
|
headers=auth_header(token),
|
|
)
|
|
assert schema.status_code == 201
|
|
|
|
duplikat = await client.post(
|
|
"/api/v1/nummernschemata",
|
|
json={"objekttyp_id": objekttyp.id, "praefix": "ANDERES"},
|
|
headers=auth_header(token),
|
|
)
|
|
assert duplikat.status_code == 409
|
|
|
|
vorschlag = await client.get(
|
|
f"/api/v1/objekte/naechster-code?objekttyp_id={objekttyp.id}", headers=auth_header(token)
|
|
)
|
|
assert vorschlag.status_code == 200
|
|
assert vorschlag.json()["code"] == "SEG-001"
|
|
|
|
aenderung = await client.patch(
|
|
f"/api/v1/nummernschemata/{schema.json()['id']}",
|
|
json={"stellenzahl": 4},
|
|
headers=auth_header(token),
|
|
)
|
|
assert aenderung.json()["stellenzahl"] == 4
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_nummernschema_ohne_konfiguration_verlangt_praefix(
|
|
client, objekttyp_factory, admin_user
|
|
):
|
|
objekttyp = await objekttyp_factory("Typ-Ohne-Schema")
|
|
token = await login(client, "admin1")
|
|
|
|
ohne_schema = await client.get(
|
|
f"/api/v1/objekte/naechster-code?objekttyp_id={objekttyp.id}", headers=auth_header(token)
|
|
)
|
|
assert ohne_schema.status_code == 404
|
|
|
|
ohne_beides = await client.get("/api/v1/objekte/naechster-code", headers=auth_header(token))
|
|
assert ohne_beides.status_code == 422
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_mitarbeiter_darf_nummernschema_nicht_anlegen(client, objekttyp_factory, mitarbeiter_user):
|
|
objekttyp = await objekttyp_factory("Typ-Nummernschema-Verboten")
|
|
token = await login(client, "mitarbeiter1")
|
|
response = await client.post(
|
|
"/api/v1/nummernschemata",
|
|
json={"objekttyp_id": objekttyp.id, "praefix": "X"},
|
|
headers=auth_header(token),
|
|
)
|
|
assert response.status_code == 403
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_naechster_objektposition_code(
|
|
client, db_session, objekt_mit_position, materialverantwortlicher_user
|
|
):
|
|
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))
|
|
position_id = positionen.json()[0]["id"]
|
|
await client.patch(
|
|
f"/api/v1/objekte/{objekt.id}/positionen/{position_id}",
|
|
json={"code": "GR-001042"},
|
|
headers=auth_header(token),
|
|
)
|
|
|
|
response = await client.get(
|
|
"/api/v1/objektpositionen/naechster-code?praefix=GR", headers=auth_header(token)
|
|
)
|
|
assert response.json()["code"] == "GR-001043"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_objekt_label_pdf(client, db_session, objekttyp_factory, standort_factory, admin_user):
|
|
objekttyp = await objekttyp_factory("Rucksack Label")
|
|
standort = await standort_factory("Wache-Label")
|
|
vorlage = await erstelle_vorlage(
|
|
db_session, objekttyp_id=objekttyp.id, name="Label Standard", positionen=[]
|
|
)
|
|
token = await login(client, "admin1")
|
|
erstellt = await client.post(
|
|
"/api/v1/objekte",
|
|
json={
|
|
"code": "LB-01",
|
|
"name": "Label Rucksack",
|
|
"objekttyp_id": objekttyp.id,
|
|
"vorlage_id": vorlage.id,
|
|
"standort_id": standort.id,
|
|
},
|
|
headers=auth_header(token),
|
|
)
|
|
objekt_id = erstellt.json()["id"]
|
|
|
|
response = await client.get(f"/api/v1/objekte/{objekt_id}/label.pdf", headers=auth_header(token))
|
|
assert response.status_code == 200
|
|
assert response.headers["content-type"] == "application/pdf"
|
|
assert response.content.startswith(b"%PDF")
|
|
|
|
qr_response = await client.get(
|
|
f"/api/v1/objekte/{objekt_id}/label.pdf?format=qr", headers=auth_header(token)
|
|
)
|
|
assert qr_response.status_code == 200
|
|
assert qr_response.headers["content-type"] == "application/pdf"
|
|
assert qr_response.content.startswith(b"%PDF")
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_objektposition_label_pdf_ohne_code_ist_404(
|
|
client, db_session, objekt_mit_position, materialverantwortlicher_user
|
|
):
|
|
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))
|
|
position_id = positionen.json()[0]["id"]
|
|
|
|
ohne_code = await client.get(f"/api/v1/objektpositionen/{position_id}/label.pdf", headers=auth_header(token))
|
|
assert ohne_code.status_code == 404
|
|
|
|
await client.patch(
|
|
f"/api/v1/objekte/{objekt.id}/positionen/{position_id}",
|
|
json={"code": "GR-000999"},
|
|
headers=auth_header(token),
|
|
)
|
|
mit_code = await client.get(f"/api/v1/objektpositionen/{position_id}/label.pdf", headers=auth_header(token))
|
|
assert mit_code.status_code == 200
|
|
assert mit_code.content.startswith(b"%PDF")
|
|
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_fahrzeug_zuordnung_bei_anlage_und_nachtraeglich(
|
|
client, db_session, objekttyp_factory, standort_factory, admin_user
|
|
):
|
|
"""Nutzer-Vorgabe: jedes Objekt (Rucksack/Gerät/Anhänger) gehört zu jedem
|
|
Zeitpunkt höchstens einem Fahrzeug (1:n) - Fahrzeug selbst ist einfach ein
|
|
Objekt ohne eigene fahrzeug_id."""
|
|
objekttyp = await objekttyp_factory("Fahrzeug/Zuordnung", ist_zugfahrzeug=True)
|
|
standort = await standort_factory("Wache-Zuordnung")
|
|
vorlage = await erstelle_vorlage(
|
|
db_session, objekttyp_id=objekttyp.id, name="Zuordnung Standard", positionen=[]
|
|
)
|
|
token = await login(client, "admin1")
|
|
|
|
fahrzeug = await client.post(
|
|
"/api/v1/objekte",
|
|
json={
|
|
"code": "ZO-FZ",
|
|
"name": "Zugfahrzeug",
|
|
"objekttyp_id": objekttyp.id,
|
|
"vorlage_id": vorlage.id,
|
|
"standort_id": standort.id,
|
|
},
|
|
headers=auth_header(token),
|
|
)
|
|
fahrzeug_id = fahrzeug.json()["id"]
|
|
assert fahrzeug.json()["fahrzeug_id"] is None
|
|
|
|
zugeordnet = await client.post(
|
|
"/api/v1/objekte",
|
|
json={
|
|
"code": "ZO-RK",
|
|
"name": "Rucksack",
|
|
"objekttyp_id": objekttyp.id,
|
|
"vorlage_id": vorlage.id,
|
|
"standort_id": standort.id,
|
|
"fahrzeug_id": fahrzeug_id,
|
|
},
|
|
headers=auth_header(token),
|
|
)
|
|
assert zugeordnet.status_code == 201
|
|
assert zugeordnet.json()["fahrzeug_id"] == fahrzeug_id
|
|
|
|
# unabhängiges Objekt, erst nachträglich zugeordnet
|
|
unabhaengig = await client.post(
|
|
"/api/v1/objekte",
|
|
json={
|
|
"code": "ZO-GR",
|
|
"name": "Gerät",
|
|
"objekttyp_id": objekttyp.id,
|
|
"vorlage_id": vorlage.id,
|
|
"standort_id": standort.id,
|
|
},
|
|
headers=auth_header(token),
|
|
)
|
|
unabhaengig_id = unabhaengig.json()["id"]
|
|
assert unabhaengig.json()["fahrzeug_id"] is None
|
|
|
|
geaendert = await client.patch(
|
|
f"/api/v1/objekte/{unabhaengig_id}", json={"fahrzeug_id": fahrzeug_id}, headers=auth_header(token)
|
|
)
|
|
assert geaendert.status_code == 200
|
|
assert geaendert.json()["fahrzeug_id"] == fahrzeug_id
|
|
|
|
geloest = await client.patch(
|
|
f"/api/v1/objekte/{unabhaengig_id}", json={"fahrzeug_id": None}, headers=auth_header(token)
|
|
)
|
|
assert geloest.json()["fahrzeug_id"] is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_fahrzeug_zuordnung_lehnt_selbstbezug_und_zyklus_ab(
|
|
client, db_session, objekttyp_factory, standort_factory, admin_user
|
|
):
|
|
objekttyp = await objekttyp_factory("Fahrzeug/Zuordnung Zyklus", ist_zugfahrzeug=True)
|
|
standort = await standort_factory("Wache-Zyklus-Zuordnung")
|
|
vorlage = await erstelle_vorlage(
|
|
db_session, objekttyp_id=objekttyp.id, name="Zyklus-Zuordnung Standard", positionen=[]
|
|
)
|
|
token = await login(client, "admin1")
|
|
|
|
a = await client.post(
|
|
"/api/v1/objekte",
|
|
json={
|
|
"code": "ZYO-A",
|
|
"name": "Objekt A",
|
|
"objekttyp_id": objekttyp.id,
|
|
"vorlage_id": vorlage.id,
|
|
"standort_id": standort.id,
|
|
},
|
|
headers=auth_header(token),
|
|
)
|
|
a_id = a.json()["id"]
|
|
|
|
selbstbezug = await client.patch(
|
|
f"/api/v1/objekte/{a_id}", json={"fahrzeug_id": a_id}, headers=auth_header(token)
|
|
)
|
|
assert selbstbezug.status_code == 409
|
|
|
|
ungueltig = await client.patch(
|
|
f"/api/v1/objekte/{a_id}", json={"fahrzeug_id": 999999}, headers=auth_header(token)
|
|
)
|
|
assert ungueltig.status_code == 404
|
|
|
|
b = await client.post(
|
|
"/api/v1/objekte",
|
|
json={
|
|
"code": "ZYO-B",
|
|
"name": "Objekt B",
|
|
"objekttyp_id": objekttyp.id,
|
|
"vorlage_id": vorlage.id,
|
|
"standort_id": standort.id,
|
|
"fahrzeug_id": a_id,
|
|
},
|
|
headers=auth_header(token),
|
|
)
|
|
b_id = b.json()["id"]
|
|
assert b.json()["fahrzeug_id"] == a_id
|
|
|
|
zyklus = await client.patch(
|
|
f"/api/v1/objekte/{a_id}", json={"fahrzeug_id": b_id}, headers=auth_header(token)
|
|
)
|
|
assert zyklus.status_code == 409
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_fahrzeug_zuordnung_lehnt_nicht_zugfahrzeug_typ_ab(
|
|
client, db_session, objekttyp_factory, standort_factory, admin_user
|
|
):
|
|
"""Karte 11 Nachtrag (Option A+C): nur Objekte, deren Objekttyp
|
|
ist_zugfahrzeug=True ist, dürfen als Fahrzeug zugeordnet werden - grenzt
|
|
z.B. Rucksäcke serverseitig aus der Fahrzeug-Auswahl aus."""
|
|
rucksack_typ = await objekttyp_factory("Rucksack-NichtZugfahrzeug", ist_zugfahrzeug=False)
|
|
standort = await standort_factory("Wache-Zugfahrzeug-Ablehnung")
|
|
vorlage = await erstelle_vorlage(
|
|
db_session, objekttyp_id=rucksack_typ.id, name="Rucksack Standard", positionen=[]
|
|
)
|
|
token = await login(client, "admin1")
|
|
|
|
kein_fahrzeug = await client.post(
|
|
"/api/v1/objekte",
|
|
json={
|
|
"code": "NZF-01",
|
|
"name": "Rucksack",
|
|
"objekttyp_id": rucksack_typ.id,
|
|
"vorlage_id": vorlage.id,
|
|
"standort_id": standort.id,
|
|
},
|
|
headers=auth_header(token),
|
|
)
|
|
rucksack_id = kein_fahrzeug.json()["id"]
|
|
|
|
abgelehnt = await client.post(
|
|
"/api/v1/objekte",
|
|
json={
|
|
"code": "NZF-02",
|
|
"name": "Anderer Rucksack",
|
|
"objekttyp_id": rucksack_typ.id,
|
|
"vorlage_id": vorlage.id,
|
|
"standort_id": standort.id,
|
|
"fahrzeug_id": rucksack_id,
|
|
},
|
|
headers=auth_header(token),
|
|
)
|
|
assert abgelehnt.status_code == 422
|
|
|
|
zweiter_rucksack = await client.post(
|
|
"/api/v1/objekte",
|
|
json={
|
|
"code": "NZF-03",
|
|
"name": "Dritter Rucksack",
|
|
"objekttyp_id": rucksack_typ.id,
|
|
"vorlage_id": vorlage.id,
|
|
"standort_id": standort.id,
|
|
},
|
|
headers=auth_header(token),
|
|
)
|
|
zweiter_rucksack_id = zweiter_rucksack.json()["id"]
|
|
|
|
abgelehnt_patch = await client.patch(
|
|
f"/api/v1/objekte/{zweiter_rucksack_id}",
|
|
json={"fahrzeug_id": rucksack_id},
|
|
headers=auth_header(token),
|
|
)
|
|
assert abgelehnt_patch.status_code == 422
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_objekt_code_neu_vergeben_mit_historie(client, objekt_mit_position, admin_user):
|
|
"""IDENT-009: neuer Code wird gesetzt, alter Code bleibt über die
|
|
Historie nachvollziehbar."""
|
|
objekt, _material = objekt_mit_position
|
|
token = await login(client, "admin1")
|
|
alter_code = objekt.code
|
|
|
|
neu = await client.post(
|
|
f"/api/v1/objekte/{objekt.id}/code-neu-vergeben",
|
|
json={"neuer_code": "KT-NEU", "begruendung": "Etikett beschädigt"},
|
|
headers=auth_header(token),
|
|
)
|
|
assert neu.status_code == 200
|
|
assert neu.json()["code"] == "KT-NEU"
|
|
|
|
historie = await client.get(
|
|
f"/api/v1/historie?entitaet_typ=objekt&entitaet_id={objekt.id}", headers=auth_header(token)
|
|
)
|
|
eintraege = [h for h in historie.json() if h["ereignistyp"] == "objekt_code_neu_vergeben"]
|
|
assert len(eintraege) == 1
|
|
assert eintraege[0]["alter_wert"] == {"code": alter_code}
|
|
assert eintraege[0]["neuer_wert"] == {"code": "KT-NEU"}
|
|
assert eintraege[0]["begruendung"] == "Etikett beschädigt"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_objekt_code_neu_vergeben_duplikat_abgelehnt(
|
|
client, db_session, admin_user, standort_factory, objekttyp_factory, hauptserver_id
|
|
):
|
|
"""Neuer Code darf nicht mit dem Code eines ANDEREN Objekts kollidieren."""
|
|
from app.models.objekt import Objekt
|
|
|
|
objekttyp = await objekttyp_factory("Typ-CodeDupe")
|
|
standort = await standort_factory("Standort-CodeDupe")
|
|
|
|
erstes = Objekt(
|
|
code="KT-A",
|
|
name="Objekt A",
|
|
objekttyp_id=objekttyp.id,
|
|
standort_id=standort.id,
|
|
zustaendiger_server_id=hauptserver_id,
|
|
)
|
|
zweites = Objekt(
|
|
code="KT-B",
|
|
name="Objekt B",
|
|
objekttyp_id=objekttyp.id,
|
|
standort_id=standort.id,
|
|
zustaendiger_server_id=hauptserver_id,
|
|
)
|
|
db_session.add_all([erstes, zweites])
|
|
await db_session.flush()
|
|
|
|
token = await login(client, "admin1")
|
|
konflikt = await client.post(
|
|
f"/api/v1/objekte/{zweites.id}/code-neu-vergeben",
|
|
json={"neuer_code": "KT-A"},
|
|
headers=auth_header(token),
|
|
)
|
|
assert konflikt.status_code == 409
|
|
|
|
eigener_code_ist_ok = await client.post(
|
|
f"/api/v1/objekte/{zweites.id}/code-neu-vergeben",
|
|
json={"neuer_code": "KT-B"},
|
|
headers=auth_header(token),
|
|
)
|
|
assert eigener_code_ist_ok.status_code == 200
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_mitarbeiter_darf_code_nicht_neu_vergeben(client, objekt_mit_position, mitarbeiter_user):
|
|
objekt, _material = objekt_mit_position
|
|
token = await login(client, "mitarbeiter1")
|
|
response = await client.post(
|
|
f"/api/v1/objekte/{objekt.id}/code-neu-vergeben",
|
|
json={"neuer_code": "KT-XYZ"},
|
|
headers=auth_header(token),
|
|
)
|
|
assert response.status_code == 403
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_objektposition_code_neu_vergeben(client, db_session, objekt_mit_position, admin_user):
|
|
objekt, material = objekt_mit_position
|
|
token = await login(client, "admin1")
|
|
|
|
positionen = await client.get(f"/api/v1/objekte/{objekt.id}/positionen", headers=auth_header(token))
|
|
position_id = positionen.json()[0]["id"]
|
|
await client.patch(
|
|
f"/api/v1/objekte/{objekt.id}/positionen/{position_id}",
|
|
json={"code": "GR-ALT"},
|
|
headers=auth_header(token),
|
|
)
|
|
|
|
neu = await client.post(
|
|
f"/api/v1/objektpositionen/{position_id}/code-neu-vergeben",
|
|
json={"neuer_code": "GR-NEU", "begruendung": "Etikett verloren"},
|
|
headers=auth_header(token),
|
|
)
|
|
assert neu.status_code == 200
|
|
assert neu.json()["code"] == "GR-NEU"
|
|
|
|
historie = await client.get(
|
|
f"/api/v1/historie?entitaet_typ=objektposition&entitaet_id={position_id}", headers=auth_header(token)
|
|
)
|
|
eintraege = [h for h in historie.json() if h["ereignistyp"] == "objektposition_code_neu_vergeben"]
|
|
assert len(eintraege) == 1
|
|
assert eintraege[0]["alter_wert"] == {"code": "GR-ALT"}
|