Vorrania/backend/tests/test_etiketten.py

"""Etiketten-Vorlagen: Spaltenwahl, eigene Felder, CSV-Form und .lbx-Prüfung.

Der Export ist eine reine Funktion über einer Datenbank-Sitzung – geprüft wird
deshalb ohne HTTP, direkt am Ergebnis.
"""

import io
import json
import unicodedata
import zipfile
from urllib.parse import unquote

import pytest
from fastapi import HTTPException

from app.models import (
    Category,
    CategoryTracking,
    EtikettenVorlage,
    FieldDefinition,
    Item,
    Location,
    Product,
    ProductFieldValue,
    Role,
    User,
)
from app.routers.etiketten import _pruefe_lbx, datei_herunterladen, spalten
from app.services.etiketten import csv_erzeugen, konfiguration_lesen


@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 kabel(db):
    """Kategorie „Kabel" mit eigenem Feld „Länge" in Metern."""
    cat = Category(name="Kabel", tracking=CategoryTracking.object.value)
    db.add(cat)
    db.flush()
    fd = FieldDefinition(
        category_id=cat.id, label="Länge", key="laenge", field_type="number", unit="m"
    )
    db.add(fd)
    db.commit()
    db.refresh(cat)
    db.refresh(fd)
    return cat, fd


def _vorlage(db, spalten, art="items", kategorie_ids=None, trenn=",", feldfilter=None):
    v = EtikettenVorlage(
        name=f"Test {len(db.query(EtikettenVorlage).all())}",
        art=art,
        konfiguration=json.dumps({
            "spalten": spalten,
            "kategorie_ids": kategorie_ids or [],
            "trennzeichen": trenn,
            "feldfilter": feldfilter or [],
        }),
    )
    db.add(v)
    db.commit()
    db.refresh(v)
    return v


def _stueck(db, cat, name="HDMI-Kabel", laenge=None, fd=None, ort=None):
    product = Product(name=name, category_id=cat.id, individual=True, package_size=1)
    db.add(product)
    db.flush()
    if laenge is not None and fd is not None:
        db.add(ProductFieldValue(
            product_id=product.id, field_definition_id=fd.id, value=laenge
        ))
    item = Item(uid=f"UID{product.id:05d}", product_id=product.id,
                location_id=(ort.id if ort else None))
    db.add(item)
    db.commit()
    db.refresh(item)
    return item


def _zeilen(csv_text):
    """Ohne BOM, als Liste von Zeilen."""
    assert csv_text.startswith(""), "BOM fehlt – Excel zeigt sonst kaputte Umlaute"
    return csv_text[1:].rstrip("\r\n").split("\r\n")


# --------------------------------------------------------------------------
# Spalten und Reihenfolge
# --------------------------------------------------------------------------
def test_spalten_kommen_in_gespeicherter_reihenfolge(db, kabel):
    cat, _ = kabel
    _stueck(db, cat)
    v = _vorlage(db, ["produkt", "uid", "qr"])

    kopf = _zeilen(csv_erzeugen(db, v))[0]
    assert kopf == "Produkt,UID,QR-Inhalt"


def test_eigenes_feld_landet_in_der_csv(db, kabel):
    cat, fd = kabel
    _stueck(db, cat, laenge="3", fd=fd)
    v = _vorlage(db, ["uid", f"feld:{fd.id}"])

    kopf, zeile = _zeilen(csv_erzeugen(db, v))
    assert kopf == "UID,Länge (m)"
    assert zeile.endswith(",3")


def test_fehlender_feldwert_gibt_eine_leere_zelle(db, kabel):
    """Nicht etwa eine Luecke – sonst verrutschen alle folgenden Spalten."""
    cat, fd = kabel
    _stueck(db, cat)          # ohne Wert
    v = _vorlage(db, ["uid", f"feld:{fd.id}", "produkt"])

    zeile = _zeilen(csv_erzeugen(db, v))[1]
    assert zeile.count(",") == 2
    assert ",," in zeile


def test_geloeschtes_feld_zerreisst_den_export_nicht(db, kabel):
    cat, fd = kabel
    _stueck(db, cat)
    v = _vorlage(db, ["uid", f"feld:{fd.id}", "produkt"])
    db.delete(fd)
    db.commit()

    kopf = _zeilen(csv_erzeugen(db, v))[0]
    assert kopf == "UID,Produkt"      # die tote Spalte faellt still weg


