patrickandClaude Sonnet 5
c1f6eaa14c
Backend: DELETE /kontrollverantwortung/{id} ergänzt (fehlte bisher). Frontend:
neuer Admin-Tab "Kontrollverantwortung" - Objekt + Benutzer/Gruppe zuordnen,
Liste, Entfernen. Reine Sichtbarkeit/Priorisierung, keine Pflichtkontrolle.
Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01CVgbozhYmuEhiEJHffRXCV
126 lines
4.7 KiB
Python
126 lines
4.7 KiB
Python
import pytest
|
|
|
|
from app.models.zustaendigkeit import Zustaendigkeit
|
|
from app.services.zustaendigkeit import zustaendige_objekte
|
|
from tests.conftest import auth_header, login
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_standort_zuordnung_vererbt_sich_auf_alle_objekte(
|
|
db_session, standort_factory, objekt_factory, mitarbeiter_user
|
|
):
|
|
"""Sprintplan E2: Standort-Zuordnung -> Sicht enthält ALLE Objekte des Standorts."""
|
|
standort_a = await standort_factory("Wache A")
|
|
standort_b = await standort_factory("Wache B")
|
|
|
|
objekt_a1 = await objekt_factory(name="Rucksack A1", code="A1", standort=standort_a)
|
|
objekt_a2 = await objekt_factory(name="Rucksack A2", code="A2", standort=standort_a)
|
|
objekt_b1 = await objekt_factory(name="Rucksack B1", code="B1", standort=standort_b)
|
|
|
|
db_session.add(Zustaendigkeit(benutzer_id=mitarbeiter_user.id, standort_id=standort_a.id))
|
|
await db_session.flush()
|
|
|
|
objekte = await zustaendige_objekte(db_session, mitarbeiter_user.id)
|
|
objekt_ids = {o.id for o in objekte}
|
|
|
|
assert objekt_a1.id in objekt_ids
|
|
assert objekt_a2.id in objekt_ids
|
|
assert objekt_b1.id not in objekt_ids
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_objekt_zuordnung_ist_vereinigung_nicht_ersatz(
|
|
db_session, standort_factory, objekt_factory, mitarbeiter_user
|
|
):
|
|
"""Objektspezifische Zuordnung ergänzt die Standort-Vererbung (Vereinigung)."""
|
|
standort_a = await standort_factory("Wache A")
|
|
standort_b = await standort_factory("Wache B")
|
|
|
|
objekt_a1 = await objekt_factory(name="Rucksack A1", code="A1", standort=standort_a)
|
|
objekt_b1 = await objekt_factory(name="Rucksack B1", code="B1", standort=standort_b)
|
|
|
|
db_session.add(Zustaendigkeit(benutzer_id=mitarbeiter_user.id, standort_id=standort_a.id))
|
|
db_session.add(Zustaendigkeit(benutzer_id=mitarbeiter_user.id, objekt_id=objekt_b1.id))
|
|
await db_session.flush()
|
|
|
|
objekte = await zustaendige_objekte(db_session, mitarbeiter_user.id)
|
|
objekt_ids = {o.id for o in objekte}
|
|
|
|
assert objekt_ids == {objekt_a1.id, objekt_b1.id}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_endpoint_liefert_vererbte_objekte(
|
|
client, db_session, standort_factory, objekt_factory, admin_user, mitarbeiter_user
|
|
):
|
|
standort_a = await standort_factory("Wache A")
|
|
objekt_a1 = await objekt_factory(name="Rucksack A1", code="A1", standort=standort_a)
|
|
|
|
db_session.add(Zustaendigkeit(benutzer_id=mitarbeiter_user.id, standort_id=standort_a.id))
|
|
await db_session.flush()
|
|
|
|
token = await login(client, "admin1")
|
|
response = await client.get(
|
|
f"/api/v1/zustaendigkeiten/benutzer/{mitarbeiter_user.id}/objekte",
|
|
headers=auth_header(token),
|
|
)
|
|
assert response.status_code == 200
|
|
ids = {o["id"] for o in response.json()}
|
|
assert objekt_a1.id in ids
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_zustaendigkeit_braucht_standort_oder_objekt(client, admin_user, mitarbeiter_user):
|
|
token = await login(client, "admin1")
|
|
response = await client.post(
|
|
"/api/v1/zustaendigkeiten",
|
|
json={"benutzer_id": mitarbeiter_user.id},
|
|
headers=auth_header(token),
|
|
)
|
|
assert response.status_code == 422
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_kontrollverantwortung_anlegen_listen_loeschen(
|
|
client, standort_factory, objekt_factory, admin_user, mitarbeiter_user
|
|
):
|
|
standort = await standort_factory("Wache K")
|
|
objekt = await objekt_factory(name="Rucksack K", code="K1", standort=standort)
|
|
token = await login(client, "admin1")
|
|
|
|
erstellt = await client.post(
|
|
"/api/v1/kontrollverantwortung",
|
|
json={"objekt_id": objekt.id, "benutzer_id": mitarbeiter_user.id},
|
|
headers=auth_header(token),
|
|
)
|
|
assert erstellt.status_code == 201
|
|
eintrag_id = erstellt.json()["id"]
|
|
|
|
liste = await client.get("/api/v1/kontrollverantwortung", headers=auth_header(token))
|
|
assert liste.status_code == 200
|
|
assert any(e["id"] == eintrag_id for e in liste.json())
|
|
|
|
geloescht = await client.delete(
|
|
f"/api/v1/kontrollverantwortung/{eintrag_id}", headers=auth_header(token)
|
|
)
|
|
assert geloescht.status_code == 204
|
|
|
|
liste_danach = await client.get("/api/v1/kontrollverantwortung", headers=auth_header(token))
|
|
assert all(e["id"] != eintrag_id for e in liste_danach.json())
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_kontrollverantwortung_braucht_benutzer_oder_gruppe(
|
|
client, standort_factory, objekt_factory, admin_user
|
|
):
|
|
standort = await standort_factory("Wache L")
|
|
objekt = await objekt_factory(name="Rucksack L", code="L1", standort=standort)
|
|
token = await login(client, "admin1")
|
|
|
|
response = await client.post(
|
|
"/api/v1/kontrollverantwortung",
|
|
json={"objekt_id": objekt.id},
|
|
headers=auth_header(token),
|
|
)
|
|
assert response.status_code == 422
|