Sicherung: Obergruppen und Mindestbestaende je Ort mitsichern

Das JSON-Backup schrieb bei Gruppen nur Name, Mindestbestand und Einheit -
Gebinde und Ort-Bedarfe fehlten schon vorher, ein Restore verlor sie also
stillschweigend. Mit den Obergruppen waere die komplette Hierarchie dazu
gekommen.

Format v3: Gruppen bringen jetzt parents (als NAMEN, nicht IDs - die sind
zwischen zwei Instanzen nicht gleich), Gebinde-Felder und ihre Mindestbestaende
je Ort mit; Artikel ebenso. Ort null = "Ueberall". Der Import laeuft dafuer
zweiphasig, weil Kanten erst gesetzt werden koennen, wenn alle Gruppen und
Lagerorte existieren, und weist Ringe ab - eine beschaedigte Datei darf keinen
einschleusen, der danach jede Auswertung im Kreis laufen liesse.

v2-Sicherungen bleiben lesbar; ihnen fehlen die neuen Listen einfach. Die
CSV-Spalte "mindestbestand" meint weiterhin den Bedarf ohne Ortsangabe und
landet in der Ueberall-Zeile.

maintenance.py raeumt die n:m-Zeilen jetzt ausdruecklich ab: das Core-DELETE
nimmt sie nicht mit, und SQLite erzwingt keine Fremdschluessel.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
Scarriffle
2026-08-15 23:48:47 +02:00
parent eaacfd03e5
commit 54261b98c0
2 changed files with 253 additions and 6 deletions

View File

@@ -0,0 +1,145 @@
"""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))
assert daten["version"] == 3
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", []))