def test_ohne_gueltige_spalte_greifen_die_festen(db, kabel):
    cat, _ = kabel
    _stueck(db, cat)
    v = _vorlage(db, ["gibtsnicht"])

    kopf = _zeilen(csv_erzeugen(db, v))[0]
    assert kopf.startswith("QR-Inhalt,UID,Produkt")


# --------------------------------------------------------------------------
# CSV-Form
# --------------------------------------------------------------------------
def test_trennzeichen_semikolon_und_tab(db, kabel):
    cat, _ = kabel
    _stueck(db, cat)
    assert _zeilen(csv_erzeugen(db, _vorlage(db, ["uid", "produkt"], trenn=";")))[0] \
        == "UID;Produkt"
    assert _zeilen(csv_erzeugen(db, _vorlage(db, ["uid", "produkt"], trenn="\t")))[0] \
        == "UID\tProdukt"


def test_zelle_mit_trennzeichen_wird_eingefasst(db, kabel):
    cat, _ = kabel
    _stueck(db, cat, name="Kabel, lang")
    v = _vorlage(db, ["produkt"])

    assert _zeilen(csv_erzeugen(db, v))[1] == '"Kabel, lang"'


def test_formel_wird_entschaerft(db, kabel):
    """Ohne das fuehrt Excel den Zellinhalt als Formel aus."""
    cat, _ = kabel
    _stueck(db, cat, name="=1+1")
    v = _vorlage(db, ["produkt"])

    assert _zeilen(csv_erzeugen(db, v))[1] == "'=1+1"


def test_zeilenende_ist_crlf(db, kabel):
    cat, _ = kabel
    _stueck(db, cat)
    assert "\r\n" in csv_erzeugen(db, _vorlage(db, ["uid"]))


# --------------------------------------------------------------------------
# Bezugsarten und Filter
# --------------------------------------------------------------------------
def test_kategoriefilter_greift_samt_unterkategorien(db, kabel):
    cat, _ = kabel
    unter = Category(name="HDMI", tracking=CategoryTracking.object.value, parent_id=cat.id)
    andere = Category(name="Werkzeug", tracking=CategoryTracking.object.value)
    db.add_all([unter, andere])
    db.commit()
    _stueck(db, unter, name="HDMI 2.1")
    _stueck(db, andere, name="Hammer")

    v = _vorlage(db, ["produkt"], kategorie_ids=[cat.id])
    zeilen = _zeilen(csv_erzeugen(db, v))
    assert zeilen[1:] == ["HDMI 2.1"]


def test_lagerorte_brauchen_die_basis_adresse_fuer_den_qr(db):
    ort = Location(name="Keller")
    db.add(ort)
    db.commit()
    v = _vorlage(db, ["qr", "name"], art="locations")

    zeile = _zeilen(csv_erzeugen(db, v, basis_url="https://haus.example/"))[1]
    assert zeile.startswith(f"https://haus.example/l/{ort.id},")


def test_artikel_als_bezugsart(db, kabel):
    cat, fd = kabel
    _stueck(db, cat, name="USB-C", laenge="1", fd=fd)
    v = _vorlage(db, ["name", f"feld:{fd.id}"], art="products")

    kopf, zeile = _zeilen(csv_erzeugen(db, v))
    assert kopf == "Name,Länge (m)"
    assert zeile == "USB-C,1"


# --------------------------------------------------------------------------
# Konfiguration lesen
# --------------------------------------------------------------------------
def test_kaputte_konfiguration_ergibt_vorgaben(db):
    v = EtikettenVorlage(name="Kaputt", art="items", konfiguration="{nicht json")
    db.add(v)
    db.commit()

    konf = konfiguration_lesen(v)
    assert konf == {
        "spalten": [],
        "kategorie_ids": [],
        "trennzeichen": ",",
        "feldfilter": [],
    }


def test_unerlaubtes_trennzeichen_faellt_auf_komma_zurueck(db):
    v = _vorlage(db, ["uid"], trenn="|")
    assert konfiguration_lesen(v)["trennzeichen"] == ","


