012e3461ae
/lots/rows liefert jetzt die Art (Lebensmittel/Gegenstand) mit und listet nur Charge-Artikel (Lebensmittel + Verbrauchsgegenstand). Lots von "Menge je Lagerort" (und etwaige Einzelstueck-Altlasten) fallen raus. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
394 lines
12 KiB
Python
394 lines
12 KiB
Python
"""Gegenstände (Non-Food): Menge je Lagerort, Umlagern, Entfernen mit Grund,
|
||
Kategorie-Modus und selbst definierte Felder.
|
||
|
||
Die Lebensmittel-Logik (Chargen/MHD/FEFO) bleibt davon unberührt – dafür sorgt
|
||
test_fefo.py weiterhin.
|
||
"""
|
||
|
||
import json
|
||
|
||
import pytest
|
||
|
||
from app.crud import product_to_out, product_tracking, products_to_out_bulk
|
||
from app.models import (
|
||
BaseUnit,
|
||
Category,
|
||
CategoryTracking,
|
||
FieldDefinition,
|
||
Item,
|
||
Location,
|
||
Lot,
|
||
Movement,
|
||
Product,
|
||
ProductFieldValue,
|
||
RemovalReason,
|
||
)
|
||
from app.services.items import generate_uid
|
||
from app.routers.categories import create_category
|
||
from app.schemas import CategoryCreate
|
||
from app.services.fields import (
|
||
FieldError,
|
||
apply_field_values,
|
||
coerce_value,
|
||
effective_field_definitions,
|
||
)
|
||
from app.services.stock import (
|
||
StockError,
|
||
current_stock,
|
||
object_add,
|
||
object_relocate,
|
||
object_remove,
|
||
removal_stats,
|
||
)
|
||
|
||
|
||
def _object_product(db, name="Unterhose"):
|
||
cat = Category(name="Kleidung", tracking=CategoryTracking.object.value)
|
||
db.add(cat)
|
||
db.flush()
|
||
product = Product(name=name, category_id=cat.id)
|
||
db.add(product)
|
||
db.commit()
|
||
db.refresh(product)
|
||
return product, cat
|
||
|
||
|
||
def _loc(db, name):
|
||
loc = Location(name=name)
|
||
db.add(loc)
|
||
db.commit()
|
||
db.refresh(loc)
|
||
return loc
|
||
|
||
|
||
# ---- Modus je Kategorie ----
|
||
|
||
def test_ohne_kategorie_bleibt_food(db):
|
||
product = Product(name="Altbestand")
|
||
db.add(product)
|
||
db.commit()
|
||
assert product_tracking(db, product) == "food"
|
||
|
||
|
||
def test_gegenstandskategorie_macht_object(db):
|
||
product, _ = _object_product(db)
|
||
assert product_tracking(db, product) == "object"
|
||
|
||
|
||
def test_verbrauchsgegenstand_ohne_kategorie_ist_object(db):
|
||
# Nach dem Umstellen Lebensmittel -> Verbrauchsgegenstand hat der Artikel oft
|
||
# keine (Gegenstands-)Kategorie mehr. Das Flag muss ihn trotzdem als
|
||
# Gegenstand führen, sonst rutscht er zurück in die Lebensmittelliste.
|
||
product = Product(name="Sonnencreme", bulk=True)
|
||
db.add(product)
|
||
db.commit()
|
||
assert product_tracking(db, product) == "object"
|
||
|
||
|
||
def test_einzelstueck_ohne_kategorie_ist_object(db):
|
||
product = Product(name="Powerbank", individual=True)
|
||
db.add(product)
|
||
db.commit()
|
||
assert product_tracking(db, product) == "object"
|
||
|
||
|
||
# ---- Gebündelte Listen-Ausgabe ----
|
||
|
||
def test_bulk_liste_zaehlt_bestand_wie_einzeln(db):
|
||
# products_to_out_bulk muss denselben Bestand liefern wie product_to_out –
|
||
# Lot-Summe für Mengen, Item-Anzahl für Einzelstücke.
|
||
reis = Product(name="Reis", base_unit=BaseUnit.gram, package_size=500)
|
||
db.add(reis)
|
||
db.flush()
|
||
a = _loc(db, "Regal")
|
||
object_add(db, reis, 1000, a.id, None) # 1000 g als Charge
|
||
powerbank = _individual_product(db, name="Powerbank")
|
||
for _ in range(2):
|
||
db.add(Item(uid=generate_uid(db), product_id=powerbank.id))
|
||
db.commit()
|
||
|
||
einzeln = {product_to_out(db, p).id: product_to_out(db, p).stock for p in [reis, powerbank]}
|
||
gebuendelt = {o.id: o.stock for o in products_to_out_bulk(db, [reis, powerbank])}
|
||
assert gebuendelt == einzeln
|
||
assert gebuendelt[reis.id] == 1000
|
||
assert gebuendelt[powerbank.id] == 2
|
||
|
||
|
||
def test_chargenliste_nur_charge_artikel(db):
|
||
# Die Chargenliste zeigt nur Lebensmittel + Verbrauchsgegenstände – nicht die
|
||
# Lots von „Menge je Lagerort".
|
||
from app.routers.stock import list_all_lots
|
||
a = _loc(db, "Regal")
|
||
food = Product(name="Reis", base_unit=BaseUnit.gram)
|
||
db.add(food)
|
||
db.flush()
|
||
object_add(db, food, 500, a.id, None)
|
||
unterhose, _ = _object_product(db, name="Unterhose") # Menge je Lagerort
|
||
object_add(db, unterhose, 5, a.id, None)
|
||
creme = _bulk_product(db, name="Sonnencreme") # Verbrauchsgegenstand
|
||
object_add(db, creme, 200, a.id, None)
|
||
db.commit()
|
||
|
||
namen = {r.product_name for r in list_all_lots(db=db, _=None)}
|
||
assert "Reis" in namen
|
||
assert "Sonnencreme" in namen
|
||
assert "Unterhose" not in namen
|
||
|
||
|
||
def test_neue_oberkategorie_ist_gegenstand(db):
|
||
out = create_category(CategoryCreate(name="Werkzeug"), db, None)
|
||
assert out.tracking == CategoryTracking.object
|
||
|
||
|
||
def test_unterkategorie_erbt_modus_vom_elternteil(db):
|
||
ober = create_category(
|
||
CategoryCreate(name="Lebensmittel", tracking=CategoryTracking.food), db, None
|
||
)
|
||
unter = create_category(CategoryCreate(name="Käse", parent_id=ober.id), db, None)
|
||
assert unter.tracking == CategoryTracking.food
|
||
|
||
|
||
# ---- Menge je Lagerort ----
|
||
|
||
def test_add_summiert_je_ort(db):
|
||
product, _ = _object_product(db)
|
||
a, b = _loc(db, "Schrank A"), _loc(db, "Schrank B")
|
||
object_add(db, product, 5, a.id, None)
|
||
object_add(db, product, 3, b.id, None)
|
||
db.commit()
|
||
assert current_stock(db, product.id) == 8
|
||
verteilung = {
|
||
lot.location_id: lot.quantity
|
||
for lot in db.query(Lot).filter(Lot.product_id == product.id).all()
|
||
}
|
||
assert verteilung == {a.id: 5, b.id: 3}
|
||
|
||
|
||
def test_add_am_gleichen_ort_erhoeht_dieselbe_zeile(db):
|
||
product, _ = _object_product(db)
|
||
a = _loc(db, "Schrank A")
|
||
object_add(db, product, 2, a.id, None)
|
||
object_add(db, product, 3, a.id, None)
|
||
db.commit()
|
||
lots = db.query(Lot).filter(Lot.product_id == product.id).all()
|
||
assert len(lots) == 1 and lots[0].quantity == 5
|
||
|
||
|
||
# ---- Umlagern (ohne Grund) ----
|
||
|
||
def test_umlagern_verschiebt_menge(db):
|
||
product, _ = _object_product(db)
|
||
a, b = _loc(db, "A"), _loc(db, "B")
|
||
object_add(db, product, 5, a.id, None)
|
||
db.commit()
|
||
object_relocate(db, product, 2, a.id, b.id, None)
|
||
db.commit()
|
||
verteilung = {
|
||
lot.location_id: lot.quantity
|
||
for lot in db.query(Lot).filter(Lot.product_id == product.id).all()
|
||
}
|
||
assert verteilung == {a.id: 3, b.id: 2}
|
||
assert current_stock(db, product.id) == 5
|
||
# Umlagern erzeugt keine Entnahme mit Grund.
|
||
assert db.query(Movement).filter(Movement.reason.isnot(None)).count() == 0
|
||
|
||
|
||
def test_umlagern_zu_wenig_bestand(db):
|
||
product, _ = _object_product(db)
|
||
a, b = _loc(db, "A"), _loc(db, "B")
|
||
object_add(db, product, 1, a.id, None)
|
||
db.commit()
|
||
with pytest.raises(StockError):
|
||
object_relocate(db, product, 5, a.id, b.id, None)
|
||
|
||
|
||
def test_umlagern_gleicher_ort_fehler(db):
|
||
product, _ = _object_product(db)
|
||
a = _loc(db, "A")
|
||
object_add(db, product, 3, a.id, None)
|
||
db.commit()
|
||
with pytest.raises(StockError):
|
||
object_relocate(db, product, 1, a.id, a.id, None)
|
||
|
||
|
||
# ---- Entfernen mit Grund + Statistik ----
|
||
|
||
def test_entfernen_mit_grund_und_statistik(db):
|
||
product, _ = _object_product(db)
|
||
a = _loc(db, "A")
|
||
object_add(db, product, 5, a.id, None)
|
||
db.commit()
|
||
object_remove(db, product, 2, a.id, RemovalReason.broken.value, None)
|
||
object_remove(db, product, 1, a.id, RemovalReason.broken.value, None)
|
||
object_remove(db, product, 1, a.id, RemovalReason.lost.value, None)
|
||
db.commit()
|
||
assert current_stock(db, product.id) == 1
|
||
stats = removal_stats(db, product.id)
|
||
assert stats["broken"] == {"quantity": 3, "count": 2}
|
||
assert stats["lost"] == {"quantity": 1, "count": 1}
|
||
|
||
|
||
def test_entfernen_zu_viel_bestand(db):
|
||
product, _ = _object_product(db)
|
||
a = _loc(db, "A")
|
||
object_add(db, product, 1, a.id, None)
|
||
db.commit()
|
||
with pytest.raises(StockError):
|
||
object_remove(db, product, 5, a.id, RemovalReason.lost.value, None)
|
||
|
||
|
||
# ---- Selbst definierte Felder ----
|
||
|
||
def test_felder_werden_vererbt_oberkategorie_zuerst(db):
|
||
ober = Category(name="Elektronik", tracking="object")
|
||
db.add(ober)
|
||
db.flush()
|
||
unter = Category(name="Powerbank", parent_id=ober.id, tracking="object")
|
||
db.add(unter)
|
||
db.commit()
|
||
db.add(FieldDefinition(category_id=ober.id, label="Kaufdatum", key="kaufdatum", field_type="date"))
|
||
db.add(
|
||
FieldDefinition(
|
||
category_id=unter.id, label="Kapazität", key="kapazitaet",
|
||
field_type="number", unit="mAh",
|
||
)
|
||
)
|
||
db.commit()
|
||
eff = effective_field_definitions(db, unter.id)
|
||
assert [fd.label for fd, _ in eff] == ["Kaufdatum", "Kapazität"]
|
||
geerbt = {fd.label: inh for fd, inh in eff}
|
||
assert geerbt["Kaufdatum"] is True and geerbt["Kapazität"] is False
|
||
|
||
|
||
def test_zahlfeld_prueft_typ(db):
|
||
fd = FieldDefinition(category_id=1, label="Kapazität", key="kap", field_type="number")
|
||
assert coerce_value(fd, "20000") == "20000"
|
||
assert coerce_value(fd, "3,5") == "3,5"
|
||
with pytest.raises(FieldError):
|
||
coerce_value(fd, "abc")
|
||
|
||
|
||
def test_auswahlfeld_prueft_optionen(db):
|
||
fd = FieldDefinition(
|
||
category_id=1, label="Größe", key="groesse", field_type="select",
|
||
options=json.dumps(["S", "M", "L"]),
|
||
)
|
||
assert coerce_value(fd, "M") == "M"
|
||
with pytest.raises(FieldError):
|
||
coerce_value(fd, "XXL")
|
||
|
||
|
||
def test_feldwert_setzen_und_leeren(db):
|
||
cat = Category(name="Elektronik", tracking="object")
|
||
db.add(cat)
|
||
db.flush()
|
||
fd = FieldDefinition(category_id=cat.id, label="Notiz", key="notiz", field_type="text")
|
||
db.add(fd)
|
||
db.commit()
|
||
product = Product(name="Gerät", category_id=cat.id)
|
||
db.add(product)
|
||
db.commit()
|
||
|
||
apply_field_values(db, product, {fd.id: "wichtig"})
|
||
db.commit()
|
||
assert (
|
||
db.query(ProductFieldValue).filter_by(product_id=product.id).one().value
|
||
== "wichtig"
|
||
)
|
||
# Leerer Wert löscht den Eintrag wieder.
|
||
apply_field_values(db, product, {fd.id: ""})
|
||
db.commit()
|
||
assert db.query(ProductFieldValue).filter_by(product_id=product.id).count() == 0
|
||
|
||
|
||
# ---- Einzelstücke (Items mit UID/QR) ----
|
||
|
||
def _individual_product(db, name="Powerbank"):
|
||
cat = Category(name="Elektronik", tracking=CategoryTracking.object.value)
|
||
db.add(cat)
|
||
db.flush()
|
||
product = Product(name=name, category_id=cat.id, individual=True)
|
||
db.add(product)
|
||
db.commit()
|
||
db.refresh(product)
|
||
return product
|
||
|
||
|
||
def test_uid_ist_eindeutig(db):
|
||
codes = {generate_uid(db) for _ in range(50)}
|
||
assert len(codes) == 50
|
||
# Keine verwechselbaren Zeichen (0/O/1/I/L) in der UID.
|
||
assert all(ch not in "01OIL" for code in codes for ch in code)
|
||
|
||
|
||
def test_einzelstueck_bestand_zaehlt_items(db):
|
||
product = _individual_product(db)
|
||
for _ in range(3):
|
||
db.add(Item(uid=generate_uid(db), product_id=product.id))
|
||
db.commit()
|
||
assert current_stock(db, product.id) == 3
|
||
|
||
|
||
def test_mengen_produkt_zaehlt_weiter_lots(db):
|
||
# Ein Nicht-Einzelstück-Produkt zählt weiterhin über Lots, nicht über Items.
|
||
product, _ = _object_product(db, name="Unterhose")
|
||
a = _loc(db, "Schrank A")
|
||
object_add(db, product, 5, a.id, None)
|
||
db.commit()
|
||
assert current_stock(db, product.id) == 5
|
||
|
||
|
||
# ---- Verbrauchsgegenstand (bulk): wie ein Lebensmittel geführt ----
|
||
|
||
def _bulk_product(db, name="Sonnencreme", base_unit=BaseUnit.milliliter, min_stock=500):
|
||
"""Gegenstand, der als Charge (Menge + Einheit) geführt wird – wie ein
|
||
Lebensmittel, aber in der Gegenstände-Welt und ohne eindeutigen Code."""
|
||
cat = Category(name="Pflege", tracking=CategoryTracking.object.value)
|
||
db.add(cat)
|
||
db.flush()
|
||
product = Product(
|
||
name=name, category_id=cat.id, bulk=True,
|
||
base_unit=base_unit, min_stock=min_stock,
|
||
)
|
||
db.add(product)
|
||
db.commit()
|
||
db.refresh(product)
|
||
return product
|
||
|
||
|
||
def test_verbrauchsgegenstand_zaehlt_ueber_lots(db):
|
||
# bulk-Gegenstand (individual=False) zählt über Chargen wie ein Lebensmittel.
|
||
product = _bulk_product(db)
|
||
assert product.bulk is True and product.individual is False
|
||
a = _loc(db, "Bad")
|
||
object_add(db, product, 200, a.id, None)
|
||
object_add(db, product, 200, a.id, None)
|
||
db.commit()
|
||
assert current_stock(db, product.id) == 400
|
||
|
||
|
||
def test_verbrauchsgegenstand_unter_mindestbestand(db):
|
||
product = _bulk_product(db, min_stock=500) # 500 ml
|
||
a = _loc(db, "Bad")
|
||
object_add(db, product, 200, a.id, None)
|
||
db.commit()
|
||
# Unter dem Mindestbestand -> gehört auf die Einkaufsliste (wie ein Lebensmittel).
|
||
assert current_stock(db, product.id) < product.min_stock
|
||
|
||
|
||
# ---- Zuletzt geändert ----
|
||
|
||
def test_updated_at_wird_bei_aenderung_nachgezogen(db):
|
||
import time
|
||
product = Product(name="Alt")
|
||
db.add(product)
|
||
db.commit()
|
||
db.refresh(product)
|
||
erst = product.updated_at
|
||
assert erst is not None
|
||
time.sleep(0.01)
|
||
product.name = "Neu"
|
||
db.commit()
|
||
db.refresh(product)
|
||
assert product.updated_at > erst
|