feat(lager): Lagerbewegung-Protokoll (Roadmap Phase 4, Modul Lager)

Objekt bleibt der Materialträger (Rucksack/Gerät/Fahrzeug, Karte 11) -
"Material verschieben" heißt: das Objekt bekommt einen neuen Standort, jede
Bewegung wird append-only protokolliert (Wer/Wann/Von/Nach/Grund, Nutzer-
Vorgabe Modul 5).

POST /objekte/{id}/lagerbewegungen setzt Standort + protokolliert in einem
Zug, GET /lagerbewegungen (Filter objekt_id/standort_id) für die Historie.
Rollen wie beim Mangel-Modul: Materialverantwortliche+Leitung+Admin dürfen
verschieben, alle Eingeloggten dürfen lesen.

Frontend: "Verschieben"-Aktion in ObjektSection, neuer Admin-Tab
"Lagerbewegungen" (reine Anzeige, analog HistorieSection).

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

backend/alembic/versions/0015_lagerbewegung.py

@@ -0,0 +1,40 @@
+"""Lagerbewegung (Roadmap Phase 4, Modul Lager): protokolliert Standortwechsel
+eines Objekts. Objekt bleibt der Materialträger (Rucksack/Gerät/Fahrzeug,
+Karte 11) - "Material verschieben" heißt hier: das Objekt, das es enthält,
+bekommt einen neuen Standort, die Bewegung wird mit Wer/Wann/Von/Nach/Grund
+protokolliert (Nutzer-Vorgabe Modul 5 "Lagerverwaltung").
+
+Revision ID: 0015_lagerbewegung
+Revises: 0014_fahrzeugdetails
+Create Date: 2026-09-05
+"""
+from typing import Sequence, Union
+
+from alembic import op
+
+revision: str = "0015_lagerbewegung"
+down_revision: Union[str, None] = "0014_fahrzeugdetails"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    op.execute(
+        """
+        CREATE TABLE lagerbewegung (
+            id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
+            objekt_id INTEGER NOT NULL REFERENCES objekt(id),
+            von_standort_id INTEGER REFERENCES standort(id),
+            nach_standort_id INTEGER NOT NULL REFERENCES standort(id),
+            benutzer_id INTEGER NOT NULL REFERENCES benutzer(id),
+            zeitpunkt TIMESTAMPTZ NOT NULL,
+            grund TEXT
+        )
+        """
+    )
+    op.execute("CREATE INDEX idx_lagerbewegung_objekt ON lagerbewegung (objekt_id)")
+    op.execute("CREATE INDEX idx_lagerbewegung_standorte ON lagerbewegung (von_standort_id, nach_standort_id)")
+
+
+def downgrade() -> None:
+    op.execute("DROP TABLE lagerbewegung")

backend/app/api/v1/api.py

