Sprint 1: Stammdaten-Modelle/-Endpunkte, Admin-Benutzerverwaltung, Zuständigkeits-Vererbung (E2)

- ORM-Modelle: Bereich, Kategorie, Standort, Objekttyp, Material (Prompt 06/07),
  minimales Objekt-Modell (voll ausgebaut erst Sprint 2), Zustaendigkeit/
  Kontrollverantwortung
- Admin-API 7a: GET/POST Stammdaten-Endpunkte, PATCH Material, Benutzerverwaltung
  (anlegen/Rollen ändern), Zuständigkeits-/Kontrollverantwortungs-CRUD
- Vererbungslogik E2 als eigener Service (app/services/zustaendigkeit.py):
  Standort-Zuordnung vererbt sich auf alle Objekte, Objekt-Zeile ist Vereinigung
  statt Ersatz - GET /zustaendigkeiten/benutzer/{id}/objekte exponiert das
- Tests: Stammdaten-CRUD + Rollenrechte, Benutzerverwaltung (inkl. 409 bei
  Login-Duplikat), E2-Vererbungslogik (Standort-only, Standort+Objekt-Vereinigung,
  End-to-End über den Endpunkt)

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01L85hmKbvX7Cqkq47KnQhFt

backend/tests/conftest.py

@@ -8,6 +8,8 @@ from app.core.security import hash_password
 from app.db.session import get_db
 from app.main import app
 from app.models.auth import Benutzer, BenutzerRolle, RolleTyp
+from app.models.objekt import Objekt
+from app.models.stammdaten import Bereich, Objekttyp, Standort
 
 # Erwartet eine bereits per Alembic migrierte Test-Datenbank (CI: install -> migrate -> pytest,
 # siehe testphasen.md Phase 0). Jeder Test läuft in einer Transaktion, die am Ende zurückgerollt
@@ -80,3 +82,62 @@ async def admin_user(db_session):
     db_session.add(BenutzerRolle(benutzer_id=benutzer.id, rolle=RolleTyp.administration))
     await db_session.flush()
     return benutzer
+
+
+async def login(client, username: str, password: str = "test-passwort-123") -> str:
+    response = await client.post(
+        "/api/v1/auth/login", data={"username": username, "password": password}
+    )
+    assert response.status_code == 200, response.text
+    return response.json()["access_token"]
+
+
+def auth_header(token: str) -> dict:
+    return {"Authorization": f"Bearer {token}"}
+
+
+@pytest_asyncio.fixture
+async def hauptserver_id(db_session):
+    # Seed-Migration 0002 legt genau einen Datensatz mit typ='haupt' an (Sprintplan E6).
+    from sqlalchemy import select
+
+    from app.models.auth import KnotenTyp, Systemknoten
+
+    result = await db_session.execute(
+        select(Systemknoten.id).where(Systemknoten.typ == KnotenTyp.haupt)
+    )
+    return result.scalar_one()
+
+
+@pytest_asyncio.fixture
+async def standort_factory(db_session):
+    async def _make(name: str) -> Standort:
+        standort = Standort(name=name)
+        db_session.add(standort)
+        await db_session.flush()
+        return standort
+
+    return _make
+
+
+@pytest_asyncio.fixture
+async def objekt_factory(db_session, hauptserver_id):
+    async def _make(*, name: str, code: str, standort: Standort) -> Objekt:
+        bereich = Bereich(name=f"Bereich-{code}")
+        db_session.add(bereich)
+        await db_session.flush()
+        objekttyp = Objekttyp(bereich_id=bereich.id, name=f"Typ-{code}")
+        db_session.add(objekttyp)
+        await db_session.flush()
+        objekt = Objekt(
+            code=code,
+            name=name,
+            objekttyp_id=objekttyp.id,
+            standort_id=standort.id,
+            zustaendiger_server_id=hauptserver_id,
+        )
+        db_session.add(objekt)
+        await db_session.flush()
+        return objekt
+
+    return _make

backend/tests/test_benutzer_admin.py

@@ -0,0 +1,56 @@
+import pytest
+
+from tests.conftest import auth_header, login
+
+
+@pytest.mark.asyncio
+async def test_admin_can_create_benutzer_with_rollen(client, admin_user):
+    token = await login(client, "admin1")
+    response = await client.post(
+        "/api/v1/benutzer",
+        json={
+            "name": "Neue Kollegin",
+            "login": "neu1",
+            "passwort": "sicheres-passwort",
+            "rollen": ["mitarbeiter", "materialverantwortlicher"],
+        },
+        headers=auth_header(token),
+    )
+    assert response.status_code == 201
+    body = response.json()
+    assert body["login"] == "neu1"
+    assert set(body["rollen"]) == {"mitarbeiter", "materialverantwortlicher"}
+
+
+@pytest.mark.asyncio
+async def test_duplicate_login_conflict(client, admin_user):
+    token = await login(client, "admin1")
+    payload = {
+        "name": "Doppelt",
+        "login": "doppelt1",
+        "passwort": "sicheres-passwort",
+        "rollen": [],
+    }
+    first = await client.post("/api/v1/benutzer", json=payload, headers=auth_header(token))
+    assert first.status_code == 201
+    second = await client.post("/api/v1/benutzer", json=payload, headers=auth_header(token))
+    assert second.status_code == 409
+
+
+@pytest.mark.asyncio
+async def test_mitarbeiter_cannot_manage_benutzer(client, mitarbeiter_user):
+    token = await login(client, "mitarbeiter1")
+    response = await client.get("/api/v1/benutzer", headers=auth_header(token))
+    assert response.status_code == 403
+
+
+@pytest.mark.asyncio
+async def test_admin_can_update_rollen(client, admin_user, mitarbeiter_user):
+    token = await login(client, "admin1")
+    response = await client.patch(
+        f"/api/v1/benutzer/{mitarbeiter_user.id}",
+        json={"rollen": ["leitungsverantwortlicher"]},
+        headers=auth_header(token),
+    )
+    assert response.status_code == 200
+    assert response.json()["rollen"] == ["leitungsverantwortlicher"]

backend/tests/test_stammdaten.py

@@ -0,0 +1,62 @@
+import pytest
+
+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

backend/tests/test_zustaendigkeit.py

@@ -0,0 +1,80 @@
+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