# --------------------------------------------------------------------------
# Spaltenliste für die Oberfläche
# --------------------------------------------------------------------------
def test_spaltenliste_nennt_eigene_felder_der_kategorie(db, admin, kabel):
    cat, fd = kabel
    liste = spalten(art="items", kategorie_ids=str(cat.id), db=db, _=admin)
    schluessel = {s.schluessel for s in liste}

    assert "qr" in schluessel and "uid" in schluessel
    assert f"feld:{fd.id}" in schluessel
    assert next(s.titel for s in liste if s.schluessel == f"feld:{fd.id}") == "Länge (m)"


def test_lagerorte_haben_keine_eigenen_felder(db, admin, kabel):
    liste = spalten(art="locations", kategorie_ids=None, db=db, _=admin)
    assert not any(s.schluessel.startswith("feld:") for s in liste)


def test_unbekannte_bezugsart_wird_abgelehnt(db, admin):
    with pytest.raises(HTTPException) as fehler:
        spalten(art="quatsch", kategorie_ids=None, db=db, _=admin)
    assert fehler.value.status_code == 400


# --------------------------------------------------------------------------
# .lbx-Prüfung
# --------------------------------------------------------------------------
def _zip(namen):
    puffer = io.BytesIO()
    with zipfile.ZipFile(puffer, "w") as z:
        for n in namen:
            z.writestr(n, "<xml/>")
    return puffer.getvalue()


def test_echte_lbx_wird_angenommen(db):
    _pruefe_lbx("etikett.lbx", _zip(["label.xml", "prop.xml"]))   # wirft nicht


def test_zip_ohne_label_xml_wird_abgelehnt(db):
    with pytest.raises(HTTPException) as fehler:
        _pruefe_lbx("etikett.lbx", _zip(["irgendwas.txt"]))
    assert fehler.value.status_code == 415


def test_nicht_zip_wird_abgelehnt(db):
    with pytest.raises(HTTPException) as fehler:
        _pruefe_lbx("etikett.lbx", b"\x89PNG\r\n\x1a\n und noch was")
    assert fehler.value.status_code == 415


def test_falsche_endung_wird_abgelehnt(db):
    with pytest.raises(HTTPException) as fehler:
        _pruefe_lbx("bild.png", _zip(["label.xml"]))
    assert fehler.value.status_code == 415


def test_leere_datei_wird_abgelehnt(db):
    with pytest.raises(HTTPException) as fehler:
        _pruefe_lbx("etikett.lbx", b"")
    assert fehler.value.status_code == 400


# --------------------------------------------------------------------------
# Backup-Umlauf
# --------------------------------------------------------------------------
def test_vorlage_uebersteht_die_sicherung(db, admin, kabel):
    """Kategorien als Pfad, eigene Felder als Label – IDs sind zwischen zwei
    Instanzen nicht dieselben."""
    import json as _json

    from app.routers.transfer import _import_json, export_backup_json
    from app.seed import ensure_builtin_units

    cat, fd = kabel
    _stueck(db, cat, laenge="5", fd=fd)
    v = _vorlage(db, ["uid", f"feld:{fd.id}"], kategorie_ids=[cat.id], trenn=";")
    v.name = "Kabelexport"
    v.datei_name = "kabel.lbx"
    v.datei = _zip(["label.xml"])
    db.commit()

    inhalt = export_backup_json(db=db, _=admin).body
    daten = _json.loads(inhalt)
    gesichert = daten["etiketten_vorlagen"]
    assert [e["name"] for e in gesichert] == ["Kabelexport"]
    # Das Feld steht als Pfad::Label drin, nicht als ID.
    assert gesichert[0]["spalten"][1] == "feld:Kabel::Länge"

    # In eine zweite, leere Instanz einspielen.
    from app.database import Base
    from sqlalchemy import create_engine
    from sqlalchemy.orm import sessionmaker
    from sqlalchemy.pool import StaticPool

    zweiter_motor = create_engine(
        "sqlite://", connect_args={"check_same_thread": False}, poolclass=StaticPool
    )
    Base.metadata.create_all(bind=zweiter_motor)
    zweite = sessionmaker(bind=zweiter_motor)()
    ensure_builtin_units(zweite)
    person = User(username="admin", password_hash="x", role=Role.admin)
    zweite.add(person)
    zweite.commit()

    _import_json(zweite, inhalt, person, "add")

    neu = zweite.query(EtikettenVorlage).filter(
        EtikettenVorlage.name == "Kabelexport"
    ).one()
    konf = konfiguration_lesen(neu)
    assert konf["trennzeichen"] == ";"
    assert neu.datei_name == "kabel.lbx"
    assert neu.datei == _zip(["label.xml"])
    # Das Feld wurde ueber Pfad und Label wiedergefunden – mit NEUER ID.
    feld_spalte = [s for s in konf["spalten"] if s.startswith("feld:")][0]
    neue_fd = zweite.query(FieldDefinition).filter(FieldDefinition.label == "Länge").one()
    assert feld_spalte == f"feld:{neue_fd.id}"
    zweite.close()


