Vorrania/backend/tests/test_backup_gruppen.py

"""Sicherung: Obergruppen und Mindestbestände je Ort überstehen den Umlauf.

Ohne diesen Weg gingen genau die Daten verloren, um die es beim Umbau ging –
die Kanten des Gruppen-Graphen und die „Überall"-Zeilen. Das JSON-Backup führt
Gruppen über ihren NAMEN, nicht über IDs: die sind zwischen zwei Instanzen
nicht gleich.
"""

import json

import pytest

from app.models import (
    BaseUnit,
    Group,
    GroupLocationMinStock,
    Location,
    Product,
    ProductLocationMinStock,
    Role,
    User,
)
from app.routers.transfer import _import_json, export_backup_json


@pytest.fixture()
def user(db):
    person = User(username="tester", password_hash="x", role=Role.admin)
    db.add(person)
    db.commit()
    db.refresh(person)
    return person


def _sicherung(db, user) -> bytes:
    return export_backup_json(db=db, _=user).body


def test_backup_nimmt_obergruppen_und_bedarfe_mit(db, user, monkeypatch):
    keller = Location(name="Keller")
    db.add(keller)
    db.flush()

    wurst = Group(name="Wurst")
    grillgut = Group(name="Grillgut")
    db.add_all([wurst, grillgut])
    db.flush()
    grillwurst = Group(name="Grillwurst", parents=[wurst, grillgut])
    db.add(grillwurst)
    db.flush()

    p = Product(name="Bell Grillwurst", base_unit=BaseUnit.gram, group_id=grillwurst.id)
    db.add(p)
    db.flush()
    db.add_all([
        GroupLocationMinStock(group_id=wurst.id, location_id=None, min_stock=5000),
        GroupLocationMinStock(group_id=grillwurst.id, location_id=keller.id, min_stock=2000),
        ProductLocationMinStock(product_id=p.id, location_id=None, min_stock=800),
    ])
    db.commit()

    daten = json.loads(_sicherung(db, user))
    # v4 seit den Etiketten-Vorlagen; aeltere Sicherungen bleiben lesbar.
    assert daten["version"] == 4
    nach_name = {g["name"]: g for g in daten["groups"]}
    assert sorted(nach_name["Grillwurst"]["parents"]) == ["Grillgut", "Wurst"]
    assert {"location": None, "min_stock": 5000} in nach_name["Wurst"]["min_stocks"]
    assert {"location": "Keller", "min_stock": 2000} in nach_name["Grillwurst"]["min_stocks"]
    artikel = daten["products"][0]
    assert {"location": None, "min_stock": 800} in artikel["min_stocks"]


def test_wiederherstellen_baut_den_graphen_neu_auf(db, user):
    """Sicherung aus einer Instanz, Einspielen in eine leere zweite."""
    keller = Location(name="Keller")
    db.add(keller)
    db.flush()
    wurst = Group(name="Wurst")
    grillgut = Group(name="Grillgut")
    db.add_all([wurst, grillgut])
    db.flush()
    grillwurst = Group(name="Grillwurst", parents=[wurst, grillgut])
    db.add(grillwurst)
    db.flush()
    p = Product(name="Bell Grillwurst", base_unit=BaseUnit.gram, group_id=grillwurst.id)
    db.add(p)
    db.flush()
    db.add_all([
        GroupLocationMinStock(group_id=wurst.id, location_id=None, min_stock=5000),
        ProductLocationMinStock(product_id=p.id, location_id=keller.id, min_stock=800),
    ])
    db.commit()
    inhalt = _sicherung(db, user)

    # Zweite, leere Instanz.
    from sqlalchemy import create_engine
    from sqlalchemy.orm import sessionmaker
    from sqlalchemy.pool import StaticPool

    from app.database import Base
    from app.seed import ensure_builtin_units

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

    neu_wurst = zweite.query(Group).filter(Group.name == "Wurst").one()
    neu_grillwurst = zweite.query(Group).filter(Group.name == "Grillwurst").one()
    assert sorted(g.name for g in neu_grillwurst.parents) == ["Grillgut", "Wurst"]
    assert neu_wurst.location_min_stocks[0].location_id is None
    assert neu_wurst.location_min_stocks[0].min_stock == pytest.approx(5000)

    neu_artikel = zweite.query(Product).filter(Product.name == "Bell Grillwurst").one()
    orte = {
        (e.location.name if e.location else None): e.min_stock
        for e in neu_artikel.location_min_stocks
    }
    assert orte == {"Keller": pytest.approx(800)}
    zweite.close()
    engine.dispose()


def test_ring_aus_der_sicherung_wird_abgewiesen(db, user):
    """Eine beschädigte Datei darf keinen Ring einschleusen."""
    daten = {
        "version": 3,
        "groups": [
            {"name": "A", "parents": ["B"]},
            {"name": "B", "parents": ["A"]},
        ],
        "products": [],
    }
    ergebnis = _import_json(db, json.dumps(daten).encode(), user, "add")

    a = db.query(Group).filter(Group.name == "A").one()
    b = db.query(Group).filter(Group.name == "B").one()
    # Eine der beiden Kanten greift, die andere wird als Ring abgewiesen.
    assert not (a.parents and b.parents)
    assert any("Ring" in f for f in ergebnis.get("errors", []))


def test_sicherung_nimmt_zweiteinheit_mit(db, user):
    """Ohne das ginge die Brücke beim Wiederherstellen stillschweigend verloren."""
    p = Product(name="Bratwurst", base_unit=BaseUnit.piece,
                secondary_base="gram", secondary_count=3, secondary_amount=250)
    db.add(p)
    db.commit()

    daten = json.loads(_sicherung(db, user))
    artikel = daten["products"][0]
    assert artikel["secondary_base"] == "gram"
    assert artikel["secondary_count"] == 3
    assert artikel["secondary_amount"] == 250