Sprint 6: Dashboard-Kennzahlen, Ablaufdaten-Warnungen, E-Mail-Benachrichtigung

- benutzer.email ergänzt (additive Migration 0003) - Prompt 20 hatte kein E-Mail-
  Feld vorgesehen, aber Karte 05 braucht es für Benachrichtigungen
- Dashboard-Kennzahlen (Prompt 12.2): offen/in_bearbeitung/nachgefuellt_teilweise
  zählen gemeinsam als "offen", genehmigte Mindermengen separat, kürzlich erledigt
  (7 Tage), problematische Objekte
- Ablaufdaten-Warnungen (Prompt 14): serverseitige Statusberechnung
  gueltig/bald_ablaufend/abgelaufen je Objektposition, Standard-Warnzeitraum 30
  Tage falls am Material nicht gesetzt, sortiert nach verbleibenden Tagen (E1:
  einzige Zeit-/Fälligkeits-Sicht in V1, keine Kontrollintervall-Logik)
- GET /fehlbestaende um Filter (Standort/Objekt/Material/genehmigt) und
  Alter-Sortierung erweitert (Prompt 12.3)
- Asynchrone E-Mail-Benachrichtigung bei neuem Fehlbestand (Karte 05): fire-and-
  forget an aktive Materialverantwortliche/Leitungsverantwortliche mit hinterlegter
  E-Mail; kein SMTP konfiguriert -> wird nur geloggt, kein harter Fehler
- Tests: Aggregationsregel, Ablaufdaten-Filterung/Sortierung, Rollenrechte,
  E-Mail-Versand (SMTP gemockt, kein Docker/Test-Mailserver im Host-Runner
  verfügbar - Aufruf mit korrekten Empfängern/Betreff wird geprüft)

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

DEVLOG.md

@@ -1304,3 +1304,49 @@ Keine Commits in dieser Session.
 - backend/tests/test_fehlbestand.py             | 138 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
 
 ---
+## 2026-09-03 23:47 – 23:53 (5m)
+**Beschreibung:** Claude Code Session
+**Projekt:** asb-material
+
+### Commits
+- 1e9012c Sprint 5: Mindermengen-Genehmigung + automatischer Ablauf, Historie für Kernkette
+
+### Geänderte Dateien
+- DEVLOG.md                                     |  63 ++++++++++++++++++++++++
+- backend/app/api/v1/endpoints/fehlbestaende.py |  56 +++++++++++++++++++++-
+- backend/app/api/v1/endpoints/kontrollen.py    |   8 +++-
+- backend/app/models/__init__.py                |   3 ++
+- backend/app/models/mindermenge.py             |  45 ++++++++++++++++++
+- backend/app/schemas/mindermenge.py            |  23 +++++++++
+- backend/app/services/fehlbestand.py           |  55 ++++++++++++++++++++-
+- backend/app/services/historie.py              |  37 +++++++++++++++
+- backend/app/services/kontrolle.py             | 116 +++++++++++++++++++++++++++++++++++++++------
+- backend/app/services/mindermenge.py           |  67 ++++++++++++++++++++++++++
+- backend/tests/conftest.py                     |  15 ++++++
+- backend/tests/test_historie.py                |  62 ++++++++++++++++++++++++
+- backend/tests/test_mindermenge.py             | 187 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
+
+---
+## 2026-09-03 23:53 – 23:53 (0m)
+**Beschreibung:** Claude Code Session
+**Projekt:** asb-material
+
+### Commits
+Keine Commits in dieser Session.
+
+### Geänderte Dateien
+- DEVLOG.md                                     |  63 ++++++++++++++++++++++++
+- backend/app/api/v1/endpoints/fehlbestaende.py |  56 +++++++++++++++++++++-
+- backend/app/api/v1/endpoints/kontrollen.py    |   8 +++-
+- backend/app/models/__init__.py                |   3 ++
+- backend/app/models/mindermenge.py             |  45 ++++++++++++++++++
+- backend/app/schemas/mindermenge.py            |  23 +++++++++
+- backend/app/services/fehlbestand.py           |  55 ++++++++++++++++++++-
+- backend/app/services/historie.py              |  37 +++++++++++++++
+- backend/app/services/kontrolle.py             | 116 +++++++++++++++++++++++++++++++++++++++------
+- backend/app/services/mindermenge.py           |  67 ++++++++++++++++++++++++++
+- backend/tests/conftest.py                     |  15 ++++++
+- backend/tests/test_historie.py                |  62 ++++++++++++++++++++++++
+- backend/tests/test_mindermenge.py             | 187 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
+
+---

