feat(zustaendigkeit): ZUST-004 Runde 3 - Wartung/Verleih/Zählerstand

GET /wartungsauftraege, GET /verleihe(+uebergabeliste), PATCH /objekte/{id}/
zaehlerstand und GET /objekte/{id}/uebergabeliste (Aussonderung) bekommen
dieselbe Standort-Sichtbarkeitsprüfung. Lagerbewegungen/Lagerplätze bleiben
bewusst blockiert - Lagerort hat keine FK auf die echte Standort-Tabelle
(eigene, unabhängige Hierarchie), braucht erst eine eigene Entscheidung
bevor ZUST-004 dort anwendbar ist.

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

DEVLOG.md

@@ -15594,3 +15594,18 @@ Keine Commits in dieser Session.
 - arbeitskacheln/12_personnel.md | 12 ++++--------
 
 ---
+## 2026-09-13 22:41 – 22:44 (2m)
+**Beschreibung:** Claude Code Session
+**Projekt:** asb-material
+
+### Commits
+- 56fae09 feat(zustaendigkeit): ZUST-004 Runde 2 - Kontrollen und Mängel eingeschränkt
+
+### Geänderte Dateien
+- DEVLOG.md                                           | 39 +++++++++++++++++++++++++++++++++++++++
+- arbeitskacheln/17_zustaendigkeit.md                 | 30 +++++++++++++++++-------------
+- backend/app/api/v1/endpoints/kontrollen.py          |  6 +++++-
+- backend/app/api/v1/endpoints/mangel.py              |  9 +++++++--
+- backend/tests/test_zust004_standort_sichtbarkeit.py | 57 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++
+
+---

arbeitskacheln/17_zustaendigkeit.md

@@ -188,15 +188,26 @@ war die Vorgängerlücke, diese hier ist die zweite gefundene Lücke).
   (schließt damit den ursprünglichen DOC-006-Befund tatsächlich). Umgesetzt
   (Runde 2): `GET /maengel` (Standort-Filter über Join auf `Objekt`),
   `GET /kontrollen/{id}` (Standort-Prüfung über `Kontrolle.objekt_id`).