@@ -10,6 +10,7 @@ from app.api.v1.endpoints import (
     health,
     historie,
     kontrollen,
+    lagerbewegung,
     mangel,
     objekte,
     personal,
@@ -34,3 +35,4 @@ api_router.include_router(geraet_instanz.router, tags=["geraet_instanz"])
 api_router.include_router(dashboard.router, tags=["dashboard"])
 api_router.include_router(personal.router, tags=["personal"])
 api_router.include_router(mangel.router, tags=["mangel"])
+api_router.include_router(lagerbewegung.router, tags=["lagerbewegung"])

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

@@ -0,0 +1,68 @@
+from fastapi import APIRouter, Depends, HTTPException, status
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.api.deps import get_current_user, require_roles
+from app.db.session import get_db
+from app.models.auth import RolleTyp
+from app.models.lagerbewegung import Lagerbewegung
+from app.models.objekt import Objekt
+from app.models.stammdaten import Standort
+from app.schemas.lagerbewegung import LagerbewegungCreate, LagerbewegungRead
+from app.services.lagerbewegung import verschiebe_objekt
+
+router = APIRouter()
+
+# Deckt sich mit der Rollen-Vorgabe aus dem Mangel-Modul (Materialwart-
+# Berechtigung "Lagerbewegungen durchführen" aus der Ursprungs-Anforderung).
+_materialverantwortliche = require_roles(
+    RolleTyp.administration, RolleTyp.materialverantwortlicher, RolleTyp.leitungsverantwortlicher
+)
+
+
+@router.get("/lagerbewegungen", response_model=list[LagerbewegungRead])
+async def liste_lagerbewegungen(
+    objekt_id: int | None = None,
+    standort_id: int | None = None,
+    db: AsyncSession = Depends(get_db),
+    _=Depends(get_current_user),
+) -> list[Lagerbewegung]:
+    stmt = select(Lagerbewegung)
+    if objekt_id is not None:
+        stmt = stmt.where(Lagerbewegung.objekt_id == objekt_id)
+    if standort_id is not None:
+        stmt = stmt.where(
+            (Lagerbewegung.von_standort_id == standort_id) | (Lagerbewegung.nach_standort_id == standort_id)
+        )
+    stmt = stmt.order_by(Lagerbewegung.zeitpunkt.desc())
+    result = await db.execute(stmt)
+    return list(result.scalars().all())
+
+
+@router.post(
+    "/objekte/{objekt_id}/lagerbewegungen",
+    response_model=LagerbewegungRead,
+    status_code=status.HTTP_201_CREATED,
+)
+async def verschiebe(
+    objekt_id: int,
+    payload: LagerbewegungCreate,
+    db: AsyncSession = Depends(get_db),
+    current_user=Depends(_materialverantwortliche),
+) -> Lagerbewegung:
+    objekt = await db.get(Objekt, objekt_id)
+    if objekt is None:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Objekt nicht gefunden")
+    ziel = await db.get(Standort, payload.nach_standort_id)
+    if ziel is None:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Standort nicht gefunden")
+    if payload.nach_standort_id == objekt.standort_id:
+        raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="Objekt ist bereits an diesem Standort")
+
+    return await verschiebe_objekt(
+        db,
+        objekt=objekt,
+        nach_standort_id=payload.nach_standort_id,
+        benutzer_id=current_user.id,
+        grund=payload.grund,
+    )

backend/app/models/lagerbewegung.py

@@ -0,0 +1,24 @@
+import uuid
+from datetime import datetime
+
+from sqlalchemy import ForeignKey, String
+from sqlalchemy.dialects.postgresql import TIMESTAMP, UUID
+from sqlalchemy.orm import Mapped, mapped_column
+
+from app.db.base import Base
+
+
+class Lagerbewegung(Base):
+    """Roadmap Phase 4 (Modul Lager): protokolliert Standortwechsel eines
+    Objekts (Rucksack/Gerät/Fahrzeug - der Materialträger, Karte 11). Append-
+    only wie Historie, keine Update/Delete-Operation vorgesehen."""
+
+    __tablename__ = "lagerbewegung"
+
+    id: Mapped[uuid.UUID] = mapped_column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
+    objekt_id: Mapped[int] = mapped_column(ForeignKey("objekt.id"), nullable=False)
+    von_standort_id: Mapped[int | None] = mapped_column(ForeignKey("standort.id"))
+    nach_standort_id: Mapped[int] = mapped_column(ForeignKey("standort.id"), nullable=False)
+    benutzer_id: Mapped[int] = mapped_column(ForeignKey("benutzer.id"), nullable=False)
+    zeitpunkt: Mapped[datetime] = mapped_column(TIMESTAMP(timezone=True), nullable=False)
+    grund: Mapped[str | None] = mapped_column(String)

backend/app/schemas/lagerbewegung.py

@@ -0,0 +1,20 @@
+import uuid
+from datetime import datetime
+
+from pydantic import BaseModel, ConfigDict
+
+
+class LagerbewegungCreate(BaseModel):
+    nach_standort_id: int
+    grund: str | None = None
+
+
+class LagerbewegungRead(BaseModel):
+    model_config = ConfigDict(from_attributes=True)
+    id: uuid.UUID
+    objekt_id: int
+    von_standort_id: int | None
+    nach_standort_id: int
+    benutzer_id: int
+    zeitpunkt: datetime
+    grund: str | None

backend/app/services/lagerbewegung.py