backend/alembic/versions/0003_add_benutzer_email.py

@@ -0,0 +1,26 @@
+"""benutzer.email ergänzen (Karte 05: E-Mail-Benachrichtigung bei neuem Fehlbestand)
+
+Revision ID: 0003_add_benutzer_email
+Revises: 0002_seed_hauptserver
+Create Date: 2026-09-04
+
+Additive Schema-Erweiterung (Sprint 6): Prompt 20 hatte kein E-Mail-Feld auf
+benutzer vorgesehen, wird aber für Karte 05 gebraucht. Nullable, damit bestehende
+Benutzer nicht brechen - Pflicht auf Anwendungsebene für neue Verantwortliche.
+"""
+from typing import Sequence, Union
+
+from alembic import op
+
+revision: str = "0003_add_benutzer_email"
+down_revision: Union[str, None] = "0002_seed_hauptserver"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    op.execute("ALTER TABLE benutzer ADD COLUMN email TEXT;")
+
+
+def downgrade() -> None:
+    op.execute("ALTER TABLE benutzer DROP COLUMN email;")

backend/app/api/v1/api.py

@@ -3,6 +3,7 @@ from fastapi import APIRouter
 from app.api.v1.endpoints import (
     auth,
     benutzer,
+    dashboard,
     fehlbestaende,
     health,
     kontrollen,
@@ -22,3 +23,4 @@ api_router.include_router(vorlagen.router, tags=["vorlagen"])
 api_router.include_router(objekte.router, tags=["objekte"])
 api_router.include_router(kontrollen.router, tags=["kontrollen"])
 api_router.include_router(fehlbestaende.router, tags=["fehlbestaende"])
+api_router.include_router(dashboard.router, tags=["dashboard"])

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

@@ -30,6 +30,7 @@ async def erstelle_benutzer(
     benutzer = Benutzer(
         name=payload.name,
         login=payload.login,
+        email=payload.email,
         passwort_hash=hash_password(payload.passwort),
         aktiv=True,
     )
@@ -55,6 +56,8 @@ async def aendere_benutzer(
 
     if payload.name is not None:
         benutzer.name = payload.name
+    if payload.email is not None:
+        benutzer.email = payload.email
     if payload.aktiv is not None:
         benutzer.aktiv = payload.aktiv
     if payload.rollen is not None:

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

@@ -0,0 +1,27 @@
+from fastapi import APIRouter, Depends
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.api.deps import require_roles
+from app.db.session import get_db
+from app.models.auth import RolleTyp
+from app.schemas.dashboard import AblaufdatumRead, KennzahlenRead
+from app.services.dashboard import bevorstehende_ablaufdaten, kennzahlen
+
+router = APIRouter()
+
+_verantwortliche = require_roles(
+    RolleTyp.administration, RolleTyp.materialverantwortlicher, RolleTyp.leitungsverantwortlicher
+)
+
+
+@router.get("/dashboard/kennzahlen", response_model=KennzahlenRead)
+async def hole_kennzahlen(db: AsyncSession = Depends(get_db), _=Depends(_verantwortliche)) -> KennzahlenRead:
+    """Prompt 12.1: Zielgruppe Materialverantwortlicher/Leitungsverantwortlicher/
+    Administration. Filterung nach Zuständigkeitsbereich (Karte 04) folgt später,
+    V1 zeigt globale Zahlen für diese Rollen."""
+    return KennzahlenRead(**await kennzahlen(db))
+
+
+@router.get("/dashboard/ablaufdaten", response_model=list[AblaufdatumRead])
+async def hole_ablaufdaten(db: AsyncSession = Depends(get_db), _=Depends(_verantwortliche)) -> list[AblaufdatumRead]:
+    return [AblaufdatumRead(**eintrag) for eintrag in await bevorstehende_ablaufdaten(db)]

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

@@ -8,6 +8,7 @@ from app.api.deps import require_roles
 from app.db.session import get_db
 from app.models.auth import KnotenTyp, RolleTyp, Systemknoten
 from app.models.fehlbestand import Fehlbestand
+from app.models.mindermenge import MindermengeStatus, MindermengenGenehmigung
 from app.schemas.fehlbestand import (
     FehlbestandRead,
     NachfuellungCreate,
@@ -41,13 +42,30 @@ async def _hauptserver_id(db: AsyncSession) -> int:
 @router.get("/fehlbestaende", response_model=list[FehlbestandRead])
 async def liste_fehlbestaende(
     status_filter: str | None = None,
+    standort_id: int | None = None,
+    objekt_id: int | None = None,
+    material_id: int | None = None,
+    nur_genehmigte: bool = False,
     db: AsyncSession = Depends(get_db),
     _=Depends(_verantwortliche),
 ) -> list[Fehlbestand]:
-    """Vollständige Filterung nach Zuständigkeit/Standort folgt Sprint 6 (Dashboard)."""
+    """Prompt 12.3 (Minimalversion): Filter Standort/Objekt/Material/Status,
+    Standardsortierung nach Alter absteigend (älteste zuerst, entstanden_am aufsteigend).
+    Vollständige Zuständigkeitsfilterung (Karte 04) ist spätere Ausbaustufe."""
     stmt = select(Fehlbestand)
     if status_filter is not None:
         stmt = stmt.where(Fehlbestand.status == status_filter)
+    if standort_id is not None:
+        stmt = stmt.where(Fehlbestand.standort_id == standort_id)
+    if objekt_id is not None:
+        stmt = stmt.where(Fehlbestand.objekt_id == objekt_id)
+    if material_id is not None:
+        stmt = stmt.where(Fehlbestand.material_id == material_id)
+    if nur_genehmigte:
+        stmt = stmt.join(
+            MindermengenGenehmigung, MindermengenGenehmigung.fehlbestand_id == Fehlbestand.id
+        ).where(MindermengenGenehmigung.status == MindermengeStatus.aktiv)
+    stmt = stmt.order_by(Fehlbestand.entstanden_am.asc())
     result = await db.execute(stmt)
     return list(result.scalars().all())
 

backend/app/core/app_settings.py

@@ -14,5 +14,15 @@ class Settings(BaseSettings):
     # ID des systemknoten-Datensatzes mit typ='haupt' (Sprintplan E6).
     systemknoten_id: int = 1
 
+    # Karte 05 / Sprint 6: E-Mail-Benachrichtigung bei neuem Fehlbestand.
+    # Leer (Default) = Versand wird übersprungen, nur geloggt (sicherer Default für
+    # Entwicklung/Tests ohne konfigurierten Mailserver).
+    smtp_host: str = ""
+    smtp_port: int = 587
+    smtp_user: str = ""
+    smtp_password: str = ""
+    smtp_from: str = "mabea@example.org"
+    smtp_use_tls: bool = True
+
 
 settings = Settings()

backend/app/core/email.py

@@ -0,0 +1,35 @@
+import logging
+from email.message import EmailMessage
+
+import aiosmtplib
+
+from app.core.app_settings import settings
+
+logger = logging.getLogger(__name__)
+
+
+async def send_email(*, to: list[str], subject: str, body: str) -> None:
+    """Karte 05: Benachrichtigung bei neuem Fehlbestand. Ohne konfigurierten SMTP_HOST
+    (Default auf Entwicklungs-/Testsystemen) wird nur geloggt, nicht gesendet - kein
+    harter Fehler, falls die Mailkonfiguration auf dem Zielsystem noch aussteht."""
+    if not to:
+        logger.info("E-Mail-Versand übersprungen (keine Empfänger): %s", subject)
+        return
+    if not settings.smtp_host:
+        logger.info("SMTP nicht konfiguriert, E-Mail nicht versendet: %s an %s", subject, to)
+        return
+
+    message = EmailMessage()
+    message["From"] = settings.smtp_from
+    message["To"] = ", ".join(to)
+    message["Subject"] = subject
+    message.set_content(body)
+
+    await aiosmtplib.send(
+        message,
+        hostname=settings.smtp_host,
+        port=settings.smtp_port,
+        username=settings.smtp_user or None,
+        password=settings.smtp_password or None,
+        start_tls=settings.smtp_use_tls,
+    )

backend/app/models/auth.py

@@ -43,6 +43,7 @@ class Benutzer(Base):
     login: Mapped[str] = mapped_column(String, unique=True, nullable=False)
     passwort_hash: Mapped[str] = mapped_column(String, nullable=False)
     aktiv: Mapped[bool] = mapped_column(Boolean, nullable=False, default=True)
+    email: Mapped[str | None] = mapped_column(String)  # Sprint 6 / Karte 05 (Migration 0003)
 
     rollen: Mapped[list["BenutzerRolle"]] = relationship(back_populates="benutzer", lazy="selectin")
 

backend/app/schemas/benutzer.py

@@ -7,6 +7,7 @@ class BenutzerCreate(BaseModel):
     name: str
     login: str
     passwort: str
+    email: str | None = None
     rollen: list[RolleTyp] = []
 
 
@@ -15,6 +16,7 @@ class BenutzerRead(BaseModel):
     id: int
     name: str
     login: str
+    email: str | None
     aktiv: bool
     rollen: list[str]
 
@@ -24,6 +26,7 @@ class BenutzerRead(BaseModel):
             id=benutzer.id,
             name=benutzer.name,
             login=benutzer.login,
+            email=benutzer.email,
             aktiv=benutzer.aktiv,
             rollen=benutzer.rollen_namen,
         )
@@ -31,5 +34,6 @@ class BenutzerRead(BaseModel):
 
 class BenutzerUpdate(BaseModel):
     name: str | None = None
+    email: str | None = None
     aktiv: bool | None = None
     rollen: list[RolleTyp] | None = None

backend/app/schemas/dashboard.py

@@ -0,0 +1,20 @@
+from datetime import date
+
+from pydantic import BaseModel
+
+
+class KennzahlenRead(BaseModel):
+    anzahl_offener_fehlbestaende: int
+    anzahl_genehmigter_mindermengen: int
+    anzahl_kuerzlich_erledigter_fehlbestaende: int
+    anzahl_bevorstehender_ablaufdaten: int
+    anzahl_problematischer_objekte: int
+    anzahl_objekte_gesamt: int
+
+
+class AblaufdatumRead(BaseModel):
+    objekt_id: int
+    material_id: int
+    ablaufdatum: date
+    verbleibende_tage: int
+    status: str

backend/app/services/benachrichtigung.py

@@ -0,0 +1,59 @@
+import asyncio
+import logging
+
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.core.email import send_email
+from app.models.auth import Benutzer, BenutzerRolle, RolleTyp
+from app.models.fehlbestand import Fehlbestand
+from app.models.stammdaten import Material
+
+logger = logging.getLogger(__name__)
+
+
+async def _empfaenger_fuer_objekt(db: AsyncSession) -> list[str]:
+    """V1-Vereinfachung: alle aktiven Materialverantwortlichen/Leitungsverantwortlichen
+    (Karte 05 spezifiziert keine feinere, zuständigkeitsbasierte Adressierung).
+    Feinere Filterung nach Zuständigkeit (Karte 04) wäre eine spätere Ausbaustufe."""
+    result = await db.execute(
+        select(Benutzer.email)
+        .join(BenutzerRolle, BenutzerRolle.benutzer_id == Benutzer.id)
+        .where(
+            BenutzerRolle.rolle.in_(
+                [RolleTyp.materialverantwortlicher, RolleTyp.leitungsverantwortlicher]
+            ),
+            Benutzer.aktiv.is_(True),
+            Benutzer.email.is_not(None),
+        )
+    )
+    return [email for (email,) in result.all() if email]
+
+
+async def benachrichtige_neuer_fehlbestand(db: AsyncSession, *, fehlbestand: Fehlbestand) -> asyncio.Task:
+    """Karte 05: asynchroner E-Mail-Versand bei neuem Fehlbestand, blockiert den
+    Kontroll-Workflow nicht (fire-and-forget, Fehler beim Versand werden geloggt,
+    nicht an den aufrufenden Request weitergereicht). Gibt den Task zurück, damit
+    Tests den Versand deterministisch abwarten können - im normalen Betrieb wird
+    der Rückgabewert bewusst nicht awaited."""
+    empfaenger = await _empfaenger_fuer_objekt(db)
+    material = await db.get(Material, fehlbestand.material_id)
+    material_name = material.name if material else str(fehlbestand.material_id)
+
+    subject = f"Neuer Fehlbestand: {material_name}"
+    body = (
+        f"Material: {material_name}\n"
+        f"Objekt-ID: {fehlbestand.objekt_id}\n"
+        f"Sollmenge: {fehlbestand.sollmenge}\n"
+        f"Istmenge: {fehlbestand.istmenge}\n"
+        f"Fehlmenge: {fehlbestand.fehlmenge}\n"
+        f"Entstanden am: {fehlbestand.entstanden_am.isoformat()}\n"
+    )
+
+    async def _senden() -> None:
+        try:
+            await send_email(to=empfaenger, subject=subject, body=body)
+        except Exception:  # noqa: BLE001 - Versandfehler dürfen den Workflow nie stören
+            logger.exception("E-Mail-Versand für Fehlbestand %s fehlgeschlagen", fehlbestand.id)
+
+    return asyncio.create_task(_senden())

backend/app/services/dashboard.py

@@ -0,0 +1,91 @@
+from datetime import date, datetime, timedelta, timezone
+
+from sqlalchemy import func, select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.models.fehlbestand import Fehlbestand, FehlbestandStatus
+from app.models.mindermenge import MindermengeStatus, MindermengenGenehmigung
+from app.models.objekt import Objekt
+from app.models.objektposition import Objektposition
+from app.models.stammdaten import Material
+
+# Prompt 12.2: offen/in_bearbeitung/nachgefuellt_teilweise zählen gemeinsam als "offen"
+# (technischer Substatus ist keine Erledigung, Prompt 03/20).
+OFFEN_STATI = (FehlbestandStatus.offen, FehlbestandStatus.in_bearbeitung, FehlbestandStatus.nachgefuellt_teilweise)
+
+STANDARD_WARNZEITRAUM_TAGE = 30
+
+
+async def kennzahlen(db: AsyncSession) -> dict:
+    jetzt = datetime.now(timezone.utc)
+
+    result = await db.execute(select(func.count()).select_from(Fehlbestand).where(Fehlbestand.status.in_(OFFEN_STATI)))
+    anzahl_offen = result.scalar_one()
+
+    result = await db.execute(
+        select(func.count(func.distinct(MindermengenGenehmigung.fehlbestand_id))).where(
+            MindermengenGenehmigung.status == MindermengeStatus.aktiv
+        )
+    )
+    anzahl_genehmigt = result.scalar_one()
+
+    result = await db.execute(
+        select(func.count())
+        .select_from(Fehlbestand)
+        .where(Fehlbestand.status == FehlbestandStatus.erledigt, Fehlbestand.erledigt_am >= jetzt - timedelta(days=7))
+    )
+    anzahl_kuerzlich_erledigt = result.scalar_one()
+
+    result = await db.execute(select(func.count(func.distinct(Fehlbestand.objekt_id))).where(Fehlbestand.status.in_(OFFEN_STATI)))
+    anzahl_problematische_objekte = result.scalar_one()
+
+    result = await db.execute(select(func.count()).select_from(Objekt))
+    anzahl_objekte_gesamt = result.scalar_one()
+
+    ablaufdaten_liste = await bevorstehende_ablaufdaten(db)
+
+    return {
+        "anzahl_offener_fehlbestaende": anzahl_offen,
+        "anzahl_genehmigter_mindermengen": anzahl_genehmigt,
+        "anzahl_kuerzlich_erledigter_fehlbestaende": anzahl_kuerzlich_erledigt,
+        "anzahl_bevorstehender_ablaufdaten": len(ablaufdaten_liste),
+        "anzahl_problematischer_objekte": anzahl_problematische_objekte,
+        "anzahl_objekte_gesamt": anzahl_objekte_gesamt,
+    }
+
+
+def _ablauf_status(ablaufdatum: date, warnzeitraum_tage: int, heute: date) -> tuple[str, int]:
+    verbleibende_tage = (ablaufdatum - heute).days
+    if verbleibende_tage < 0:
+        return "abgelaufen", verbleibende_tage
+    if verbleibende_tage <= warnzeitraum_tage:
+        return "bald_ablaufend", verbleibende_tage
+    return "gueltig", verbleibende_tage
+
+
+async def bevorstehende_ablaufdaten(db: AsyncSession) -> list[dict]:
+    """Prompt 14.5: nur bald_ablaufend/abgelaufen, sortiert nach verbleibenden Tagen
+    aufsteigend (dringendste zuerst). "gueltig" gehört nicht in diese Liste."""
+    heute = datetime.now(timezone.utc).date()
+    result = await db.execute(
+        select(Objektposition, Material)
+        .join(Material, Objektposition.material_id == Material.id)
+        .where(Objektposition.ablaufdatum.is_not(None))
+    )
+    eintraege = []
+    for objektposition, material in result.all():
+        warnzeitraum = material.warnzeitraum_tage or STANDARD_WARNZEITRAUM_TAGE
+        status_wert, verbleibende_tage = _ablauf_status(objektposition.ablaufdatum, warnzeitraum, heute)
+        if status_wert == "gueltig":
+            continue
+        eintraege.append(
+            {
+                "objekt_id": objektposition.objekt_id,
+                "material_id": objektposition.material_id,
+                "ablaufdatum": objektposition.ablaufdatum,
+                "verbleibende_tage": verbleibende_tage,
+                "status": status_wert,
+            }
+        )
+    eintraege.sort(key=lambda e: e["verbleibende_tage"])
+    return eintraege

backend/app/services/kontrolle.py

@@ -10,6 +10,7 @@ from app.models.kontrolle import Kontrolle, KontrollStatus, Kontrollposition
 from app.models.mindermenge import MindermengeStatus, MindermengenGenehmigung
 from app.models.objekt import Objekt
 from app.models.objektposition import Objektposition, ObjektpositionStatus
+from app.services import benachrichtigung as benachrichtigung_service
 from app.services import historie as historie_service
 from app.services.objekte import sollmenge_effektiv, sollmenge_map
 
@@ -257,6 +258,7 @@ async def _synchronisiere_fehlbestand(
                     "fehlmenge": str(unterdeckung),
                 },
             )
+            await benachrichtigung_service.benachrichtige_neuer_fehlbestand(db, fehlbestand=bestehender)
         else:
             # Korrektur vor Abschluss (z. B. Tippfehler nochmal erfasst) - solange die
             # Kontrolle noch läuft, ist der von IHR erzeugte Fehlbestand noch nicht

backend/pyproject.toml

@@ -17,6 +17,7 @@ dependencies = [
     "bcrypt>=4.0,<4.1",
     "pyjwt>=2.9,<2.10",
     "python-multipart>=0.0.12,<0.1",
+    "aiosmtplib>=3.0,<3.1",
 ]
 
 [project.optional-dependencies]

backend/tests/conftest.py

@@ -96,6 +96,7 @@ async def materialverantwortlicher_user(db_session):
     benutzer = Benutzer(
         name="Test Materialverantwortlicher",
         login="materialverantwortlicher1",
+        email="mv1@example.org",
         passwort_hash=hash_password("test-passwort-123"),
         aktiv=True,
     )

backend/tests/test_dashboard.py

@@ -0,0 +1,147 @@
+import pytest
+
+from tests.conftest import auth_header, login
+
+
+async def _fehlbestand(client, token, objekt, material, istmenge: str, abschliessen=True):
+    start = await client.post(
+        f"/api/v1/objekte/{objekt.id}/kontrollen", json={"uebernehmen": False}, headers=auth_header(token)
+    )
+    kontrolle_id = start.json()["id"]
+    put_response = await client.put(
+        f"/api/v1/kontrollen/{kontrolle_id}/positionen/{material.id}",
+        json={"istmenge": istmenge},
+        headers=auth_header(token),
+    )
+    if abschliessen:
+        await client.post(f"/api/v1/kontrollen/{kontrolle_id}/abschliessen", headers=auth_header(token))
+    return put_response.json()["fehlbestand_id"]
+
+
+@pytest.mark.asyncio
+async def test_offene_stati_werden_gemeinsam_gezaehlt(
+    client, objekt_mit_position, mitarbeiter_user, materialverantwortlicher_user
+):
+    """Prompt 12.2: offen + in_bearbeitung + nachgefuellt_teilweise = "offen"."""
+    objekt, material = objekt_mit_position
+    mitarbeiter_token = await login(client, "mitarbeiter1")
+    fehlbestand_id = await _fehlbestand(client, mitarbeiter_token, objekt, material, "6")
+
+    # Teilnachfüllung -> Status nachgefuellt_teilweise, zählt trotzdem als offen
+    await client.post(
+        f"/api/v1/fehlbestaende/{fehlbestand_id}/nachfuellungen",
+        json={"menge": "1"},
+        headers=auth_header(mitarbeiter_token),
+    )
+
+    verantwortlicher_token = await login(client, "materialverantwortlicher1")
+    response = await client.get("/api/v1/dashboard/kennzahlen", headers=auth_header(verantwortlicher_token))
+    assert response.status_code == 200
+    assert response.json()["anzahl_offener_fehlbestaende"] == 1
+
+
+@pytest.mark.asyncio
+async def test_genehmigte_mindermenge_zaehlt_als_offen_und_separat(
+    client, objekt_mit_position, mitarbeiter_user, materialverantwortlicher_user
+):
+    objekt, material = objekt_mit_position
+    mitarbeiter_token = await login(client, "mitarbeiter1")
+    fehlbestand_id = await _fehlbestand(client, mitarbeiter_token, objekt, material, "6")
+
+    verantwortlicher_token = await login(client, "materialverantwortlicher1")
+    await client.post(
+        f"/api/v1/fehlbestaende/{fehlbestand_id}/mindermenge",
+        json={"begruendung": "Lieferengpass"},
+        headers=auth_header(verantwortlicher_token),
+    )
+
+    response = await client.get("/api/v1/dashboard/kennzahlen", headers=auth_header(verantwortlicher_token))
+    body = response.json()
+    assert body["anzahl_genehmigter_mindermengen"] == 1
+    assert body["anzahl_offener_fehlbestaende"] == 1  # Genehmigung ist keine Erledigung
+
+
+@pytest.mark.asyncio
+async def test_kuerzlich_erledigt_wird_gezaehlt(
+    client, objekt_mit_position, mitarbeiter_user, materialverantwortlicher_user
+):
+    objekt, material = objekt_mit_position
+    mitarbeiter_token = await login(client, "mitarbeiter1")
+    fehlbestand_id = await _fehlbestand(client, mitarbeiter_token, objekt, material, "6")
+    await client.post(
+        f"/api/v1/fehlbestaende/{fehlbestand_id}/nachfuellungen",
+        json={"menge": "4"},
+        headers=auth_header(mitarbeiter_token),
+    )
+
+    verantwortlicher_token = await login(client, "materialverantwortlicher1")
+    response = await client.get("/api/v1/dashboard/kennzahlen", headers=auth_header(verantwortlicher_token))
+    body = response.json()
+    assert body["anzahl_kuerzlich_erledigter_fehlbestaende"] == 1
+    assert body["anzahl_offener_fehlbestaende"] == 0
+
+
+@pytest.mark.asyncio
+async def test_mitarbeiter_darf_dashboard_nicht_sehen(client, mitarbeiter_user):
+    token = await login(client, "mitarbeiter1")
+    response = await client.get("/api/v1/dashboard/kennzahlen", headers=auth_header(token))
+    assert response.status_code == 403
+
+
+@pytest.mark.asyncio
+async def test_ablaufdaten_liste_sortiert_und_gefiltert(
+    client, db_session, objekt_mit_position, materialverantwortlicher_user
+):
+    """Prompt 14: nur bald_ablaufend/abgelaufen, sortiert nach verbleibenden Tagen aufsteigend."""
+    from datetime import date, timedelta
+
+    objekt, material = objekt_mit_position
+    verantwortlicher_token = await login(client, "materialverantwortlicher1")
+
+    positionen = (
+        await client.get(f"/api/v1/objekte/{objekt.id}/positionen", headers=auth_header(verantwortlicher_token))
+    ).json()
+    position_id = positionen[0]["id"]
+
+    # weit in der Zukunft -> "gueltig", darf NICHT in der Liste auftauchen
+    await client.patch(
+        f"/api/v1/objekte/{objekt.id}/positionen/{position_id}",
+        json={"ablaufdatum": (date.today() + timedelta(days=365)).isoformat()},
+        headers=auth_header(verantwortlicher_token),
+    )
+    response = await client.get("/api/v1/dashboard/ablaufdaten", headers=auth_header(verantwortlicher_token))
+    assert response.json() == []
+
+    # bald ablaufend (Standard-Warnzeitraum 30 Tage, kein warnzeitraum_tage auf Material gesetzt)
+    await client.patch(
+        f"/api/v1/objekte/{objekt.id}/positionen/{position_id}",
+        json={"ablaufdatum": (date.today() + timedelta(days=5)).isoformat()},
+        headers=auth_header(verantwortlicher_token),
+    )
+    response = await client.get("/api/v1/dashboard/ablaufdaten", headers=auth_header(verantwortlicher_token))
+    body = response.json()
+    assert len(body) == 1
+    assert body[0]["status"] == "bald_ablaufend"
+
+
+@pytest.mark.asyncio
+async def test_abgelaufenes_material_wird_als_abgelaufen_markiert(
+    client, objekt_mit_position, materialverantwortlicher_user
+):
+    from datetime import date, timedelta
+
+    objekt, material = objekt_mit_position
+    token = await login(client, "materialverantwortlicher1")
+    positionen = (await client.get(f"/api/v1/objekte/{objekt.id}/positionen", headers=auth_header(token))).json()
+    position_id = positionen[0]["id"]
+
+    await client.patch(
+        f"/api/v1/objekte/{objekt.id}/positionen/{position_id}",
+        json={"ablaufdatum": (date.today() - timedelta(days=3)).isoformat()},
+        headers=auth_header(token),
+    )
+    response = await client.get("/api/v1/dashboard/ablaufdaten", headers=auth_header(token))
+    body = response.json()
+    assert len(body) == 1
+    assert body[0]["status"] == "abgelaufen"
+    assert body[0]["verbleibende_tage"] == -3

backend/tests/test_email.py

@@ -0,0 +1,68 @@
+import asyncio
+from unittest.mock import AsyncMock, patch
+
+import pytest
+
+from tests.conftest import auth_header, login
+
+
+@pytest.mark.asyncio
+async def test_email_wird_bei_neuem_fehlbestand_an_verantwortliche_versendet(
+    client, objekt_mit_position, mitarbeiter_user, materialverantwortlicher_user
+):
+    """Karte 05: asynchrone E-Mail-Benachrichtigung bei neuem Fehlbestand.
+
+    Kein echter Test-Mailserver im Host-Runner (kein Docker) verfügbar - der
+    tatsächliche SMTP-Versand wird gemockt, geprüft wird der korrekte Aufruf
+    (Empfänger = aktive Materialverantwortliche mit hinterlegter E-Mail-Adresse)."""
+    objekt, material = objekt_mit_position
+
+    with patch("app.services.benachrichtigung.send_email", new=AsyncMock()) as mock_send:
+        token = await login(client, "mitarbeiter1")
+        start = await client.post(
+            f"/api/v1/objekte/{objekt.id}/kontrollen", json={"uebernehmen": False}, headers=auth_header(token)
+        )
+        kontrolle_id = start.json()["id"]
+        await client.put(
+            f"/api/v1/kontrollen/{kontrolle_id}/positionen/{material.id}",
+            json={"istmenge": "6"},
+            headers=auth_header(token),
+        )
+
+        for _ in range(20):
+            await asyncio.sleep(0.01)
+            if mock_send.await_count:
+                break
+
+    mock_send.assert_awaited_once()
+    kwargs = mock_send.await_args.kwargs
+    assert kwargs["to"] == ["mv1@example.org"]
+    assert "Fehlbestand" in kwargs["subject"]
+
+
+@pytest.mark.asyncio
+async def test_kein_email_versand_ohne_empfaenger(client, objekt_mit_position, mitarbeiter_user):
+    """Kein Materialverantwortlicher mit E-Mail vorhanden -> send_email wird gar nicht
+    mit Empfängern aufgerufen (leere Liste), kein Fehler im Kontroll-Workflow."""
+    objekt, material = objekt_mit_position
+
+    with patch("app.services.benachrichtigung.send_email", new=AsyncMock()) as mock_send:
+        token = await login(client, "mitarbeiter1")
+        start = await client.post(
+            f"/api/v1/objekte/{objekt.id}/kontrollen", json={"uebernehmen": False}, headers=auth_header(token)
+        )
+        kontrolle_id = start.json()["id"]
+        response = await client.put(
+            f"/api/v1/kontrollen/{kontrolle_id}/positionen/{material.id}",
+            json={"istmenge": "6"},
+            headers=auth_header(token),
+        )
+        assert response.status_code == 200  # Workflow funktioniert trotz fehlender Empfänger
+
+        for _ in range(20):
+            await asyncio.sleep(0.01)
+            if mock_send.await_count:
+                break
+
+    mock_send.assert_awaited_once()
+    assert mock_send.await_args.kwargs["to"] == []