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=","):
    v = EtikettenVorlage(
        name=f"Test {len(db.query(EtikettenVorlage).all())}",
        art=art,
        konfiguration=json.dumps({
            "spalten": spalten,
            "kategorie_ids": kategorie_ids or [],
            "trennzeichen": trenn,
        }),
    )
    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": ","}


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