# --------------------------------------------------------------------------
# Download der hinterlegten .lbx
# --------------------------------------------------------------------------
def _disposition_pruefen(antwort, erwarteter_name):
    """Header muss sich senden lassen UND den echten Namen tragen."""
    roh = antwort.headers["content-disposition"]
    # Starlette kodiert Header als latin-1. Wirft das hier, antwortet die
    # Route im Betrieb mit 500 statt mit der Datei.
    roh.encode("latin-1")
    stern = roh.split("filename*=UTF-8''", 1)[1]
    assert unquote(stern) == erwarteter_name


def test_lbx_mit_zerlegtem_umlaut_laesst_sich_herunterladen(db, admin):
    """macOS legt Dateinamen ZERLEGT ab: das "ä" ist dort "a" + U+0308.

    Am Bildschirm sieht das wie die zusammengesetzte Form aus, aber nur die
    passt in latin-1 - die Kodierung, mit der HTTP-Header geschrieben werden.
    Ein Name, den der Dateiwaehler eines Macs liefert, darf den Download
    nicht sprengen.
    """
    name = unicodedata.normalize("NFD", "Lagerplätze.lbx")
    v = _vorlage(db, ["uid"])
    v.datei_name = name
    v.datei = _zip(["label.xml"])
    db.commit()

    antwort = datei_herunterladen(vorlage_id=v.id, db=db, _=admin)

    assert antwort.body == _zip(["label.xml"])
    _disposition_pruefen(antwort, name)


def test_lbx_mit_schlichtem_namen_bleibt_unveraendert(db, admin):
    """Der haeufige Fall darf von der Umstellung nicht beruehrt werden."""
    v = _vorlage(db, ["uid"])
    v.datei_name = "kabel.lbx"
    v.datei = _zip(["label.xml"])
    db.commit()

    antwort = datei_herunterladen(vorlage_id=v.id, db=db, _=admin)

    assert 'filename="kabel.lbx"' in antwort.headers["content-disposition"]
    _disposition_pruefen(antwort, "kabel.lbx")


# --------------------------------------------------------------------------
# Wertefilter: nur Zeilen mit bestimmten Feldwerten
# --------------------------------------------------------------------------
@pytest.fixture()
def haken(db):
    """Kategorie „Kleidung" mit einem Ja/Nein-Feld „Verliehen"."""
    cat = Category(name="Kleidung", tracking=CategoryTracking.object.value)
    db.add(cat)
    db.flush()
    fd = FieldDefinition(
        category_id=cat.id, label="Verliehen", key="verliehen", field_type="boolean"
    )
    db.add(fd)
    db.commit()
    db.refresh(cat)
    db.refresh(fd)
    return cat, fd


def _artikel(db, cat, name, fd=None, wert=None):
    """Artikel, optional mit einem Feldwert. ``wert=None`` legt KEINE Zeile an."""
    product = Product(name=name, category_id=cat.id, package_size=1)
    db.add(product)
    db.flush()
    if fd is not None and wert is not None:
        db.add(ProductFieldValue(
            product_id=product.id, field_definition_id=fd.id, value=wert
        ))
    db.commit()
    db.refresh(product)
    return product


def _namen(db, vorlage):
    """Die Namensspalte des erzeugten CSV, ohne Kopfzeile."""
    return _zeilen(csv_erzeugen(db, vorlage))[1:]


