Modulare Startseite: Kartenraster, eigene Anordnung je Benutzer, Auswertungen
Die Uebersicht war fest verdrahtet und fuer alle gleich. Jetzt besteht sie aus Karten in einem 12-Spalten-Raster, die sich ziehen, in der Groesse aendern und einrasten lassen (react-grid-layout). "Anordnen" schaltet das frei, gespeichert wird ausdruecklich - Probieren bleibt folgenlos. Anordnung: - Neue Tabelle dashboard_layouts; user_id = NULL ist die Admin-Vorgabe. - Jeder Benutzer hat seine eigene Anordnung. Admins speichern eine Vorgabe fuer alle und koennen sie ueber die Einstellung dashboard_enforced verbindlich machen; dann lehnt das Backend das Speichern eigener Anordnungen ab. - "Auf Vorgabe zuruecksetzen" verwirft die persoenliche Anordnung. 15 Karten (Verzeichnis web/src/dashboard/cards.jsx, neue Karte = ein Eintrag): Schnellzugriff, Status, Artikelzahl, Artikeleinheiten, Bald ablaufend, Abgelaufen, Einkaufsliste, letzte Bewegungen sowie sechs Auswertungen (Ablauf-Ring, Kategorien-Ring, Kategorien nach Zustand, Bestandsverlauf, Verlauf eines Artikels, Ein-/Auslagerungen). Diagramme als eigene SVG-Komponenten statt Diagramm-Paket: - Zustandsfarben (ok/bald/abgelaufen) sind Statusangaben aus den Tokens und stehen nie ohne Beschriftung; Kategorien nutzen eine gepruefte Farbreihe in fester Ordnung, ab sieben Kategorien wird zu "Andere" gebuendelt. - Eine Werteachse, duenne Marken, zurueckhaltendes Gitter, Fadenkreuz mit Kurzinfo. Hell und Dunkel haben eigene Farbstufen. Mengen durchgehend in Artikeleinheiten (Glaeser, Packungen, Stueck): Gramm und Stueck lassen sich nicht addieren, Gebinde schon. Der bisher dreifach vorhandene Helfer liegt jetzt einmal in services/conversion.py::article_unit. Der Verlauf wird rueckwaerts vom heutigen Bestand aus den Bewegungen rekonstruiert. Bekannte Ungenauigkeit, im Code und in der Roadmap vermerkt: Die Umrechnung nutzt die heutige Packungsgroesse. Geprueft: "npm run build" laeuft durch. Die neuen pytest-Tests (backend/tests/test_dashboard.py) konnten hier nicht laufen - auf diesem Rechner ist kein Python installiert. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
Scarriffle
@@ -11,6 +11,7 @@ from .routers import (
|
||||
auth,
|
||||
branding,
|
||||
categories,
|
||||
dashboard,
|
||||
groups,
|
||||
locations,
|
||||
maintenance,
|
||||
@@ -110,3 +111,4 @@ app.include_router(settings_router.router)
|
||||
app.include_router(branding.router)
|
||||
app.include_router(categories.router)
|
||||
app.include_router(maintenance.router)
|
||||
app.include_router(dashboard.router)
|
||||
|
||||
@@ -295,6 +295,27 @@ class BrandingAsset(Base):
|
||||
)
|
||||
|
||||
|
||||
class DashboardLayout(Base):
|
||||
"""Anordnung der Karten auf der Startseite.
|
||||
|
||||
``user_id = NULL`` ist die Vorgabe des Administrators: Sie dient neuen
|
||||
Benutzern als Ausgangspunkt und lässt sich – falls in den Einstellungen so
|
||||
gesetzt – auch verbindlich für alle machen.
|
||||
"""
|
||||
|
||||
__tablename__ = "dashboard_layouts"
|
||||
|
||||
id: Mapped[int] = mapped_column(Integer, primary_key=True)
|
||||
user_id: Mapped[int | None] = mapped_column(
|
||||
ForeignKey("users.id", ondelete="CASCADE"), unique=True, nullable=True
|
||||
)
|
||||
# JSON-Text: [{"i": "expiring", "x": 0, "y": 0, "w": 4, "h": 3, "props": {}}]
|
||||
layout: Mapped[str] = mapped_column(Text, nullable=False)
|
||||
updated_at: Mapped[datetime] = mapped_column(
|
||||
DateTime(timezone=True), default=_now, onupdate=_now
|
||||
)
|
||||
|
||||
|
||||
class Setting(Base):
|
||||
__tablename__ = "settings"
|
||||
|
||||
|
||||
374
backend/app/routers/dashboard.py
Normal file
374
backend/app/routers/dashboard.py
Normal file
@@ -0,0 +1,374 @@
|
||||
"""Startseite: gespeicherte Kartenanordnung und die Auswertungen dahinter.
|
||||
|
||||
Die Auswertungen laufen bewusst hier und nicht in der Oberfläche: Sonst müsste
|
||||
der Browser alle Chargen und Bewegungen laden, nur um ein paar Summen zu bilden.
|
||||
|
||||
Alle Mengenangaben sind **Artikeleinheiten** (Gläser, Packungen, Stück) – siehe
|
||||
:func:`app.services.conversion.article_unit`. Nur die lassen sich über
|
||||
verschiedene Artikel hinweg addieren.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
from collections import defaultdict
|
||||
from datetime import date, datetime, timedelta, timezone
|
||||
|
||||
from fastapi import APIRouter, Depends, HTTPException, status
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from ..database import get_db
|
||||
from ..deps import get_current_user, require_admin
|
||||
from ..models import (
|
||||
Category,
|
||||
DashboardLayout,
|
||||
Group,
|
||||
Lot,
|
||||
Movement,
|
||||
MovementType,
|
||||
Product,
|
||||
Setting,
|
||||
User,
|
||||
)
|
||||
from ..schemas import (
|
||||
ActivityPoint,
|
||||
CategoryShare,
|
||||
DashboardLayoutIn,
|
||||
DashboardLayoutOut,
|
||||
DashboardStats,
|
||||
ExpirySplit,
|
||||
TimelinePoint,
|
||||
)
|
||||
from ..services.conversion import article_unit
|
||||
from ..services.stock import current_stock
|
||||
from .settings import get_expiry_warning_days
|
||||
|
||||
router = APIRouter(prefix="/dashboard", tags=["dashboard"])
|
||||
|
||||
ENFORCED_KEY = "dashboard_enforced"
|
||||
|
||||
# Eingebaute Anordnung – wird benutzt, solange weder eigene noch Vorgabe existiert.
|
||||
BUILTIN_LAYOUT: list[dict] = [
|
||||
{"i": "actions", "x": 0, "y": 0, "w": 4, "h": 1},
|
||||
{"i": "status", "x": 4, "y": 0, "w": 8, "h": 1},
|
||||
{"i": "expiring", "x": 0, "y": 1, "w": 6, "h": 3},
|
||||
{"i": "shopping", "x": 6, "y": 1, "w": 6, "h": 3},
|
||||
{"i": "expiry-donut", "x": 0, "y": 4, "w": 4, "h": 3},
|
||||
{"i": "category-donut", "x": 4, "y": 4, "w": 4, "h": 3},
|
||||
{"i": "stock-timeline", "x": 8, "y": 4, "w": 4, "h": 3},
|
||||
]
|
||||
|
||||
|
||||
def _enforced(db: Session) -> bool:
|
||||
row = db.get(Setting, ENFORCED_KEY)
|
||||
return bool(row and row.value == "1")
|
||||
|
||||
|
||||
def _load(db: Session, user_id: int | None) -> list[dict] | None:
|
||||
row = (
|
||||
db.query(DashboardLayout)
|
||||
.filter(DashboardLayout.user_id.is_(None) if user_id is None else DashboardLayout.user_id == user_id)
|
||||
.first()
|
||||
)
|
||||
if row is None:
|
||||
return None
|
||||
try:
|
||||
data = json.loads(row.layout)
|
||||
except ValueError:
|
||||
return None
|
||||
return data if isinstance(data, list) else None
|
||||
|
||||
|
||||
def _store(db: Session, user_id: int | None, layout: list[dict]) -> None:
|
||||
row = (
|
||||
db.query(DashboardLayout)
|
||||
.filter(DashboardLayout.user_id.is_(None) if user_id is None else DashboardLayout.user_id == user_id)
|
||||
.first()
|
||||
)
|
||||
text = json.dumps(layout, ensure_ascii=False)
|
||||
if row is None:
|
||||
db.add(DashboardLayout(user_id=user_id, layout=text))
|
||||
else:
|
||||
row.layout = text
|
||||
db.commit()
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# Anordnung
|
||||
# --------------------------------------------------------------------------
|
||||
@router.get("/layout", response_model=DashboardLayoutOut)
|
||||
def get_layout(
|
||||
db: Session = Depends(get_db), user: User = Depends(get_current_user)
|
||||
) -> DashboardLayoutOut:
|
||||
"""Eigene Anordnung, sonst die Vorgabe, sonst die eingebaute."""
|
||||
erzwungen = _enforced(db)
|
||||
vorgabe = _load(db, None)
|
||||
|
||||
if erzwungen and vorgabe is not None:
|
||||
return DashboardLayoutOut(layout=vorgabe, source="default", enforced=True, has_default=True)
|
||||
|
||||
eigene = _load(db, user.id)
|
||||
if eigene is not None:
|
||||
return DashboardLayoutOut(
|
||||
layout=eigene, source="user", enforced=erzwungen, has_default=vorgabe is not None
|
||||
)
|
||||
if vorgabe is not None:
|
||||
return DashboardLayoutOut(
|
||||
layout=vorgabe, source="default", enforced=erzwungen, has_default=True
|
||||
)
|
||||
return DashboardLayoutOut(
|
||||
layout=BUILTIN_LAYOUT, source="builtin", enforced=erzwungen, has_default=False
|
||||
)
|
||||
|
||||
|
||||
@router.put("/layout", response_model=DashboardLayoutOut)
|
||||
def put_layout(
|
||||
payload: DashboardLayoutIn,
|
||||
db: Session = Depends(get_db),
|
||||
user: User = Depends(get_current_user),
|
||||
) -> DashboardLayoutOut:
|
||||
if _enforced(db) and _load(db, None) is not None:
|
||||
raise HTTPException(
|
||||
status.HTTP_403_FORBIDDEN,
|
||||
"Die Startseite ist vom Administrator fest vorgegeben.",
|
||||
)
|
||||
_store(db, user.id, payload.layout)
|
||||
return get_layout(db=db, user=user)
|
||||
|
||||
|
||||
@router.delete("/layout", status_code=status.HTTP_204_NO_CONTENT)
|
||||
def reset_layout(
|
||||
db: Session = Depends(get_db), user: User = Depends(get_current_user)
|
||||
) -> None:
|
||||
"""Eigene Anordnung verwerfen – danach gilt wieder die Vorgabe."""
|
||||
row = db.query(DashboardLayout).filter(DashboardLayout.user_id == user.id).first()
|
||||
if row is not None:
|
||||
db.delete(row)
|
||||
db.commit()
|
||||
|
||||
|
||||
@router.put("/layout/default", response_model=DashboardLayoutOut)
|
||||
def put_default_layout(
|
||||
payload: DashboardLayoutIn,
|
||||
db: Session = Depends(get_db),
|
||||
admin: User = Depends(require_admin),
|
||||
) -> DashboardLayoutOut:
|
||||
_store(db, None, payload.layout)
|
||||
return get_layout(db=db, user=admin)
|
||||
|
||||
|
||||
@router.put("/layout/enforced", response_model=DashboardLayoutOut)
|
||||
def set_enforced(
|
||||
value: bool,
|
||||
db: Session = Depends(get_db),
|
||||
admin: User = Depends(require_admin),
|
||||
) -> DashboardLayoutOut:
|
||||
row = db.get(Setting, ENFORCED_KEY)
|
||||
if row is None:
|
||||
db.add(Setting(key=ENFORCED_KEY, value="1" if value else "0"))
|
||||
else:
|
||||
row.value = "1" if value else "0"
|
||||
db.commit()
|
||||
return get_layout(db=db, user=admin)
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# Auswertungen
|
||||
# --------------------------------------------------------------------------
|
||||
def _zustand(best_before: date | None, heute: date, warnfrist: int) -> str:
|
||||
"""Ablaufzustand einer Charge: ohne MHD / abgelaufen / bald / in Ordnung."""
|
||||
if best_before is None:
|
||||
return "no_date"
|
||||
if best_before < heute:
|
||||
return "expired"
|
||||
if (best_before - heute).days <= warnfrist:
|
||||
return "soon"
|
||||
return "ok"
|
||||
|
||||
|
||||
def _lots_mit_produkt(db: Session) -> list[tuple[Lot, Product]]:
|
||||
return (
|
||||
db.query(Lot, Product)
|
||||
.join(Product, Lot.product_id == Product.id)
|
||||
.filter(Lot.quantity > 0)
|
||||
.all()
|
||||
)
|
||||
|
||||
|
||||
@router.get("/stats", response_model=DashboardStats)
|
||||
def stats(
|
||||
db: Session = Depends(get_db), _: User = Depends(get_current_user)
|
||||
) -> DashboardStats:
|
||||
heute = date.today()
|
||||
warnfrist = get_expiry_warning_days(db)
|
||||
|
||||
einheiten = 0.0
|
||||
bestand_produkte: set[int] = set()
|
||||
bald = abgelaufen = 0
|
||||
for lot, product in _lots_mit_produkt(db):
|
||||
faktor, _label = article_unit(product)
|
||||
einheiten += lot.quantity / faktor if faktor else 0.0
|
||||
bestand_produkte.add(product.id)
|
||||
zustand = _zustand(lot.best_before, heute, warnfrist)
|
||||
if zustand == "expired":
|
||||
abgelaufen += 1
|
||||
elif zustand == "soon":
|
||||
bald += 1
|
||||
|
||||
# Einkaufsbedarf: Produkte und Gruppen unter Mindestbestand.
|
||||
bedarf = 0
|
||||
for product in db.query(Product).filter(Product.min_stock.isnot(None), Product.min_stock > 0):
|
||||
if current_stock(db, product.id) < product.min_stock:
|
||||
bedarf += 1
|
||||
for group in db.query(Group).filter(Group.min_stock.isnot(None), Group.min_stock > 0):
|
||||
unit = group.min_stock_unit
|
||||
produkte = list(group.products)
|
||||
summe = float(sum(current_stock(db, p.id) for p in produkte))
|
||||
if unit is not None:
|
||||
summe /= unit.factor or 1.0
|
||||
if summe < group.min_stock:
|
||||
bedarf += 1
|
||||
|
||||
return DashboardStats(
|
||||
products_in_stock=len(bestand_produkte),
|
||||
article_units=round(einheiten, 3),
|
||||
expiring_soon=bald,
|
||||
expired=abgelaufen,
|
||||
shopping_items=bedarf,
|
||||
products_total=db.query(Product).count(),
|
||||
)
|
||||
|
||||
|
||||
@router.get("/expiry-split", response_model=ExpirySplit)
|
||||
def expiry_split(
|
||||
db: Session = Depends(get_db), _: User = Depends(get_current_user)
|
||||
) -> ExpirySplit:
|
||||
heute = date.today()
|
||||
warnfrist = get_expiry_warning_days(db)
|
||||
summen = {"ok": 0.0, "soon": 0.0, "expired": 0.0, "no_date": 0.0}
|
||||
for lot, product in _lots_mit_produkt(db):
|
||||
faktor, _label = article_unit(product)
|
||||
summen[_zustand(lot.best_before, heute, warnfrist)] += lot.quantity / (faktor or 1.0)
|
||||
return ExpirySplit(**{k: round(v, 3) for k, v in summen.items()})
|
||||
|
||||
|
||||
@router.get("/by-category", response_model=list[CategoryShare])
|
||||
def by_category(
|
||||
db: Session = Depends(get_db), _: User = Depends(get_current_user)
|
||||
) -> list[CategoryShare]:
|
||||
"""Artikeleinheiten je Kategorie, zusätzlich nach Ablaufzustand aufgeteilt."""
|
||||
heute = date.today()
|
||||
warnfrist = get_expiry_warning_days(db)
|
||||
namen = {c.id: c.name for c in db.query(Category).all()}
|
||||
|
||||
leer = lambda: {"article_units": 0.0, "ok": 0.0, "soon": 0.0, "expired": 0.0, "no_date": 0.0}
|
||||
eimer: dict[int | None, dict] = defaultdict(leer)
|
||||
for lot, product in _lots_mit_produkt(db):
|
||||
faktor, _label = article_unit(product)
|
||||
menge = lot.quantity / (faktor or 1.0)
|
||||
topf = eimer[product.category_id]
|
||||
topf["article_units"] += menge
|
||||
topf[_zustand(lot.best_before, heute, warnfrist)] += menge
|
||||
|
||||
ergebnis = [
|
||||
CategoryShare(
|
||||
category_id=cid,
|
||||
name=namen.get(cid, "Ohne Kategorie") if cid is not None else "Ohne Kategorie",
|
||||
**{k: round(v, 3) for k, v in werte.items()},
|
||||
)
|
||||
for cid, werte in eimer.items()
|
||||
]
|
||||
ergebnis.sort(key=lambda c: c.article_units, reverse=True)
|
||||
return ergebnis
|
||||
|
||||
|
||||
def _signiert(movement: Movement) -> float:
|
||||
"""Bewegung als vorzeichenbehaftete Änderung des Bestands."""
|
||||
if movement.type == MovementType.in_:
|
||||
return movement.quantity
|
||||
if movement.type == MovementType.out:
|
||||
return -movement.quantity
|
||||
return movement.quantity # Korrekturen sind bereits vorzeichenbehaftet
|
||||
|
||||
|
||||
@router.get("/timeline", response_model=list[TimelinePoint])
|
||||
def timeline(
|
||||
days: int = 90,
|
||||
product_id: int | None = None,
|
||||
db: Session = Depends(get_db),
|
||||
_: User = Depends(get_current_user),
|
||||
) -> list[TimelinePoint]:
|
||||
"""Artikeleinheiten im Bestand je Tag – rückwärts aus den Bewegungen.
|
||||
|
||||
Ausgangspunkt ist der heutige Bestand; für jeden Tag rückwärts wird die
|
||||
Netto-Bewegung dieses Tages wieder herausgerechnet.
|
||||
|
||||
Bekannte Ungenauigkeit: Die Umrechnung in Artikeleinheiten nutzt die *heutige*
|
||||
Packungsgröße. Wird sie später geändert, verschiebt sich auch die Historie.
|
||||
"""
|
||||
days = max(2, min(days, 730))
|
||||
heute = date.today()
|
||||
|
||||
produkte = db.query(Product)
|
||||
if product_id is not None:
|
||||
produkte = produkte.filter(Product.id == product_id)
|
||||
produkte = produkte.all()
|
||||
if not produkte:
|
||||
return []
|
||||
|
||||
faktoren = {p.id: (article_unit(p)[0] or 1.0) for p in produkte}
|
||||
bestand = {p.id: current_stock(db, p.id) for p in produkte}
|
||||
|
||||
beginn = datetime.combine(heute - timedelta(days=days - 1), datetime.min.time(), tzinfo=timezone.utc)
|
||||
bewegungen = db.query(Movement).filter(Movement.created_at >= beginn)
|
||||
if product_id is not None:
|
||||
bewegungen = bewegungen.filter(Movement.product_id == product_id)
|
||||
|
||||
delta: dict[date, dict[int, float]] = defaultdict(lambda: defaultdict(float))
|
||||
for m in bewegungen.all():
|
||||
if m.product_id in faktoren:
|
||||
delta[m.created_at.date()][m.product_id] += _signiert(m)
|
||||
|
||||
def summe() -> float:
|
||||
return round(sum(bestand[pid] / faktoren[pid] for pid in bestand), 3)
|
||||
|
||||
punkte = [TimelinePoint(date=heute, article_units=summe())]
|
||||
tag = heute
|
||||
for _ in range(days - 1):
|
||||
for pid, wert in delta.get(tag, {}).items():
|
||||
bestand[pid] = bestand.get(pid, 0.0) - wert
|
||||
tag = tag - timedelta(days=1)
|
||||
punkte.append(TimelinePoint(date=tag, article_units=summe()))
|
||||
|
||||
punkte.reverse()
|
||||
return punkte
|
||||
|
||||
|
||||
@router.get("/activity", response_model=list[ActivityPoint])
|
||||
def activity(
|
||||
days: int = 30,
|
||||
db: Session = Depends(get_db),
|
||||
_: User = Depends(get_current_user),
|
||||
) -> list[ActivityPoint]:
|
||||
"""Anzahl der Ein- und Auslagerungen je Tag."""
|
||||
days = max(2, min(days, 365))
|
||||
heute = date.today()
|
||||
beginn = datetime.combine(heute - timedelta(days=days - 1), datetime.min.time(), tzinfo=timezone.utc)
|
||||
|
||||
ein: dict[date, int] = defaultdict(int)
|
||||
aus: dict[date, int] = defaultdict(int)
|
||||
for m in db.query(Movement).filter(Movement.created_at >= beginn).all():
|
||||
tag = m.created_at.date()
|
||||
if m.type == MovementType.in_:
|
||||
ein[tag] += 1
|
||||
elif m.type == MovementType.out:
|
||||
aus[tag] += 1
|
||||
|
||||
return [
|
||||
ActivityPoint(
|
||||
date=heute - timedelta(days=days - 1 - i),
|
||||
checked_in=ein.get(heute - timedelta(days=days - 1 - i), 0),
|
||||
checked_out=aus.get(heute - timedelta(days=days - 1 - i), 0),
|
||||
)
|
||||
for i in range(days)
|
||||
]
|
||||
@@ -386,3 +386,56 @@ class MovementOut(BaseModel):
|
||||
class SettingOut(BaseModel):
|
||||
key: str
|
||||
value: str
|
||||
|
||||
|
||||
# ---- Startseite ----
|
||||
class DashboardLayoutIn(BaseModel):
|
||||
"""Anordnung der Karten. Die Felder entsprechen react-grid-layout."""
|
||||
layout: list[dict]
|
||||
|
||||
|
||||
class DashboardLayoutOut(BaseModel):
|
||||
layout: list[dict]
|
||||
# Woher die Anordnung stammt: eigene, Admin-Vorgabe oder eingebaut.
|
||||
source: str
|
||||
# Ist die Vorgabe verbindlich, darf der Benutzer nicht umbauen.
|
||||
enforced: bool
|
||||
has_default: bool
|
||||
|
||||
|
||||
class DashboardStats(BaseModel):
|
||||
products_in_stock: int # Artikel mit Bestand > 0
|
||||
article_units: float # Summe der Artikeleinheiten
|
||||
expiring_soon: int # Chargen innerhalb der Warnfrist (nicht abgelaufen)
|
||||
expired: int # bereits abgelaufene Chargen
|
||||
shopping_items: int # Produkte + Gruppen unter Mindestbestand
|
||||
products_total: int
|
||||
|
||||
|
||||
class ExpirySplit(BaseModel):
|
||||
"""Artikeleinheiten je Ablaufzustand – Grundlage für Ring und Säulen."""
|
||||
ok: float
|
||||
soon: float
|
||||
expired: float
|
||||
no_date: float # Chargen ohne MHD zählen separat, nicht als "ok"
|
||||
|
||||
|
||||
class CategoryShare(BaseModel):
|
||||
category_id: int | None
|
||||
name: str
|
||||
article_units: float
|
||||
ok: float
|
||||
soon: float
|
||||
expired: float
|
||||
no_date: float
|
||||
|
||||
|
||||
class TimelinePoint(BaseModel):
|
||||
date: date
|
||||
article_units: float
|
||||
|
||||
|
||||
class ActivityPoint(BaseModel):
|
||||
date: date
|
||||
checked_in: int
|
||||
checked_out: int
|
||||
|
||||
@@ -88,3 +88,23 @@ def display_unit_info(product: Product) -> tuple[str, float]:
|
||||
if product.display_unit is not None:
|
||||
return product.display_unit.name, product.display_unit.factor
|
||||
return BASE_LABEL[product.base_unit], 1.0
|
||||
|
||||
|
||||
def article_unit(product: Product) -> tuple[float, str]:
|
||||
"""Faktor und Bezeichnung der *Artikeleinheit* – das zählbare Gebinde.
|
||||
|
||||
Gibt es eine Packungsgröße, ist eine Einheit ein Gebinde (Glas, Packung …),
|
||||
sonst die Anzeigeeinheit des Produkts. Nur diese Größe lässt sich über
|
||||
verschiedene Artikel hinweg sinnvoll addieren: 3 Gläser + 2 Packungen sind
|
||||
5 Einheiten, während 1000 g + 2 Stück keine sinnvolle Summe ergeben.
|
||||
"""
|
||||
if product.package_size and product.package_size > 0:
|
||||
return product.package_size, (product.package_label or "Packung")
|
||||
name, factor = display_unit_info(product)
|
||||
return (factor or 1.0), name
|
||||
|
||||
|
||||
def article_units(product: Product, quantity_base: float) -> float:
|
||||
"""Rechnet eine Menge in Basiseinheiten in Artikeleinheiten um."""
|
||||
factor, _ = article_unit(product)
|
||||
return quantity_base / factor if factor else quantity_base
|
||||
|
||||
187
backend/tests/test_dashboard.py
Normal file
187
backend/tests/test_dashboard.py
Normal file
@@ -0,0 +1,187 @@
|
||||
from datetime import date, timedelta
|
||||
|
||||
import pytest
|
||||
|
||||
from app.models import BaseUnit, Category, Product, Role, User
|
||||
from app.routers import dashboard
|
||||
from app.services.conversion import article_unit, article_units
|
||||
from app.services.stock import check_in, check_out
|
||||
|
||||
|
||||
@pytest.fixture()
|
||||
def user(db):
|
||||
person = User(username="tester", password_hash="x", role=Role.admin)
|
||||
db.add(person)
|
||||
db.commit()
|
||||
db.refresh(person)
|
||||
return person
|
||||
|
||||
|
||||
# ---------------------------------------------------------------- Einheiten
|
||||
def test_artikeleinheit_ist_das_gebinde(rice):
|
||||
# rice: 500 g je Packung -> eine Artikeleinheit sind 500 g
|
||||
faktor, label = article_unit(rice)
|
||||
assert faktor == 500
|
||||
assert label == "Packung"
|
||||
assert article_units(rice, 1500) == 3
|
||||
|
||||
|
||||
def test_artikeleinheit_ohne_gebinde_ist_die_produkteinheit(db):
|
||||
eier = Product(name="Eier", base_unit=BaseUnit.piece)
|
||||
db.add(eier)
|
||||
db.commit()
|
||||
faktor, _ = article_unit(eier)
|
||||
assert faktor == 1
|
||||
assert article_units(eier, 6) == 6
|
||||
|
||||
|
||||
def test_gebinde_bezeichnung_wird_uebernommen(db):
|
||||
pesto = Product(name="Pesto", base_unit=BaseUnit.gram, package_size=195,
|
||||
package_label="Glas")
|
||||
db.add(pesto)
|
||||
db.commit()
|
||||
assert article_unit(pesto) == (195, "Glas")
|
||||
|
||||
|
||||
# ------------------------------------------------------------ Ablaufzustand
|
||||
def test_zustand_unterscheidet_die_vier_faelle():
|
||||
heute = date(2026, 7, 1)
|
||||
assert dashboard._zustand(None, heute, 7) == "no_date"
|
||||
assert dashboard._zustand(heute - timedelta(days=1), heute, 7) == "expired"
|
||||
assert dashboard._zustand(heute + timedelta(days=3), heute, 7) == "soon"
|
||||
assert dashboard._zustand(heute, heute, 7) == "soon" # heute ist noch nicht abgelaufen
|
||||
assert dashboard._zustand(heute + timedelta(days=30), heute, 7) == "ok"
|
||||
|
||||
|
||||
def test_expiry_split_zaehlt_in_artikeleinheiten(db, rice, user):
|
||||
heute = date.today()
|
||||
check_in(db, rice, 2, "package", heute + timedelta(days=60), None, user) # ok
|
||||
check_in(db, rice, 1, "package", heute + timedelta(days=3), None, user) # bald
|
||||
check_in(db, rice, 3, "package", heute - timedelta(days=2), None, user) # abgelaufen
|
||||
check_in(db, rice, 4, "package", None, None, user) # ohne MHD
|
||||
db.commit()
|
||||
|
||||
split = dashboard.expiry_split(db=db, _=user)
|
||||
assert split.ok == 2
|
||||
assert split.soon == 1
|
||||
assert split.expired == 3
|
||||
assert split.no_date == 4
|
||||
|
||||
|
||||
def test_by_category_bündelt_und_sortiert(db, user):
|
||||
suess = Category(name="Süßwaren")
|
||||
db.add(suess)
|
||||
db.commit()
|
||||
|
||||
chips = Product(name="Chips", base_unit=BaseUnit.gram, package_size=175,
|
||||
package_label="Tüte", category_id=suess.id)
|
||||
ohne = Product(name="Salz", base_unit=BaseUnit.gram, package_size=500)
|
||||
db.add_all([chips, ohne])
|
||||
db.commit()
|
||||
|
||||
check_in(db, chips, 5, "package", None, None, user)
|
||||
check_in(db, ohne, 2, "package", None, None, user)
|
||||
db.commit()
|
||||
|
||||
anteile = dashboard.by_category(db=db, _=user)
|
||||
assert [a.name for a in anteile] == ["Süßwaren", "Ohne Kategorie"]
|
||||
assert anteile[0].article_units == 5
|
||||
assert anteile[1].article_units == 2
|
||||
|
||||
|
||||
# ----------------------------------------------------------------- Verlauf
|
||||
def test_timeline_rechnet_bewegungen_zurueck(db, rice, user):
|
||||
"""Der Verlauf entsteht rueckwaerts aus dem heutigen Bestand."""
|
||||
check_in(db, rice, 4, "package", None, None, user) # +4 Einheiten heute
|
||||
check_out(db, rice, 1, "package", user) # -1 Einheit heute
|
||||
db.commit()
|
||||
|
||||
punkte = dashboard.timeline(days=3, product_id=None, db=db, _=user)
|
||||
assert len(punkte) == 3
|
||||
# Heute: 3 Einheiten; die Tage davor lag nichts im Lager.
|
||||
assert punkte[-1].date == date.today()
|
||||
assert punkte[-1].article_units == 3
|
||||
assert punkte[0].article_units == 0
|
||||
assert punkte[1].article_units == 0
|
||||
|
||||
|
||||
def test_timeline_kann_auf_ein_produkt_eingegrenzt_werden(db, rice, user):
|
||||
anderes = Product(name="Zucker", base_unit=BaseUnit.gram, package_size=1000)
|
||||
db.add(anderes)
|
||||
db.commit()
|
||||
check_in(db, rice, 2, "package", None, None, user)
|
||||
check_in(db, anderes, 5, "package", None, None, user)
|
||||
db.commit()
|
||||
|
||||
gesamt = dashboard.timeline(days=2, product_id=None, db=db, _=user)
|
||||
nur_reis = dashboard.timeline(days=2, product_id=rice.id, db=db, _=user)
|
||||
assert gesamt[-1].article_units == 7
|
||||
assert nur_reis[-1].article_units == 2
|
||||
|
||||
|
||||
def test_activity_zaehlt_ein_und_auslagerungen(db, rice, user):
|
||||
check_in(db, rice, 2, "package", None, None, user)
|
||||
check_in(db, rice, 1, "package", None, None, user)
|
||||
check_out(db, rice, 1, "package", user)
|
||||
db.commit()
|
||||
|
||||
tage = dashboard.activity(days=3, db=db, _=user)
|
||||
assert len(tage) == 3
|
||||
heute = tage[-1]
|
||||
assert heute.date == date.today()
|
||||
assert heute.checked_in == 2
|
||||
assert heute.checked_out == 1
|
||||
|
||||
|
||||
# ---------------------------------------------------------------- Anordnung
|
||||
def test_layout_faellt_auf_die_eingebaute_zurueck(db, user):
|
||||
antwort = dashboard.get_layout(db=db, user=user)
|
||||
assert antwort.source == "builtin"
|
||||
assert antwort.enforced is False
|
||||
assert antwort.has_default is False
|
||||
assert any(k["i"] == "expiring" for k in antwort.layout)
|
||||
|
||||
|
||||
def test_eigene_anordnung_wird_gespeichert(db, user):
|
||||
from app.schemas import DashboardLayoutIn
|
||||
|
||||
eigen = [{"i": "shopping", "x": 0, "y": 0, "w": 6, "h": 3}]
|
||||
dashboard.put_layout(DashboardLayoutIn(layout=eigen), db=db, user=user)
|
||||
|
||||
antwort = dashboard.get_layout(db=db, user=user)
|
||||
assert antwort.source == "user"
|
||||
assert antwort.layout == eigen
|
||||
|
||||
|
||||
def test_erzwungene_vorgabe_verhindert_eigene_anordnung(db, user):
|
||||
from fastapi import HTTPException
|
||||
|
||||
from app.schemas import DashboardLayoutIn
|
||||
|
||||
vorgabe = [{"i": "status", "x": 0, "y": 0, "w": 12, "h": 1}]
|
||||
dashboard.put_default_layout(DashboardLayoutIn(layout=vorgabe), db=db, admin=user)
|
||||
dashboard.set_enforced(True, db=db, admin=user)
|
||||
|
||||
antwort = dashboard.get_layout(db=db, user=user)
|
||||
assert antwort.source == "default"
|
||||
assert antwort.enforced is True
|
||||
assert antwort.layout == vorgabe
|
||||
|
||||
with pytest.raises(HTTPException) as fehler:
|
||||
dashboard.put_layout(DashboardLayoutIn(layout=[]), db=db, user=user)
|
||||
assert fehler.value.status_code == 403
|
||||
|
||||
|
||||
def test_zuruecksetzen_stellt_die_vorgabe_wieder_her(db, user):
|
||||
from app.schemas import DashboardLayoutIn
|
||||
|
||||
vorgabe = [{"i": "status", "x": 0, "y": 0, "w": 12, "h": 1}]
|
||||
dashboard.put_default_layout(DashboardLayoutIn(layout=vorgabe), db=db, admin=user)
|
||||
dashboard.put_layout(DashboardLayoutIn(layout=[{"i": "shopping", "x": 0, "y": 0, "w": 4, "h": 3}]),
|
||||
db=db, user=user)
|
||||
assert dashboard.get_layout(db=db, user=user).source == "user"
|
||||
|
||||
dashboard.reset_layout(db=db, user=user)
|
||||
antwort = dashboard.get_layout(db=db, user=user)
|
||||
assert antwort.source == "default"
|
||||
assert antwort.layout == vorgabe
|
||||
Reference in New Issue
Block a user