Vorrania/backend/tests/test_verwendung.py

"""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