def test_feldfilter_waehlt_nur_passende_artikel(db, haken):
    cat, fd = haken
    _artikel(db, cat, "Jacke", fd, "true")
    _artikel(db, cat, "Hose", fd, "false")
    v = _vorlage(db, ["name"], art="products",
                 feldfilter=[{"feld_id": fd.id, "wert": "ja"}])

    assert _namen(db, v) == ["Jacke"]


def test_feldfilter_nein_trifft_auch_artikel_ohne_zeile(db, haken):
    """Der Kern: Haken wegnehmen loescht die Zeile.

    „nein" und „nie gesetzt" sind in den Daten dasselbe. Ohne diese Gleichung
    faende „alle mit Nein" nur die, die einmal angehakt und wieder abgehakt
    wurden – und das ist fast niemand.
    """
    cat, fd = haken
    _artikel(db, cat, "Jacke", fd, "true")
    _artikel(db, cat, "Hose", fd, "false")
    _artikel(db, cat, "Schal")          # nie angefasst, gar keine Zeile
    v = _vorlage(db, ["name"], art="products",
                 feldfilter=[{"feld_id": fd.id, "wert": "nein"}])

    assert sorted(_namen(db, v)) == ["Hose", "Schal"]


def test_leerer_wert_trifft_artikel_ohne_wert(db, kabel):
    cat, fd = kabel
    _artikel(db, cat, "Mit Laenge", fd, "5")
    _artikel(db, cat, "Ohne Laenge")
    v = _vorlage(db, ["name"], art="products",
                 feldfilter=[{"feld_id": fd.id, "wert": ""}])

    assert _namen(db, v) == ["Ohne Laenge"]


def test_feldfilter_vergleicht_genau_nicht_teilweise(db, kabel):
    cat, fd = kabel
    _artikel(db, cat, "Kurz", fd, "3")
    _artikel(db, cat, "Lang", fd, "30")
    v = _vorlage(db, ["name"], art="products",
                 feldfilter=[{"feld_id": fd.id, "wert": "3"}])

    assert _namen(db, v) == ["Kurz"]


def test_mehrere_bedingungen_muessen_alle_zutreffen(db, haken):
    cat, fd_bool = haken
    fd_text = FieldDefinition(
        category_id=cat.id, label="Farbe", key="farbe", field_type="text"
    )
    db.add(fd_text)
    db.commit()
    db.refresh(fd_text)

    blau = _artikel(db, cat, "Blaue Jacke", fd_bool, "true")
    db.add(ProductFieldValue(
        product_id=blau.id, field_definition_id=fd_text.id, value="blau"
    ))
    rot = _artikel(db, cat, "Rote Jacke", fd_bool, "true")
    db.add(ProductFieldValue(
        product_id=rot.id, field_definition_id=fd_text.id, value="rot"
    ))
    db.commit()

    v = _vorlage(db, ["name"], art="products", feldfilter=[
        {"feld_id": fd_bool.id, "wert": "ja"},
        {"feld_id": fd_text.id, "wert": "blau"},
    ])

    assert _namen(db, v) == ["Blaue Jacke"]


def test_feldfilter_und_kategorie_greifen_zusammen(db, haken):
    cat, fd = haken
    zweite = Category(name="Werkzeug", tracking=CategoryTracking.object.value)
    db.add(zweite)
    db.commit()
    db.refresh(zweite)

    _artikel(db, cat, "Jacke", fd, "true")
    # Gleicher Feldwert, andere Kategorie – die Kategorie muss ihn aussperren.
    fremder = _artikel(db, zweite, "Hammer")
    db.add(ProductFieldValue(
        product_id=fremder.id, field_definition_id=fd.id, value="true"
    ))
    db.commit()

    v = _vorlage(db, ["name"], art="products", kategorie_ids=[cat.id],
                 feldfilter=[{"feld_id": fd.id, "wert": "ja"}])

    assert _namen(db, v) == ["Jacke"]


