From 98e2d181428459a3ca62a247fe941dd6f67017cd Mon Sep 17 00:00:00 2001 From: Scarriffle Date: Tue, 20 Oct 2026 21:47:02 +0200 Subject: [PATCH] Verwendungsstatus: Backend fuer "getragen / in Reinigung / in Reparatur" Einzelstuecke hatten bisher keinen Zustand - der Lagerort sagt, wohin ein Stueck gehoert, aber nicht, woran es gerade ist. Neue Tabelle item_usages nach dem Vorbild von item_loans: ein Zeitraum je Zustand, ended_at NULL heisst "laeuft noch", und ein Teilindex laesst nur einen laufenden zu. "Verfuegbar" wird bewusst nicht gespeichert - keine offene Zeile IST verfuegbar, wie bei lots.opened_on und beim Verleih. Damit kostet der Ruhezustand nichts und die Tabelle fuellt sich nicht mit Rauschen. Abgeschlossene Zeilen bleiben stehen, sie sind der Verlauf. Eigener Endpunkt (PUT /items/{id}/verwendung) und ausdruecklich nicht PATCH /items/{id}: Das ist Admins vorbehalten, und die App schickt dort immer alle Felder mitsamt null - jedes Speichern wuerde den Status stillschweigend zuruecksetzen. Umstellen darf deshalb jeder angemeldete Benutzer, wie Verleihen. Der Zustand aendert den Bestand nicht und schreibt keine Bewegung: ein getragenes Hemd ist weiterhin im Haushalt, anders als ein verliehenes. Sichtbar nur bei Kategorien mit usage_tracking. Ein Haken an einer Oberkategorie wandert in den Unterbaum, weil die Garderobe als Kategoriebaum ja schon existiert. Die Markierung fahrt in beiden Sicherungswegen mit; der Verlauf nicht, weil Einzelstuecke dort ueberhaupt nicht vorkommen. Co-Authored-By: Claude Opus 5 --- backend/app/main.py | 8 + backend/app/models.py | 86 ++++++ backend/app/routers/categories.py | 23 +- backend/app/routers/items.py | 7 +- backend/app/routers/transfer.py | 9 +- backend/app/routers/verwendung.py | 104 ++++++++ backend/app/schemas.py | 36 +++ backend/app/services/items.py | 29 ++- backend/app/services/master_data.py | 4 +- backend/tests/test_verwendung.py | 391 ++++++++++++++++++++++++++++ 10 files changed, 691 insertions(+), 6 deletions(-) create mode 100644 backend/app/routers/verwendung.py create mode 100644 backend/tests/test_verwendung.py diff --git a/backend/app/main.py b/backend/app/main.py index a0a6581..4b24790 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -32,6 +32,7 @@ from .routers import ( transfer, units, users, + verwendung, views, ) from .seed import ( @@ -244,6 +245,12 @@ def _ensure_schema() -> None: # Oberkategorie -> globale Einstellung), siehe services/vererbung.py. "ALTER TABLE products ADD COLUMN IF NOT EXISTS open_days INTEGER", "ALTER TABLE categories ADD COLUMN IF NOT EXISTS open_days INTEGER", + # Verwendungsstatus an Einzelstuecken dieser Kategorie (Kleidung, + # Werkzeug). Aus, solange niemand sie setzt - bestehende Kategorien + # aendern sich also nicht. Die Tabelle item_usages selbst kommt aus + # create_all, nur Spalten brauchen diesen Weg hier. + "ALTER TABLE categories ADD COLUMN IF NOT EXISTS usage_tracking " + "BOOLEAN NOT NULL DEFAULT FALSE", ] with engine.begin() as conn: # Zuerst die Lagerort-ID auf den Code umstellen (einmalig, idempotent), @@ -338,3 +345,4 @@ app.include_router(shops.router) app.include_router(field_definitions.router) app.include_router(items.router) app.include_router(loans.router) +app.include_router(verwendung.router) diff --git a/backend/app/models.py b/backend/app/models.py index a013e2d..ab4b597 100644 --- a/backend/app/models.py +++ b/backend/app/models.py @@ -98,6 +98,24 @@ class RemovalReason(str, enum.Enum): other = "other" # sonstiges +class UsageStatus(str, enum.Enum): + """Woran ein Einzelstueck gerade ist (Hemd getragen, Auto in der Wasche). + + Bewusst KEIN Wert fuer "verfuegbar": Fehlt die offene Zeile in + :class:`ItemUsage`, ist das Stueck verfuegbar - wie bei ``lots.opened_on`` + und beim Verleih. Ein eigener Wert daneben koennte ihr widersprechen. + + Absichtlich allgemein und nicht auf Kleidung zugeschnitten: "in_use" traegt + auch eine Bohrmaschine auf der Baustelle. + + Als kurzer String gespeichert (wie CategoryTracking, RemovalReason), damit + sich die Spalte per ADD COLUMN nachziehen laesst. + """ + in_use = "in_use" # wird gerade benutzt/getragen + cleaning = "cleaning" # in der Waesche/Reinigung + repair = "repair" # in Reparatur + + class Unit(Base): """Vom Admin verwaltbare Einheit mit Umrechnungsfaktor zur kanonischen Basis. @@ -353,6 +371,15 @@ class Category(Base): # NULL = erbt von der Oberkategorie, zuletzt vom globalen Wert # (services/vererbung.py::oeffnungsfrist). open_days: Mapped[int | None] = mapped_column(Integer, nullable=True) + # Ob Einzelstuecke dieser Kategorie einen Verwendungsstatus fuehren + # (Kleidung, Werkzeug - nicht Konserven). Wie tracking eine Kopie beim + # Anlegen und keine Aufloesung beim Lesen: So kommt die Markierung ohne + # zusaetzliche Abfrage am Einzelstueck an (services/items.py::item_to_out). + # Ein Haken an einer Oberkategorie wandert beim Speichern in den Unterbaum, + # siehe routers/categories.py::update_category. + usage_tracking: Mapped[bool] = mapped_column( + Boolean, nullable=False, default=False, server_default=text("false") + ) class Product(Base): @@ -1069,6 +1096,12 @@ class Item(Base): back_populates="item", cascade="all, delete-orphan", order_by="ItemLoan.lent_on.desc(), ItemLoan.id.desc()", ) + # Verwendungs-Historie, neueste zuerst. Der laufende Zustand (ended_at NULL) + # ist hoechstens einer, siehe ItemUsage. Keine offene Zeile = verfuegbar. + usages: Mapped[list["ItemUsage"]] = relationship( + back_populates="item", cascade="all, delete-orphan", + order_by="ItemUsage.started_at.desc(), ItemUsage.id.desc()", + ) class ItemDocument(Base): @@ -1141,6 +1174,59 @@ class ItemLoan(Base): user: Mapped[User | None] = relationship() +class ItemUsage(Base): + """Ein Zeitraum, in dem ein Einzelstueck in einem Verwendungszustand war. + + Bewusst eine eigene Tabelle und keine Spalte am Einzelstueck: So bleibt + nachvollziehbar, seit wann ein Hemd in der Waesche liegt und wie oft es + getragen wurde. Abgeschlossene Zeitraeume bleiben stehen - damit ist die + Tabelle gleichzeitig die Historie je Stueck. + + ``ended_at`` NULL heisst "laeuft noch"; ein Stueck kann nur einen solchen + laufenden Zustand haben (Teilindex unten). Es gibt keine Zeile fuer + "verfuegbar" - keine offene Zeile IST verfuegbar. + + Anders als der Verleih aendert der Zustand den Bestand NICHT: ein getragenes + Hemd ist weiterhin im Haushalt. Und wie beim Verleih wird keine Bewegung + (Movement) geschrieben, siehe routers/loans.py. + """ + + __tablename__ = "item_usages" + __table_args__ = ( + # Zwei Zustaende gleichzeitig kann ein Stueck nicht haben. Als Teilindex, + # weil nur die LAUFENDEN eindeutig sein duerfen: abgeschlossene bleiben + # beliebig oft je Stueck stehen. + Index( + "uq_item_usage_offen", + "item_id", + unique=True, + postgresql_where=text("ended_at IS NULL"), + sqlite_where=text("ended_at IS NULL"), + ), + ) + + id: Mapped[int] = mapped_column(Integer, primary_key=True) + item_id: Mapped[int] = mapped_column( + ForeignKey("items.id", ondelete="CASCADE"), nullable=False, index=True + ) + # UsageStatus-Wert als kurzer String (kein Postgres-Enumtyp). + status: Mapped[str] = mapped_column(String(16), nullable=False) + # Zeitstempel und nicht nur ein Datum: Ein Hemd kann an einem Tag getragen, + # gewaschen und wieder eingeraeumt werden. + started_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), nullable=False, default=_now + ) + ended_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True) + # Wer umgestellt hat. SET NULL, damit ein geloeschtes Konto den Vorgang + # nicht mitnimmt. + user_id: Mapped[int | None] = mapped_column( + ForeignKey("users.id", ondelete="SET NULL"), nullable=True + ) + + item: Mapped[Item] = relationship(back_populates="usages") + user: Mapped[User | None] = relationship() + + class ProductLocationMinStock(Base): """Mindestbestand eines Produkts an EINEM Lagerort (in Basiseinheiten). diff --git a/backend/app/routers/categories.py b/backend/app/routers/categories.py index a555a10..30ef373 100644 --- a/backend/app/routers/categories.py +++ b/backend/app/routers/categories.py @@ -85,9 +85,17 @@ def create_category( tracking = parent.tracking else: tracking = CategoryTracking.object.value + # Verwendungsstatus genauso: gewaehlt > vom Elternteil geerbt > aus. + # Eine Unterkategorie von "Kleidung" soll ihn ohne Zutun fuehren. + if payload.usage_tracking is not None: + usage_tracking = payload.usage_tracking + elif parent is not None: + usage_tracking = parent.usage_tracking + else: + usage_tracking = False category = Category( name=payload.name, parent_id=payload.parent_id, tracking=tracking, - open_days=payload.open_days, + open_days=payload.open_days, usage_tracking=usage_tracking, ) db.add(category) db.commit() @@ -125,6 +133,19 @@ def update_category( data["tracking"] = CategoryTracking(data["tracking"]).value for field, value in data.items(): setattr(category, field, value) + # Der Verwendungsstatus wandert in den ganzen Unterbaum. Ohne das erreichte + # ein Haken an einer bestehenden Kategorie ("Kleidung") deren vorhandene + # Unterkategorien nicht - und genau die gibt es ja schon. Bewusst sichtbar: + # Die Haken stehen danach auch dort und lassen sich einzeln wieder abwaehlen. + if data.get("usage_tracking") is not None: + kinder = descendant_ids(db, category_id) + if kinder: + ( + db.query(Category) + .filter(Category.id.in_(kinder)) + .update({"usage_tracking": data["usage_tracking"]}, + synchronize_session=False) + ) db.commit() db.refresh(category) return _to_out(db, category) diff --git a/backend/app/routers/items.py b/backend/app/routers/items.py index 96f93c2..f3d9c6f 100644 --- a/backend/app/routers/items.py +++ b/backend/app/routers/items.py @@ -9,7 +9,7 @@ import re from fastapi import APIRouter, Depends, File, HTTPException, UploadFile, status from fastapi.responses import Response -from sqlalchemy.orm import Session, joinedload +from sqlalchemy.orm import Session, joinedload, selectinload from ..database import get_db from ..deps import get_current_user, require_admin @@ -17,6 +17,7 @@ from ..models import ( Item, ItemDocument, ItemLoan, + ItemUsage, Location, Movement, MovementType, @@ -166,6 +167,10 @@ def list_all_items( # Verleihe samt Konto mitladen: Sonst fragt item_to_out je # verliehenem Stueck einzeln nach (N+1). joinedload(Item.loans).joinedload(ItemLoan.user), + # selectin und nicht joined: Hier haengen schon drei Sammlungen + # dran, eine vierte vervielfachte die Zeilen ueber Kreuz. So kostet + # es eine zusaetzliche Abfrage statt eines Kreuzprodukts. + selectinload(Item.usages).joinedload(ItemUsage.user), ) .order_by(Item.id) .all() diff --git a/backend/app/routers/transfer.py b/backend/app/routers/transfer.py index 19ef529..0220eb7 100644 --- a/backend/app/routers/transfer.py +++ b/backend/app/routers/transfer.py @@ -200,7 +200,11 @@ def export_backup_json( # Kategorien mit Verwaltungsart, damit Lebensmittel/Gegenstände beim # Wiederherstellen erhalten bleiben (auch leere Kategorien). "categories": [ - {"path": _category_path(db, c), "tracking": c.tracking} + { + "path": _category_path(db, c), + "tracking": c.tracking, + "usage_tracking": c.usage_tracking, + } for c in db.query(Category).order_by(Category.id).all() ], # Bezugsquellen (nur für Gegenstände). @@ -963,6 +967,9 @@ def _import_json(db: Session, content: bytes, user: User, mode: str) -> dict: ) if cat is not None and entry.get("tracking"): cat.tracking = entry["tracking"] + # Fehlt der Schluessel (aelteres Backup), bleibt die Markierung aus. + if cat is not None: + cat.usage_tracking = bool(entry.get("usage_tracking", False)) for entry in data.get("shops", []): _get_or_create_shop(db, entry.get("name"), entry.get("website")) db.flush() diff --git a/backend/app/routers/verwendung.py b/backend/app/routers/verwendung.py new file mode 100644 index 0000000..ff61635 --- /dev/null +++ b/backend/app/routers/verwendung.py @@ -0,0 +1,104 @@ +"""Verwendungsstatus von Einzelstücken: umstellen und Verlauf. + +Woran ist das Stück gerade? Getragen, in der Wäsche, in Reparatur – oder +einfach verfügbar. Gedacht für eine Garderobe, aber bewusst allgemein: ein Auto +in der Waschanlage ist genauso „in Reinigung". + +Umstellen ist eine **Buchung**, keine Stammdatenänderung – deshalb darf jeder +angemeldete Benutzer es tun (wie Verleihen, Ein- und Auslagern), und es wird mit +dem Konto protokolliert. Ausdrücklich **nicht** über ``PATCH /items/{id}``: +Diese Route ist Admins vorbehalten, und die App schickt dort immer alle Felder +mitsamt ``null`` (siehe ``ItemUpdateRequest.encode`` in ios/Sources/Models.swift) +– jedes beliebige Speichern würde den Status stillschweigend zurücksetzen. +Verleih und Zubehör haben aus demselben Grund eigene Endpunkte. + +Bewusst **keine** Bewegung (``Movement``), aus demselben Grund wie beim Verleih +(siehe ``routers/loans.py``). Und bewusst **keine** Auswirkung auf den Bestand: +Ein getragenes Hemd ist weiterhin im Haushalt – anders als ein verliehenes. +""" + +from datetime import UTC, datetime + +from fastapi import APIRouter, Depends, HTTPException, status +from sqlalchemy.orm import Session, joinedload + +from ..database import get_db +from ..deps import get_current_user +from ..models import Item, ItemUsage, User +from ..schemas import ItemOut, VerwendungOut, VerwendungSetzen +from ..services.items import item_to_out, verwendung_to_out + +router = APIRouter(tags=["verwendung"]) + + +def _item_or_404(db: Session, item_id: int) -> Item: + item = db.get(Item, item_id) + if item is None: + raise HTTPException(status.HTTP_404_NOT_FOUND, "Einzelstück nicht gefunden") + return item + + +def _laufend(db: Session, item_id: int) -> ItemUsage | None: + """Der laufende Zeitraum dieses Stücks, falls einer offen ist.""" + return ( + db.query(ItemUsage) + .filter(ItemUsage.item_id == item_id, ItemUsage.ended_at.is_(None)) + .first() + ) + + +@router.put("/items/{item_id}/verwendung", response_model=ItemOut) +def setze_verwendung( + item_id: int, + payload: VerwendungSetzen, + db: Session = Depends(get_db), + user: User = Depends(get_current_user), +) -> ItemOut: + """Verwendungszustand setzen. Leerer Status = verfügbar. + + Eine Route für alle Zustände, auch für den Ruhezustand: „verfügbar" ist + nichts anderes als „kein laufender Zeitraum", also beendet der leere Status + nur den aktuellen. Damit ist der Aufruf idempotent. + + Antwortet mit dem ganzen Stück, damit die Oberflächen nach der Buchung nicht + noch einmal laden müssen. + """ + item = _item_or_404(db, item_id) + neu = payload.status.value if payload.status is not None else None + laufend = _laufend(db, item.id) + + if laufend is not None and laufend.status == neu: + # Zweimal dasselbe angetippt – sonst entstünde ein Zeitraum der Länge 0. + return item_to_out(item) + + if laufend is not None: + laufend.ended_at = datetime.now(UTC) + # Vor dem Einfügen schreiben: Sonst darf SQLAlchemy das INSERT vor das + # UPDATE stellen, und dann liegen kurzzeitig zwei offene Zeiträume an – + # der Teilindex uq_item_usage_offen schlägt zu. + db.flush() + + if neu is not None: + db.add(ItemUsage(item_id=item.id, status=neu, user_id=user.id)) + + db.commit() + db.refresh(item) + return item_to_out(item) + + +@router.get("/items/{item_id}/verwendungen", response_model=list[VerwendungOut]) +def list_verwendungen( + item_id: int, + db: Session = Depends(get_db), + _: User = Depends(get_current_user), +) -> list[VerwendungOut]: + """Der ganze Verlauf dieses Stücks, neueste zuerst.""" + _item_or_404(db, item_id) + rows = ( + db.query(ItemUsage) + .options(joinedload(ItemUsage.user)) + .filter(ItemUsage.item_id == item_id) + .order_by(ItemUsage.started_at.desc(), ItemUsage.id.desc()) + .all() + ) + return [verwendung_to_out(u) for u in rows] diff --git a/backend/app/schemas.py b/backend/app/schemas.py index c182f01..e05272d 100644 --- a/backend/app/schemas.py +++ b/backend/app/schemas.py @@ -12,6 +12,7 @@ from .models import ( RemovalReason, Role, UnitKind, + UsageStatus, ) @@ -283,6 +284,8 @@ class CategoryOut(BaseModel): # Maske: Ein leeres Feld soll zeigen, was es erbt, statt nur leer zu sein. inherited_open_days: int = 0 open_days_source: str = "" + # Ob Einzelstuecke dieser Kategorie einen Verwendungsstatus fuehren. + usage_tracking: bool = False class CategoryCreate(BaseModel): @@ -291,6 +294,8 @@ class CategoryCreate(BaseModel): # None: erbt vom Elternteil bzw. neue Oberkategorie = Gegenstand. tracking: CategoryTracking | None = None open_days: int | None = Field(default=None, ge=1) + # None: erbt vom Elternteil bzw. neue Oberkategorie = aus. + usage_tracking: bool | None = None class CategoryUpdate(BaseModel): @@ -300,6 +305,9 @@ class CategoryUpdate(BaseModel): # 0 ist nicht erlaubt: „sofort verbrauchen" waere keine Frist, und None # heisst schon „erben" – zwei Bedeutungen fuer leer gaebe es sonst. open_days: int | None = Field(default=None, ge=1) + # Wird beim Speichern in den ganzen Unterbaum uebernommen, siehe + # routers/categories.py::update_category. + usage_tracking: bool | None = None # ---- Shops (Bezugsquellen, nur für Gegenstände) ---- @@ -1064,6 +1072,28 @@ class LoanOut(BaseModel): location_name: str | None = None +class VerwendungSetzen(BaseModel): + """Verwendungsstatus eines Einzelstuecks setzen. + + ``status`` leer heisst verfuegbar – dafuer gibt es keinen eigenen Wert, es + wird nur der laufende Zustand beendet (siehe models.py::UsageStatus). + """ + status: UsageStatus | None = None + + +class VerwendungOut(BaseModel): + """Ein Zeitraum in einem Verwendungszustand – laufend oder abgeschlossen.""" + model_config = ConfigDict(from_attributes=True) + id: int + item_id: int + status: UsageStatus + started_at: datetime + # None = laeuft noch. + ended_at: datetime | None = None + # Wer umgestellt hat (Konto kann spaeter geloescht sein). + username: str | None = None + + class ItemKurz(BaseModel): """Ein Einzelstueck in Kurzform – fuer Zubehoer-Listen in beide Richtungen. @@ -1112,6 +1142,12 @@ class ItemOut(BaseModel): # Der offene Verleih, falls das Stück gerade draußen ist – sonst None. # Der Lagerort bleibt daneben stehen: dorthin gehört es zurück. loan: LoanOut | None = None + # Der laufende Verwendungszustand – None heisst verfuegbar. Anders als beim + # Verleih bleibt das Stueck dabei im Bestand: getragen ist nicht weg. + verwendung: VerwendungOut | None = None + # Ob die Kategorie dieses Stuecks ueberhaupt einen Verwendungsstatus fuehrt. + # Steuert, ob die Oberflaeche die Bedienung dafuer zeigt. + verwendung_moeglich: bool = False # Zubehör: die Teile dieses Geräts und, umgekehrt, das Gerät, zu dem dieses # Stück gehört. Beide Richtungen, weil beide Fragen im Alltag vorkommen. zubehoer: list[ItemKurz] = [] diff --git a/backend/app/services/items.py b/backend/app/services/items.py index b082af0..d6d1b6d 100644 --- a/backend/app/services/items.py +++ b/backend/app/services/items.py @@ -6,8 +6,8 @@ import secrets from sqlalchemy.orm import Session -from ..models import Item, ItemLoan -from ..schemas import ItemKurz, ItemOut +from ..models import Item, ItemLoan, ItemUsage +from ..schemas import ItemKurz, ItemOut, VerwendungOut from .loans import loan_to_out # Ohne 0/O/1/I/L, damit die UID auf einem Etikett eindeutig lesbar bleibt. @@ -34,6 +34,26 @@ def _offener(item: Item) -> ItemLoan | None: return None +def _offene_verwendung(item: Item) -> ItemUsage | None: + """Der laufende Verwendungszustand aus den schon geladenen Zeitraeumen. + + Wie ``_offener``: ueber die Beziehung statt per eigener Abfrage, damit die + Einzelstueck-Liste nicht je Zeile nachfragt. Keine offene Zeile heisst + verfuegbar - dafuer gibt es keinen gespeicherten Wert. + """ + for usage in item.usages: + if usage.ended_at is None: + return usage + return None + + +def verwendung_to_out(usage: ItemUsage) -> VerwendungOut: + """VerwendungOut samt Konto, das umgestellt hat.""" + out = VerwendungOut.model_validate(usage) + out.username = usage.user.username if usage.user else None + return out + + def item_kurz(item: Item) -> ItemKurz: """Ein verknuepftes Stueck in Kurzform (Zubehoer, Geraet).""" return ItemKurz( @@ -59,6 +79,11 @@ def item_to_out(item: Item) -> ItemOut: out.category_name = kategorie.name if kategorie else None loan = _offener(item) out.loan = loan_to_out(loan) if loan is not None else None + # Die Kategorie steckt schon oben in ``kategorie`` - die Markierung kostet + # deshalb keine zusaetzliche Abfrage (siehe Category.usage_tracking). + out.verwendung_moeglich = bool(kategorie.usage_tracking) if kategorie else False + verwendung = _offene_verwendung(item) + out.verwendung = verwendung_to_out(verwendung) if verwendung is not None else None # Beide Richtungen ausdruecklich setzen: model_validate wuerde die Beziehung # zwar finden, aber nur die Spalten fuellen - Artikelname und Lagerort des # verknuepften Stuecks blieben leer. diff --git a/backend/app/services/master_data.py b/backend/app/services/master_data.py index 22fc559..6cf8dc8 100644 --- a/backend/app/services/master_data.py +++ b/backend/app/services/master_data.py @@ -35,7 +35,8 @@ def export_master_data(db: Session) -> dict: return { FORMAT_KEY: FORMAT_VERSION, "categories": [ - {"id": c.id, "name": c.name, "parent_id": c.parent_id, "tracking": c.tracking} + {"id": c.id, "name": c.name, "parent_id": c.parent_id, + "tracking": c.tracking, "usage_tracking": c.usage_tracking} for c in db.query(Category).order_by(Category.id).all() ], "field_definitions": [ @@ -152,6 +153,7 @@ def import_master_data(db: Session, data: dict, mode: str = "skip") -> dict: for row in cats: upsert(Category, row, "categories", lambda r: { "name": r["name"], "tracking": r.get("tracking") or "food", + "usage_tracking": bool(r.get("usage_tracking", False)), }, is_tree=True) set_parents(Category, cats) diff --git a/backend/tests/test_verwendung.py b/backend/tests/test_verwendung.py new file mode 100644 index 0000000..0363cf4 --- /dev/null +++ b/backend/tests/test_verwendung.py @@ -0,0 +1,391 @@ +"""Verwendungsstatus von Einzelstücken: umstellen, Verlauf, Kategorie-Markierung.""" + +import pytest +from fastapi import HTTPException +from sqlalchemy.exc import IntegrityError + +from app.models import ( + Category, + CategoryTracking, + Item, + ItemUsage, + Location, + Product, + Role, + UsageStatus, + User, +) +from app.routers.categories import create_category, update_category +from app.routers.items import create_items, delete_item, get_item, list_all_items +from app.routers.verwendung import list_verwendungen, setze_verwendung +from app.schemas import ( + CategoryCreate, + CategoryUpdate, + ItemCreate, + VerwendungSetzen, +) +from app.services.stock import current_stock + + +@pytest.fixture() +def admin(db): + person = User(username="admin", password_hash="x", role=Role.admin) + db.add(person) + db.commit() + db.refresh(person) + return person + + +@pytest.fixture() +def nutzer(db): + person = User(username="mitbewohner", password_hash="x", role=Role.user) + db.add(person) + db.commit() + db.refresh(person) + return person + + +@pytest.fixture() +def schrank(db): + ort = Location(name="Schrank") + db.add(ort) + db.commit() + db.refresh(ort) + return ort + + +@pytest.fixture() +def hemd(db): + """Ein Einzelstueck in einer Kategorie, die den Status fuehrt.""" + cat = Category( + name="Kleidung", tracking=CategoryTracking.object.value, usage_tracking=True + ) + db.add(cat) + db.flush() + product = Product(name="Hemd", category_id=cat.id, individual=True, package_size=1) + db.add(product) + db.commit() + db.refresh(product) + return product + + +@pytest.fixture() +def konserve(db): + """Ein Einzelstueck in einer Kategorie OHNE Verwendungsstatus.""" + cat = Category(name="Vorrat", tracking=CategoryTracking.object.value) + db.add(cat) + db.flush() + product = Product(name="Bohrmaschine", category_id=cat.id, individual=True, + package_size=1) + db.add(product) + db.commit() + db.refresh(product) + return product + + +def _stueck(db, admin, product, ort=None): + return create_items( + product.id, + ItemCreate(count=1, location_id=(ort.id if ort else None)), + db=db, user=admin, + )[0] + + +def _setze(db, user, item_id, status): + return setze_verwendung(item_id, VerwendungSetzen(status=status), db=db, user=user) + + +# ---- Umstellen ---- + +def test_neuer_status_legt_offene_zeile_an(db, admin, hemd): + stueck = _stueck(db, admin, hemd) + out = _setze(db, admin, stueck.id, UsageStatus.in_use) + + assert out.verwendung is not None + assert out.verwendung.status == UsageStatus.in_use + assert out.verwendung.ended_at is None + assert out.verwendung.username == "admin" + + +def test_frisches_stueck_ist_verfuegbar(db, admin, hemd): + stueck = _stueck(db, admin, hemd) + assert get_item(stueck.id, db=db, _=admin).verwendung is None + assert db.query(ItemUsage).count() == 0 + + +def test_wechsel_schliesst_die_vorige_zeile(db, admin, hemd): + stueck = _stueck(db, admin, hemd) + _setze(db, admin, stueck.id, UsageStatus.in_use) + out = _setze(db, admin, stueck.id, UsageStatus.cleaning) + + assert out.verwendung.status == UsageStatus.cleaning + zeilen = db.query(ItemUsage).order_by(ItemUsage.id).all() + assert len(zeilen) == 2 + assert zeilen[0].status == UsageStatus.in_use.value + assert zeilen[0].ended_at is not None + assert zeilen[1].ended_at is None + + +def test_nur_eine_offene_zeile_je_stueck(db, admin, hemd): + """Der Teilindex muss einen zweiten laufenden Zeitraum verhindern.""" + stueck = _stueck(db, admin, hemd) + _setze(db, admin, stueck.id, UsageStatus.in_use) + + db.add(ItemUsage(item_id=stueck.id, status=UsageStatus.repair.value)) + with pytest.raises(IntegrityError): + db.commit() + db.rollback() + + +def test_verfuegbar_schliesst_nur_und_legt_nichts_an(db, admin, hemd): + stueck = _stueck(db, admin, hemd) + _setze(db, admin, stueck.id, UsageStatus.cleaning) + out = _setze(db, admin, stueck.id, None) + + assert out.verwendung is None + zeilen = db.query(ItemUsage).all() + assert len(zeilen) == 1 + assert zeilen[0].ended_at is not None + + +def test_verfuegbar_ohne_offene_zeile_ist_kein_fehler(db, admin, hemd): + stueck = _stueck(db, admin, hemd) + out = _setze(db, admin, stueck.id, None) + assert out.verwendung is None + assert db.query(ItemUsage).count() == 0 + + +def test_gleicher_status_zweimal_legt_keine_zweite_zeile_an(db, admin, hemd): + """Sonst entstuende bei jedem Antippen ein Zeitraum der Laenge 0.""" + stueck = _stueck(db, admin, hemd) + _setze(db, admin, stueck.id, UsageStatus.repair) + _setze(db, admin, stueck.id, UsageStatus.repair) + + zeilen = db.query(ItemUsage).all() + assert len(zeilen) == 1 + assert zeilen[0].ended_at is None + + +def test_unbekanntes_stueck_ergibt_404(db, admin): + with pytest.raises(HTTPException) as fehler: + _setze(db, admin, 9999, UsageStatus.in_use) + assert fehler.value.status_code == 404 + + +def test_normaler_nutzer_darf_status_setzen(db, admin, nutzer, hemd): + """Umstellen ist eine Buchung, keine Stammdatenaenderung.""" + stueck = _stueck(db, admin, hemd) + out = _setze(db, nutzer, stueck.id, UsageStatus.cleaning) + assert out.verwendung.username == "mitbewohner" + + +# ---- Bestand ---- + +def test_status_aendert_den_bestand_nicht(db, admin, hemd, schrank): + """Ein getragenes Hemd ist weiterhin im Haushalt - anders als ein verliehenes.""" + stueck = _stueck(db, admin, hemd, schrank) + vorher = current_stock(db, hemd.id) + + _setze(db, admin, stueck.id, UsageStatus.in_use) + assert current_stock(db, hemd.id) == vorher + + _setze(db, admin, stueck.id, UsageStatus.cleaning) + assert current_stock(db, hemd.id) == vorher + + +def test_status_schreibt_keine_bewegung(db, admin, hemd): + """Der Dashboard-Verlauf rechnet aus Bewegungen - siehe routers/loans.py.""" + from app.models import Movement + + stueck = _stueck(db, admin, hemd) + vorher = db.query(Movement).count() + _setze(db, admin, stueck.id, UsageStatus.in_use) + assert db.query(Movement).count() == vorher + + +# ---- Verlauf ---- + +def test_verlauf_kommt_neueste_zuerst(db, admin, hemd): + stueck = _stueck(db, admin, hemd) + for status in (UsageStatus.in_use, UsageStatus.cleaning, UsageStatus.repair): + _setze(db, admin, stueck.id, status) + + verlauf = list_verwendungen(stueck.id, db=db, _=admin) + assert [v.status for v in verlauf] == [ + UsageStatus.repair, UsageStatus.cleaning, UsageStatus.in_use + ] + assert verlauf[0].ended_at is None + assert all(v.ended_at is not None for v in verlauf[1:]) + + +def test_geloeschtes_stueck_nimmt_den_verlauf_mit(db, admin, hemd): + stueck = _stueck(db, admin, hemd) + _setze(db, admin, stueck.id, UsageStatus.cleaning) + assert db.query(ItemUsage).count() == 1 + + delete_item(stueck.id, db=db, _=admin) + assert db.query(Item).filter(Item.id == stueck.id).count() == 0 + assert db.query(ItemUsage).count() == 0 + + +# ---- Kategorie-Markierung ---- + +def test_verwendung_moeglich_steht_am_einzelstueck(db, admin, hemd, konserve): + mit = _stueck(db, admin, hemd) + ohne = _stueck(db, admin, konserve) + + assert get_item(mit.id, db=db, _=admin).verwendung_moeglich is True + assert get_item(ohne.id, db=db, _=admin).verwendung_moeglich is False + + nach_uid = {i.uid: i for i in list_all_items(db=db, _=admin)} + assert nach_uid[mit.uid].verwendung_moeglich is True + assert nach_uid[ohne.uid].verwendung_moeglich is False + + +def test_markierung_erbt_beim_anlegen(db, admin, hemd): + eltern = db.query(Category).filter(Category.name == "Kleidung").one() + kind = create_category( + CategoryCreate(name="Hemden", parent_id=eltern.id), db=db, _=admin + ) + assert kind.usage_tracking is True + + +def test_neue_oberkategorie_ist_ohne_markierung(db, admin): + neu = create_category(CategoryCreate(name="Werkzeug"), db=db, _=admin) + assert neu.usage_tracking is False + + +def test_markierung_wandert_in_die_unterkategorien(db, admin): + """Der Fall aus der Praxis: Die Garderobe ist schon erfasst, erst dann + kommt der Haken an die Oberkategorie.""" + oben = create_category(CategoryCreate(name="Kleidung"), db=db, _=admin) + mitte = create_category( + CategoryCreate(name="Oberteile", parent_id=oben.id), db=db, _=admin + ) + unten = create_category( + CategoryCreate(name="Hemden", parent_id=mitte.id), db=db, _=admin + ) + assert unten.usage_tracking is False + + update_category(oben.id, CategoryUpdate(usage_tracking=True), db=db, _=admin) + + for cid in (oben.id, mitte.id, unten.id): + assert db.get(Category, cid).usage_tracking is True + + +def test_markierung_abwaehlen_wandert_auch(db, admin): + oben = create_category( + CategoryCreate(name="Kleidung", usage_tracking=True), db=db, _=admin + ) + unten = create_category( + CategoryCreate(name="Hosen", parent_id=oben.id), db=db, _=admin + ) + assert unten.usage_tracking is True + + update_category(oben.id, CategoryUpdate(usage_tracking=False), db=db, _=admin) + assert db.get(Category, unten.id).usage_tracking is False + + +def test_andere_aenderung_laesst_die_markierung_in_ruhe(db, admin): + """Ein Umbenennen darf die Unterkategorien nicht anfassen.""" + oben = create_category(CategoryCreate(name="Kleidung"), db=db, _=admin) + unten = create_category( + CategoryCreate(name="Hemden", parent_id=oben.id, usage_tracking=True), + db=db, _=admin, + ) + update_category(oben.id, CategoryUpdate(name="Garderobe"), db=db, _=admin) + assert db.get(Category, unten.id).usage_tracking is True + + +# ---- Sicherung ---- + +def test_markierung_uebersteht_die_volle_sicherung(db, admin): + """Die Markierung haengt an der Kategorie und muss im Backup mitfahren. + + Ohne die Umschreibung in transfer.py waere sie nach dem Zurueckspielen aus + - und niemand wuerde merken, dass die Garderobe ihren Status verloren hat. + """ + import json + + from sqlalchemy import create_engine + from sqlalchemy.orm import sessionmaker + from sqlalchemy.pool import StaticPool + + from app.database import Base + from app.routers.transfer import _import_json, export_backup_json + from app.seed import ensure_builtin_units + + oben = create_category( + CategoryCreate(name="Kleidung", usage_tracking=True), db=db, _=admin + ) + create_category(CategoryCreate(name="Hemden", parent_id=oben.id), db=db, _=admin) + create_category(CategoryCreate(name="Werkzeug"), db=db, _=admin) + + inhalt = export_backup_json(db=db, _=admin).body + daten = json.loads(inhalt) + nach_pfad = {c["path"]: c for c in daten["categories"]} + assert nach_pfad["Kleidung"]["usage_tracking"] is True + assert nach_pfad["Kleidung > Hemden"]["usage_tracking"] is True + assert nach_pfad["Werkzeug"]["usage_tracking"] is False + + # Zweite, leere Instanz. + engine = create_engine("sqlite://", connect_args={"check_same_thread": False}, + poolclass=StaticPool) + Base.metadata.create_all(bind=engine) + zweite = sessionmaker(bind=engine, autoflush=False, autocommit=False)() + ensure_builtin_units(zweite) + person = User(username="zwei", password_hash="x", role=Role.admin) + zweite.add(person) + zweite.commit() + + _import_json(zweite, inhalt, person, "add") + + def markierung(name): + return zweite.query(Category).filter(Category.name == name).one().usage_tracking + + assert markierung("Kleidung") is True + assert markierung("Hemden") is True + assert markierung("Werkzeug") is False + zweite.close() + engine.dispose() + + +def test_markierung_uebersteht_die_stammdaten_sicherung(db, admin): + from app.services.master_data import export_master_data, import_master_data + + from sqlalchemy import create_engine + from sqlalchemy.orm import sessionmaker + from sqlalchemy.pool import StaticPool + + from app.database import Base + + create_category(CategoryCreate(name="Kleidung", usage_tracking=True), db=db, _=admin) + create_category(CategoryCreate(name="Werkzeug"), db=db, _=admin) + daten = export_master_data(db) + + engine = create_engine("sqlite://", connect_args={"check_same_thread": False}, + poolclass=StaticPool) + Base.metadata.create_all(bind=engine) + zweite = sessionmaker(bind=engine, autoflush=False, autocommit=False)() + import_master_data(zweite, daten, "add") + + def markierung(name): + return zweite.query(Category).filter(Category.name == name).one().usage_tracking + + assert markierung("Kleidung") is True + assert markierung("Werkzeug") is False + zweite.close() + engine.dispose() + + +def test_altes_backup_ohne_markierung_bleibt_lesbar(db, admin): + """Eine Sicherung von vor dieser Aenderung kennt den Schluessel nicht.""" + from app.services.master_data import FORMAT_KEY, import_master_data + + import_master_data( + db, + {FORMAT_KEY: 1, + "categories": [{"id": 1, "name": "Alt", "parent_id": None, "tracking": "food"}]}, + "add", + ) + assert db.query(Category).filter(Category.name == "Alt").one().usage_tracking is False