@@ -0,0 +1,26 @@
+from datetime import datetime, timezone
+
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.models.lagerbewegung import Lagerbewegung
+from app.models.objekt import Objekt
+
+
+async def verschiebe_objekt(
+    db: AsyncSession, *, objekt: Objekt, nach_standort_id: int, benutzer_id: int, grund: str | None
+) -> Lagerbewegung:
+    """Roadmap Phase 4 (Modul Lager): setzt objekt.standort_id und protokolliert
+    die Bewegung in einem Zug - von_standort_id ist der ALTE Stand vor dem
+    Setzen (muss vor der Zuweisung gelesen werden)."""
+    bewegung = Lagerbewegung(
+        objekt_id=objekt.id,
+        von_standort_id=objekt.standort_id,
+        nach_standort_id=nach_standort_id,
+        benutzer_id=benutzer_id,
+        zeitpunkt=datetime.now(timezone.utc),
+        grund=grund,
+    )
+    objekt.standort_id = nach_standort_id
+    db.add(bewegung)
+    await db.flush()
+    return bewegung

backend/tests/test_lagerbewegung.py

@@ -0,0 +1,89 @@
+import pytest
+
+from tests.conftest import auth_header, login
+
+
+@pytest.mark.asyncio
+async def test_objekt_verschieben_protokolliert_bewegung(
+    client, objekt_mit_position, standort_factory, materialverantwortlicher_user
+):
+    objekt, _material = objekt_mit_position
+    ziel = await standort_factory("Aussenlager")
+    token = await login(client, "materialverantwortlicher1")
+
+    bewegung = await client.post(
+        f"/api/v1/objekte/{objekt.id}/lagerbewegungen",
+        json={"nach_standort_id": ziel.id, "grund": "Umzug"},
+        headers=auth_header(token),
+    )
+    assert bewegung.status_code == 201
+    body = bewegung.json()
+    assert body["objekt_id"] == objekt.id
+    assert body["von_standort_id"] == objekt.standort_id
+    assert body["nach_standort_id"] == ziel.id
+    assert body["grund"] == "Umzug"
+
+    objekt_neu = await client.get(f"/api/v1/objekte/{objekt.id}", headers=auth_header(token))
+    assert objekt_neu.json()["standort_id"] == ziel.id
+
+    liste = await client.get(f"/api/v1/lagerbewegungen?objekt_id={objekt.id}", headers=auth_header(token))
+    assert len(liste.json()) == 1
+
+
+@pytest.mark.asyncio
+async def test_mitarbeiter_darf_nicht_verschieben(
+    client, objekt_mit_position, standort_factory, mitarbeiter_user
+):
+    objekt, _material = objekt_mit_position
+    ziel = await standort_factory("Aussenlager2")
+    token = await login(client, "mitarbeiter1")
+
+    response = await client.post(
+        f"/api/v1/objekte/{objekt.id}/lagerbewegungen",
+        json={"nach_standort_id": ziel.id},
+        headers=auth_header(token),
+    )
+    assert response.status_code == 403
+
+
+@pytest.mark.asyncio
+async def test_verschieben_an_gleichen_standort_ist_409(client, objekt_mit_position, materialverantwortlicher_user):
+    objekt, _material = objekt_mit_position
+    token = await login(client, "materialverantwortlicher1")
+
+    response = await client.post(
+        f"/api/v1/objekte/{objekt.id}/lagerbewegungen",
+        json={"nach_standort_id": objekt.standort_id},
+        headers=auth_header(token),
+    )
+    assert response.status_code == 409
+
+
+@pytest.mark.asyncio
+async def test_verschieben_zu_unbekanntem_standort_ist_404(client, objekt_mit_position, materialverantwortlicher_user):
+    objekt, _material = objekt_mit_position
+    token = await login(client, "materialverantwortlicher1")
+
+    response = await client.post(
+        f"/api/v1/objekte/{objekt.id}/lagerbewegungen",
+        json={"nach_standort_id": 999999},
+        headers=auth_header(token),
+    )
+    assert response.status_code == 404
+
+
+@pytest.mark.asyncio
+async def test_lagerbewegungen_nach_standort_filterbar(
+    client, objekt_mit_position, standort_factory, materialverantwortlicher_user
+):
+    objekt, _material = objekt_mit_position
+    ziel = await standort_factory("Aussenlager3")
+    token = await login(client, "materialverantwortlicher1")
+    await client.post(
+        f"/api/v1/objekte/{objekt.id}/lagerbewegungen",
+        json={"nach_standort_id": ziel.id},
+        headers=auth_header(token),
+    )
+
+    gefiltert = await client.get(f"/api/v1/lagerbewegungen?standort_id={ziel.id}", headers=auth_header(token))
+    assert len(gefiltert.json()) == 1