def test_feldfilter_wirkt_auch_bei_einzelstuecken(db, haken):
    """Der Wert haengt am Artikel – das Stueck erbt ihn."""
    cat, fd = haken
    verliehen = _artikel(db, cat, "Jacke", fd, "true")
    daheim = _artikel(db, cat, "Hose", fd, "false")
    db.add(Item(uid="UIDAAAAA", product_id=verliehen.id))
    db.add(Item(uid="UIDBBBBB", product_id=daheim.id))
    db.commit()

    v = _vorlage(db, ["produkt"], art="items",
                 feldfilter=[{"feld_id": fd.id, "wert": "ja"}])

    assert _namen(db, v) == ["Jacke"]


def test_geloeschtes_feld_im_filter_zerreisst_den_export_nicht(db, haken):
    """Wie bei einer Spalte zu einem geloeschten Feld: die Bedingung faellt weg.

    Der Export wird dadurch WEITER, nicht leer – die harmlosere der beiden
    Fehlrichtungen, und hier festgehalten, damit sie nicht versehentlich kippt.
    """
    cat, fd = haken
    _artikel(db, cat, "Jacke", fd, "true")
    _artikel(db, cat, "Hose", fd, "false")
    v = _vorlage(db, ["name"], art="products",
                 feldfilter=[{"feld_id": fd.id, "wert": "ja"}])
    db.delete(fd)
    db.commit()

    assert sorted(_namen(db, v)) == ["Hose", "Jacke"]


def test_kaputter_feldfilter_ergibt_keinen_filter(db):
    for nr, roh in enumerate(('{"feldfilter": "quatsch"}',
                              '{"feldfilter": [{"feld_id": "abc"}]}',
                              '{"feldfilter": [42]}')):
        v = EtikettenVorlage(name=f"Kaputt {nr}", art="items", konfiguration=roh)
        db.add(v)
        db.commit()
        assert konfiguration_lesen(v)["feldfilter"] == []


def test_feldfilter_uebersteht_die_sicherung(db, haken):
    """Feld-IDs sind zwischen zwei Instanzen nicht dieselben.

    Deshalb steht in der Sicherung „Kategorie::Label" statt einer ID – und
    zwar fuer Spalten UND Filter. Fehlte die Umschreibung beim Filter, zeigte
    er nach dem Zurueckspielen auf ein fremdes Feld oder waere still weg, und
    der Export gaebe mehr Zeilen aus als das Original.
    """
    import json as _json

    from app.routers.transfer import _import_json, export_backup_json
    from app.seed import ensure_builtin_units

    cat, fd = haken
    person = User(username="chef", password_hash="x", role=Role.admin)
    db.add(person)
    _artikel(db, cat, "Jacke", fd, "true")
    _artikel(db, cat, "Hose", fd, "false")
    v = _vorlage(db, ["name"], art="products",
                 feldfilter=[{"feld_id": fd.id, "wert": "ja"}])
    v.name = "Verliehenes"
    db.commit()

    inhalt = export_backup_json(db=db, _=person).body
    gesichert = _json.loads(inhalt)["etiketten_vorlagen"][0]
    # Nach Pfad und Label, nie nach ID.
    assert gesichert["feldfilter"] == [{"feld": "Kleidung::Verliehen", "wert": "ja"}]

    from app.database import Base
    from sqlalchemy import create_engine
    from sqlalchemy.orm import sessionmaker
    from sqlalchemy.pool import StaticPool

    motor = create_engine(
        "sqlite://", connect_args={"check_same_thread": False}, poolclass=StaticPool
    )
    Base.metadata.create_all(bind=motor)
    zweite = sessionmaker(bind=motor)()
    ensure_builtin_units(zweite)
    zweiter_nutzer = User(username="chef", password_hash="x", role=Role.admin)
    zweite.add(zweiter_nutzer)
    zweite.commit()

    _import_json(zweite, inhalt, zweiter_nutzer, "add")

    neu = zweite.query(EtikettenVorlage).filter(
        EtikettenVorlage.name == "Verliehenes"
    ).one()
    neues_feld = zweite.query(FieldDefinition).filter(
        FieldDefinition.label == "Verliehen"
    ).one()
    konf = konfiguration_lesen(neu)
    # Auf die ID der ZWEITEN Instanz umgeschrieben, Wert unveraendert.
    assert konf["feldfilter"] == [{"feld_id": neues_feld.id, "wert": "ja"}]
    assert _namen(zweite, neu) == ["Jacke"]
    zweite.close()