a70a99b387
Auswertungen (by_category, expiry_split, stats) iterierten nur Lots. Gegenstaende mit Menge-je-Lagerort liegen als Lots und tauchten auf – Einzelstuecke liegen aber als Items und fehlten dadurch komplett (Dashboard zeigte z.B. nur die Powerbank, nicht Kamera/Kabel). Neuer Helfer _bestand_beitraege deckt beide Speicherformen ab: jedes Item zaehlt als ein Stueck ohne MHD. Regressionstest ergaenzt; volle Suite gruen (120). CategorySelect: Scrollen IM aufklappbaren Menue schloss es, weil jeder Scroll als Seiten-Scroll gewertet wurde. Jetzt schliesst nur ein Scroll ausserhalb des Menues; die lange Liste laesst sich scrollen. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
334 lines
13 KiB
Python
334 lines
13 KiB
Python
from datetime import date, timedelta # noqa: F401 (date fuer die Zustands-Tests)
|
||
|
||
import pytest
|
||
|
||
from app.models import BaseUnit, Category, Item, Product, Role, User
|
||
from app.routers import dashboard
|
||
from app.services.conversion import article_unit, article_units
|
||
from app.services.items import generate_uid
|
||
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
|
||
|
||
|
||
def test_einzelstuecke_zaehlen_in_auswertungen(db, user):
|
||
"""Regression: Einzelstueck-Produkte liegen als Items, nicht als Lots. Sie
|
||
muessen trotzdem in Kategorie-, Ablauf- und Bestandsauswertung erscheinen –
|
||
sonst zeigte das Dashboard nur die Mengen-Gegenstaende (Bug: nur Powerbank)."""
|
||
kamera_kat = Category(name="Kamera", tracking="object")
|
||
db.add(kamera_kat)
|
||
db.commit()
|
||
kamera = Product(name="Alpha 7 IV", category_id=kamera_kat.id, individual=True)
|
||
db.add(kamera)
|
||
db.commit()
|
||
for _ in range(2):
|
||
db.add(Item(uid=generate_uid(db), product_id=kamera.id))
|
||
db.commit()
|
||
|
||
anteil = next(a for a in dashboard.by_category(db=db, _=user) if a.name == "Kamera")
|
||
assert anteil.article_units == 2
|
||
assert anteil.no_date == 2 # Einzelstuecke haben kein MHD
|
||
|
||
assert dashboard.expiry_split(db=db, _=user).no_date == 2
|
||
|
||
st = dashboard.stats(db=db, _=user)
|
||
assert st.article_units == 2
|
||
assert st.products_in_stock == 1
|
||
|
||
|
||
# ----------------------------------------------------------------- Verlauf
|
||
def test_schrittweite_haengt_am_zeitraum():
|
||
"""Kurze Zeitraeume werden fein abgetastet, lange grob."""
|
||
assert dashboard._schrittweite(1) == timedelta(hours=1)
|
||
assert dashboard._schrittweite(2) == timedelta(hours=1)
|
||
assert dashboard._schrittweite(7) == timedelta(hours=6)
|
||
assert dashboard._schrittweite(90) == timedelta(days=1)
|
||
assert dashboard._schrittweite(365) == timedelta(days=7)
|
||
|
||
|
||
def test_schrittweite_haelt_die_punktzahl_im_rahmen():
|
||
"""Egal welcher Zeitraum – die Kurve bleibt zeichenbar."""
|
||
for tage in (1, 2, 7, 14, 30, 90, 180, 365, 730):
|
||
anzahl = timedelta(days=tage) / dashboard._schrittweite(tage)
|
||
assert 20 <= anzahl <= 180, f"{tage} Tage ergeben {anzahl} Punkte"
|
||
|
||
|
||
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 jetzt
|
||
check_out(db, rice, 1, "package", user) # -1 Einheit jetzt
|
||
db.commit()
|
||
|
||
punkte = dashboard.timeline(days=3, product_id=None, db=db, _=user)
|
||
# Jetzt 3 Einheiten; vor dem Einlagern war das Lager leer.
|
||
assert punkte[-1].article_units == 3
|
||
assert punkte[0].article_units == 0
|
||
# Aufsteigend sortiert, damit die Linie von links nach rechts laeuft.
|
||
assert punkte[0].at < punkte[-1].at
|
||
|
||
|
||
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_flow_liefert_eine_geschlossene_bruecke(db, rice, user):
|
||
"""Anfangsbestand + Zugang - Abgang muss den Endbestand ergeben."""
|
||
check_in(db, rice, 5, "package", None, None, user)
|
||
check_out(db, rice, 2, "package", user)
|
||
db.commit()
|
||
|
||
abschnitte = dashboard.flow(days=3, product_id=None, db=db, _=user)
|
||
juengster = abschnitte[-1]
|
||
assert juengster.opening == 0 # vorher war das Lager leer
|
||
assert juengster.checked_in == 5 # Artikeleinheiten, nicht Vorgaenge
|
||
assert juengster.checked_out == 2
|
||
assert juengster.opening + juengster.checked_in - juengster.checked_out == 3
|
||
|
||
|
||
def test_flow_schliesst_ohne_luecke_an_den_vorherigen_abschnitt_an(db, rice, user):
|
||
"""Der Endbestand eines Abschnitts ist der Anfangsbestand des naechsten –
|
||
sonst haette die Bruecke einen Sprung, den kein Balken erklaert."""
|
||
check_in(db, rice, 4, "package", None, None, user)
|
||
db.commit()
|
||
|
||
abschnitte = dashboard.flow(days=7, product_id=None, db=db, _=user)
|
||
for vorher, nachher in zip(abschnitte, abschnitte[1:]):
|
||
schluss = vorher.opening + vorher.checked_in - vorher.checked_out
|
||
assert round(schluss, 3) == round(nachher.opening, 3)
|
||
|
||
|
||
def test_flow_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()
|
||
|
||
nur_reis = dashboard.flow(days=2, product_id=rice.id, db=db, _=user)
|
||
assert nur_reis[-1].checked_in == 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()
|
||
|
||
abschnitte = dashboard.activity(days=3, db=db, _=user)
|
||
# Alles gerade eben passiert -> liegt im juengsten Abschnitt.
|
||
juengster = abschnitte[-1]
|
||
assert juengster.checked_in == 2
|
||
assert juengster.checked_out == 1
|
||
assert sum(a.checked_in for a in abschnitte) == 2
|
||
assert abschnitte[0].at < juengster.at
|
||
|
||
|
||
# ---------------------------------------------------------------- Anordnung
|
||
def _erstes(antwort):
|
||
return antwort.dashboards[0]
|
||
|
||
|
||
def test_layout_faellt_auf_die_eingebaute_zurueck(db, user):
|
||
antwort = dashboard.list_dashboards(db=db, user=user)
|
||
assert antwort.source == "builtin"
|
||
assert antwort.enforced is False
|
||
assert antwort.has_default is False
|
||
assert len(antwort.dashboards) == 1
|
||
assert any(k["type"] == "expiry-all" for k in _erstes(antwort).layout)
|
||
|
||
|
||
def test_eigene_anordnung_wird_gespeichert(db, user):
|
||
from app.schemas import DashboardCreate, DashboardUpdate
|
||
|
||
neu = dashboard.create_dashboard(
|
||
DashboardCreate(name="Vorräte", layout=[]), db=db, user=user
|
||
)
|
||
eigen = [{"i": "k1", "type": "shopping", "x": 0, "y": 0, "w": 6, "h": 3, "props": {}}]
|
||
dashboard.update_dashboard(neu.id, DashboardUpdate(layout=eigen), db=db, user=user)
|
||
|
||
antwort = dashboard.list_dashboards(db=db, user=user)
|
||
assert antwort.source == "user"
|
||
assert _erstes(antwort).name == "Vorräte"
|
||
assert _erstes(antwort).layout == eigen
|
||
|
||
|
||
def test_dieselbe_kartenart_mehrfach(db, user):
|
||
"""Der eigentliche Zweck der Instanz-Kennung: zwei Karten derselben Art,
|
||
jede auf einen anderen Artikel eingestellt."""
|
||
from app.schemas import DashboardCreate
|
||
|
||
layout = [
|
||
{"i": "a", "type": "product-timeline", "x": 0, "y": 0, "w": 6, "h": 6,
|
||
"props": {"product_id": 1}},
|
||
{"i": "b", "type": "product-timeline", "x": 6, "y": 0, "w": 6, "h": 6,
|
||
"props": {"product_id": 2}},
|
||
]
|
||
neu = dashboard.create_dashboard(
|
||
DashboardCreate(name="Beobachtung", layout=layout), db=db, user=user
|
||
)
|
||
assert len(neu.layout) == 2
|
||
assert {k["props"]["product_id"] for k in neu.layout} == {1, 2}
|
||
|
||
|
||
def test_alte_anordnung_bekommt_die_art_aus_der_kennung(db):
|
||
"""Gespeicherte Anordnungen von frueher haben kein "type" - dort war die
|
||
Kennung die Art. Ohne das Nachziehen waere jede alte Karte unbekannt."""
|
||
alt = [{"i": "shopping", "x": 0, "y": 0, "w": 6, "h": 3}]
|
||
normalisiert = dashboard._normalize(alt)
|
||
assert normalisiert[0]["type"] == "shopping"
|
||
assert normalisiert[0]["props"] == {}
|
||
|
||
|
||
def test_erzwungene_vorgabe_verhindert_eigene_anordnung(db, user):
|
||
from fastapi import HTTPException
|
||
|
||
from app.schemas import DashboardCreate
|
||
|
||
dashboard.create_dashboard(DashboardCreate(name="Vorgabe", layout=[]), db=db, user=user)
|
||
dashboard.put_default_dashboards(db=db, admin=user)
|
||
dashboard.set_enforced(True, db=db, admin=user)
|
||
|
||
antwort = dashboard.list_dashboards(db=db, user=user)
|
||
assert antwort.source == "default"
|
||
assert antwort.enforced is True
|
||
|
||
with pytest.raises(HTTPException) as fehler:
|
||
dashboard.create_dashboard(DashboardCreate(name="X", 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 DashboardCreate
|
||
|
||
dashboard.create_dashboard(DashboardCreate(name="Vorgabe", layout=[]), db=db, user=user)
|
||
dashboard.put_default_dashboards(db=db, admin=user)
|
||
dashboard.create_dashboard(DashboardCreate(name="Eigenes", layout=[]), db=db, user=user)
|
||
assert dashboard.list_dashboards(db=db, user=user).source == "user"
|
||
|
||
dashboard.reset_dashboards(db=db, user=user)
|
||
antwort = dashboard.list_dashboards(db=db, user=user)
|
||
assert antwort.source == "default"
|
||
assert _erstes(antwort).name == "Vorgabe"
|
||
|
||
|
||
def test_erstes_eigenes_uebernimmt_die_vorgabe(db, user):
|
||
"""Sonst verschwaende das Anlegen eines zweiten Dashboards die Vorgabe:
|
||
Sobald eigene Zeilen existieren, zaehlen nur noch die."""
|
||
from app.schemas import DashboardCreate
|
||
|
||
dashboard.create_dashboard(DashboardCreate(name="Vorgabe", layout=[]), db=db, user=user)
|
||
dashboard.put_default_dashboards(db=db, admin=user)
|
||
dashboard.reset_dashboards(db=db, user=user)
|
||
assert dashboard.list_dashboards(db=db, user=user).source == "default"
|
||
|
||
dashboard.create_dashboard(DashboardCreate(name="Getränke", layout=[]), db=db, user=user)
|
||
antwort = dashboard.list_dashboards(db=db, user=user)
|
||
assert [d.name for d in antwort.dashboards] == ["Vorgabe", "Getränke"]
|
||
|
||
|
||
def test_fremdes_dashboard_ist_unsichtbar(db, user):
|
||
from fastapi import HTTPException
|
||
|
||
from app.models import Role, User
|
||
from app.schemas import DashboardCreate, DashboardUpdate
|
||
|
||
fremd = User(username="andere", password_hash="x", role=Role.user)
|
||
db.add(fremd)
|
||
db.commit()
|
||
db.refresh(fremd)
|
||
|
||
meins = dashboard.create_dashboard(DashboardCreate(name="Meins", layout=[]), db=db, user=user)
|
||
with pytest.raises(HTTPException) as fehler:
|
||
dashboard.update_dashboard(meins.id, DashboardUpdate(name="Geklaut"), db=db, user=fremd)
|
||
assert fehler.value.status_code == 404
|