-  **Noch NICHT eingeschränkt (Follow-up bei Bedarf, nach demselben
-  Muster):** Historie, Dashboard (Einsatzbereitschaft/Prüftermine),
-  Wartungsaufträge, Lagerbewegungen/Lagerplätze, Verleih/Übergabeliste/
-  Zählerstand-Endpunkte aus ASSET-013/009. Wer diese Lücken ausnutzt, sieht
+  Umgesetzt (Runde 3): `GET /wartungsauftraege`, `GET /verleihe`,
+  `GET /verleihe/{id}/uebergabeliste`, `PATCH /objekte/{id}/zaehlerstand`,
+  `GET /objekte/{id}/uebergabeliste` (Aussonderung).
+  **Blockiert, braucht erst eine eigene Entscheidung:** Lagerbewegungen/
+  Lagerplätze - `Lagerort` hat KEINE FK auf die echte `Standort`-Tabelle
+  (eigene, unabhängige Hierarchie mit `standort` nur als Typ-Label, siehe
+  Docstring `app/models/lager.py:Lagerort` - "Objekt/Zustaendigkeit bleiben
+  unverändert an `standort` hängen, Lagerort ist ein zusätzliches, feineres
+  Konzept"). ZUST-004 lässt sich hier nicht 1:1 anwenden, ohne vorher zu
+  klären, wie ein `Lagerort`-Knoten einem echten `Standort` zugeordnet wird.
+  **Noch NICHT eingeschränkt (Follow-up bei Bedarf):** Historie, Dashboard
+  (Einsatzbereitschaft/Prüftermine). Wer diese Lücken ausnutzt, sieht
   weiterhin standortübergreifend - kein vollständiger Schutz, bis diese
   Runden nachgezogen sind. Tests:
   `test_mangel_liste_standort_eingeschraenkt`,
-  `test_kontroll_fortschritt_standort_eingeschraenkt` in
-  `backend/tests/test_zust004_standort_sichtbarkeit.py`.
+  `test_kontroll_fortschritt_standort_eingeschraenkt`,
+  `test_wartungsauftraege_liste_standort_eingeschraenkt`,
+  `test_verleih_liste_und_uebergabeliste_standort_eingeschraenkt`,
+  `test_objekt_zaehlerstand_und_aussonderung_uebergabeliste_standort_eingeschraenkt`
+  in `backend/tests/test_zust004_standort_sichtbarkeit.py`.
 
 ---
 

backend/app/api/v1/endpoints/objekte.py

@@ -597,6 +597,7 @@ async def aendere_zaehlerstand(
     objekt = await db.get(Objekt, objekt_id)
     if objekt is None:
         raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Objekt nicht gefunden")
+    await pruefe_standort_sichtbar(db, benutzer_id=current_user.id, standort_id=objekt.standort_id)
 
     ist_admin = RolleTyp.administration in {RolleTyp(r) for r in current_user.rollen_namen}
     vorher = await db.get(Fahrzeugdetails, objekt_id)
@@ -634,7 +635,7 @@ async def aendere_zaehlerstand(
 
 @router.get("/objekte/{objekt_id}/uebergabeliste")
 async def objekt_uebergabeliste(
-    objekt_id: int, db: AsyncSession = Depends(get_db), _=Depends(_materialverantwortliche)
+    objekt_id: int, db: AsyncSession = Depends(get_db), current_user=Depends(_materialverantwortliche)
 ) -> Response:
     """ASSET-013 (Aussonderungs-Teil): reines Druck-PDF zum handschriftlichen
     Unterschreiben bei Aussonderung, gleicher Mechanismus wie
@@ -644,6 +645,7 @@ async def objekt_uebergabeliste(
     objekt = await db.get(Objekt, objekt_id)
     if objekt is None:
         raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Objekt nicht gefunden")
+    await pruefe_standort_sichtbar(db, benutzer_id=current_user.id, standort_id=objekt.standort_id)
 
     standort_name = await hole_standort_name(db, objekt.standort_id)
     positionen = await hole_uebergabepositionen(db, objekt)

backend/app/api/v1/endpoints/verleih.py

@@ -2,6 +2,7 @@ import uuid
 from datetime import date, datetime
 
 from fastapi import APIRouter, Depends, HTTPException, Response, status
+from sqlalchemy import select
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from app.api.deps import require_roles
@@ -10,6 +11,7 @@ from app.models.auth import Benutzer, RolleTyp
 from app.models.objekt import Objekt
 from app.models.verleih import ExterneStelle, Verleih
 from app.schemas.verleih import ExterneStelleCreate, ExterneStelleRead, VerleihCreate, VerleihRead
+from app.services.standort_sichtbarkeit import pruefe_standort_sichtbar, sichtbare_standort_ids
 from app.services.uebergabeliste import (
     generiere_uebergabeliste_pdf,
     hole_standort_name,
@@ -64,9 +66,17 @@ async def erstelle_externe_stelle_endpunkt(
 
 @router.get("/verleihe", response_model=list[VerleihRead])
 async def hole_verleihe(
-    offen: bool = False, db: AsyncSession = Depends(get_db), _=Depends(_materialverantwortliche)
+    offen: bool = False, db: AsyncSession = Depends(get_db), current_user=Depends(_materialverantwortliche)
 ) -> list[VerleihRead]:
-    return [_verleih_read(v) for v in await liste_verleihe(db, nur_offene=offen)]
+    """ZUST-004-Erweiterung: Standort-Sichtbarkeitsgrenze wie bei Objekten."""
+    verleihe = await liste_verleihe(db, nur_offene=offen)
+    sichtbare = await sichtbare_standort_ids(db, current_user.id)
+    if sichtbare is None:
+        return [_verleih_read(v) for v in verleihe]
+    objekt_ids = {v.objekt_id for v in verleihe}
+    result = await db.execute(select(Objekt.id).where(Objekt.id.in_(objekt_ids), Objekt.standort_id.in_(sichtbare)))
+    erlaubte_objekt_ids = {row[0] for row in result.all()}
+    return [_verleih_read(v) for v in verleihe if v.objekt_id in erlaubte_objekt_ids]
 
 
 @router.post("/verleihe", response_model=VerleihRead, status_code=status.HTTP_201_CREATED)
@@ -111,7 +121,7 @@ async def verleih_rueckgabe(
 
 @router.get("/verleihe/{verleih_id}/uebergabeliste")
 async def verleih_uebergabeliste(
-    verleih_id: uuid.UUID, db: AsyncSession = Depends(get_db), _=Depends(_materialverantwortliche)
+    verleih_id: uuid.UUID, db: AsyncSession = Depends(get_db), current_user=Depends(_materialverantwortliche)
 ) -> Response:
     """ASSET-013 (Verleih-Teil): reines Druck-PDF zum handschriftlichen
     Unterschreiben, siehe app/services/uebergabeliste.py."""
@@ -121,6 +131,7 @@ async def verleih_uebergabeliste(
     objekt = await db.get(Objekt, verleih.objekt_id)
     if objekt is None:
         raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Objekt nicht gefunden")
+    await pruefe_standort_sichtbar(db, benutzer_id=current_user.id, standort_id=objekt.standort_id)
     externe_stelle = await db.get(ExterneStelle, verleih.externe_stelle_id)
 
     standort_name = await hole_standort_name(db, objekt.standort_id)

backend/app/api/v1/endpoints/wartung.py

@@ -35,6 +35,7 @@ from app.services.wartung import (
     liste_wartungsauftraege,
     liste_wartungsauftrag_teile,
 )
+from app.services.standort_sichtbarkeit import sichtbare_standort_ids
 
 router = APIRouter()
 
@@ -100,9 +101,19 @@ async def liste_auftraege(
     objekt_id: int | None = None,
     status_filter: WartungsauftragStatus | None = None,
     db: AsyncSession = Depends(get_db),
-    _=Depends(get_current_user),
+    current_user=Depends(get_current_user),
 ) -> list[Wartungsauftrag]:
-    return await liste_wartungsauftraege(db, objekt_id=objekt_id, status=status_filter)
+    """ZUST-004-Erweiterung: Standort-Sichtbarkeitsgrenze wie bei Objekten."""
+    auftraege = await liste_wartungsauftraege(db, objekt_id=objekt_id, status=status_filter)
+    sichtbare = await sichtbare_standort_ids(db, current_user.id)
+    if sichtbare is None:
+        return auftraege
+    objekt_ids = {a.objekt_id for a in auftraege}
+    if not objekt_ids:
+        return auftraege
+    result = await db.execute(select(Objekt.id).where(Objekt.id.in_(objekt_ids), Objekt.standort_id.in_(sichtbare)))
+    erlaubte_objekt_ids = {row[0] for row in result.all()}
+    return [a for a in auftraege if a.objekt_id in erlaubte_objekt_ids]
 
 
 @router.post("/wartungsauftraege", response_model=WartungsauftragRead, status_code=status.HTTP_201_CREATED)

backend/tests/test_zust004_standort_sichtbarkeit.py

@@ -201,6 +201,96 @@ async def test_kontroll_fortschritt_standort_eingeschraenkt(
     assert response.status_code == 403
 
 
+@pytest.mark.asyncio
+async def test_wartungsauftraege_liste_standort_eingeschraenkt(
+    client, admin_user, materialverantwortlicher_user, standort_factory, objekt_factory
+):
+    standort_a = await standort_factory("Standort A7")
+    standort_b = await standort_factory("Standort B7")
+    objekt_a = await objekt_factory(name="Objekt A7", code="ZUST4-A7", standort=standort_a)
+    objekt_b = await objekt_factory(name="Objekt B7", code="ZUST4-B7", standort=standort_b)
+
+    admin_token = await login(client, "admin1")
+    for objekt in (objekt_a, objekt_b):
+        await client.post(
+            "/api/v1/wartungsauftraege",
+            json={"objekt_id": objekt.id, "wartungsart": "Sichtprüfung", "werkstatt_intern": True},
+            headers=auth_header(admin_token),
+        )
+    await client.put(
+        f"/api/v1/benutzer/{materialverantwortlicher_user.id}/standorte",
+        json={"standort_ids": [standort_a.id]},
+        headers=auth_header(admin_token),
+    )
+
+    token = await login(client, "materialverantwortlicher1")
+    liste = await client.get("/api/v1/wartungsauftraege", headers=auth_header(token))
+    objekt_ids = {a["objekt_id"] for a in liste.json()}
+    assert objekt_a.id in objekt_ids
+    assert objekt_b.id not in objekt_ids
+
+
+@pytest.mark.asyncio
+async def test_verleih_liste_und_uebergabeliste_standort_eingeschraenkt(
+    client, admin_user, materialverantwortlicher_user, standort_factory, objekt_factory
+):
+    standort_a = await standort_factory("Standort A8")
+    standort_b = await standort_factory("Standort B8")
+    objekt_b = await objekt_factory(name="Objekt B8", code="ZUST4-B8", standort=standort_b)
+
+    admin_token = await login(client, "admin1")
+    stelle = await client.post("/api/v1/externe-stellen", json={"name": "Verein Z"}, headers=auth_header(admin_token))
+    verliehen = await client.post(
+        "/api/v1/verleihe",
+        json={
+            "objekt_id": objekt_b.id,
+            "externe_stelle_id": stelle.json()["id"],
+            "verliehen_am": "2026-01-01",
+        },
+        headers=auth_header(admin_token),
+    )
+    verleih_id = verliehen.json()["id"]
+
+    await client.put(
+        f"/api/v1/benutzer/{materialverantwortlicher_user.id}/standorte",
+        json={"standort_ids": [standort_a.id]},
+        headers=auth_header(admin_token),
+    )
+    token = await login(client, "materialverantwortlicher1")
+
+    liste = await client.get("/api/v1/verleihe", headers=auth_header(token))
+    assert all(v["id"] != verleih_id for v in liste.json())
+
+    uebergabeliste = await client.get(f"/api/v1/verleihe/{verleih_id}/uebergabeliste", headers=auth_header(token))
+    assert uebergabeliste.status_code == 403
+
+
+@pytest.mark.asyncio
+async def test_objekt_zaehlerstand_und_aussonderung_uebergabeliste_standort_eingeschraenkt(
+    client, admin_user, materialverantwortlicher_user, standort_factory, objekt_factory
+):
+    standort_a = await standort_factory("Standort A9")
+    standort_b = await standort_factory("Standort B9")
+    objekt_b = await objekt_factory(name="Objekt B9", code="ZUST4-B9", standort=standort_b)
+
+    admin_token = await login(client, "admin1")
+    await client.put(
+        f"/api/v1/benutzer/{materialverantwortlicher_user.id}/standorte",
+        json={"standort_ids": [standort_a.id]},
+        headers=auth_header(admin_token),
+    )
+    token = await login(client, "materialverantwortlicher1")
+
+    zaehlerstand = await client.patch(
+        f"/api/v1/objekte/{objekt_b.id}/zaehlerstand", json={"kilometerstand": 100}, headers=auth_header(token)
+    )
+    assert zaehlerstand.status_code == 403
+
+    await client.patch(f"/api/v1/objekte/{objekt_b.id}", json={"status": "ausgesondert"}, headers=auth_header(admin_token))
+    uebergabeliste = await client.get(f"/api/v1/objekte/{objekt_b.id}/uebergabeliste", headers=auth_header(token))
+    assert uebergabeliste.status_code == 403
+
+
 @pytest.mark.asyncio
 async def test_standort_zugriff_setzen_nur_admin(client, mitarbeiter_user, mitarbeiter2_user):
     token = await login(client, "mitarbeiter1")