diff --git a/DEVLOG.md b/DEVLOG.md index a83967a..93e272e 100644 --- a/DEVLOG.md +++ b/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 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++ + +--- diff --git a/arbeitskacheln/17_zustaendigkeit.md b/arbeitskacheln/17_zustaendigkeit.md index caddf6d..e7053eb 100644 --- a/arbeitskacheln/17_zustaendigkeit.md +++ b/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`. --- diff --git a/backend/app/api/v1/endpoints/objekte.py b/backend/app/api/v1/endpoints/objekte.py index 2645cb2..bc0065f 100644 --- a/backend/app/api/v1/endpoints/objekte.py +++ b/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) diff --git a/backend/app/api/v1/endpoints/verleih.py b/backend/app/api/v1/endpoints/verleih.py index 44f6a8d..cd4a6d6 100644 --- a/backend/app/api/v1/endpoints/verleih.py +++ b/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) diff --git a/backend/app/api/v1/endpoints/wartung.py b/backend/app/api/v1/endpoints/wartung.py index 59eed57..952b6a4 100644 --- a/backend/app/api/v1/endpoints/wartung.py +++ b/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) diff --git a/backend/tests/test_zust004_standort_sichtbarkeit.py b/backend/tests/test_zust004_standort_sichtbarkeit.py index 3078eaf..69f0c3a 100644 --- a/backend/tests/test_zust004_standort_sichtbarkeit.py +++ b/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")