6e48cd0a7a
- Backend: services/master_data.py exportiert/importiert Kategorien (inkl. eigener Felder), Lagerorte, Einheiten und Gebinde als JSON – MIT IDs, damit gedruckte QR-Codes und Verweise nach dem Wiederherstellen passen. Import-Modus skip (Vorhandenes lassen) oder overwrite (per ID aktualisieren); Namens- Konflikte werden uebersprungen, nicht abgebrochen. Postgres-Sequenzen werden danach angehoben. Endpunkte GET /export/master-data, POST /import/master-data. 5 neue Tests, Suite 162 gruen. - Web: neue Karte "Stammdaten sichern & wiederherstellen (JSON)" in Import/Export. - Web: beim Lagerort-Anlegen bleibt der uebergeordnete Ort in der Auswahl. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
92 lines
3.2 KiB
Python
92 lines
3.2 KiB
Python
"""Stammdaten-Export/Import (JSON) mit ID-Erhalt und Konfliktmodi."""
|
|
|
|
import pytest
|
|
from sqlalchemy import create_engine
|
|
from sqlalchemy.orm import sessionmaker
|
|
from sqlalchemy.pool import StaticPool
|
|
|
|
from app.database import Base
|
|
from app.models import Category, FieldDefinition, Location, PackageType, Unit
|
|
from app.services.master_data import (
|
|
FORMAT_KEY,
|
|
export_master_data,
|
|
import_master_data,
|
|
)
|
|
|
|
|
|
def _fresh_db():
|
|
engine = create_engine(
|
|
"sqlite://", connect_args={"check_same_thread": False}, poolclass=StaticPool
|
|
)
|
|
Base.metadata.create_all(bind=engine)
|
|
return sessionmaker(bind=engine, autoflush=False, autocommit=False)()
|
|
|
|
|
|
def test_roundtrip_erhaelt_ids_und_baum(db):
|
|
ober = Category(name="Elektronik", tracking="object")
|
|
db.add(ober)
|
|
db.flush()
|
|
unter = Category(name="Kabel", parent_id=ober.id, tracking="object")
|
|
db.add(unter)
|
|
db.flush()
|
|
db.add(FieldDefinition(category_id=ober.id, label="Kapazität", key="kapazitaet",
|
|
field_type="number", unit="mAh"))
|
|
haus = Location(name="Hedingen")
|
|
db.add(haus)
|
|
db.flush()
|
|
keller = Location(name="Keller", parent_id=haus.id)
|
|
db.add(keller)
|
|
beutel = PackageType(singular="Beutel", plural="Beutel")
|
|
db.add(beutel)
|
|
db.commit()
|
|
beutel_id = beutel.id
|
|
|
|
data = export_master_data(db)
|
|
|
|
ziel = _fresh_db()
|
|
import_master_data(ziel, data, mode="skip")
|
|
|
|
assert ziel.get(Category, ober.id).name == "Elektronik"
|
|
assert ziel.get(Category, unter.id).parent_id == ober.id # Baum erhalten
|
|
assert ziel.get(Location, keller.id).parent_id == haus.id
|
|
assert ziel.query(FieldDefinition).filter_by(category_id=ober.id).first().label == "Kapazität"
|
|
assert ziel.get(PackageType, beutel_id).singular == "Beutel" # id erhalten
|
|
|
|
|
|
def test_skip_laesst_vorhandenes_unangetastet(db):
|
|
a = Location(name="Alt")
|
|
db.add(a)
|
|
db.commit()
|
|
data = {FORMAT_KEY: 1, "locations": [{"id": a.id, "name": "Neu"}]}
|
|
rep = import_master_data(db, data, mode="skip")
|
|
assert db.get(Location, a.id).name == "Alt"
|
|
assert rep["skipped"].get("locations") == 1
|
|
|
|
|
|
def test_overwrite_aktualisiert_per_id(db):
|
|
a = Location(name="Alt")
|
|
db.add(a)
|
|
db.commit()
|
|
data = {FORMAT_KEY: 1, "locations": [{"id": a.id, "name": "Neu"}]}
|
|
rep = import_master_data(db, data, mode="overwrite")
|
|
assert db.get(Location, a.id).name == "Neu"
|
|
assert rep["updated"].get("locations") == 1
|
|
|
|
|
|
def test_falsche_datei_wirft():
|
|
with pytest.raises(ValueError):
|
|
import_master_data(_fresh_db(), {"foo": 1})
|
|
|
|
|
|
def test_namenskonflikt_wird_uebersprungen_nicht_abgebrochen(db):
|
|
# Vorhandene Einheit unter id X; Import bringt gleichen Namen unter anderer id.
|
|
vorhanden = Unit(name="Spezialeinheit", kind="count", factor=1)
|
|
db.add(vorhanden)
|
|
db.commit()
|
|
data = {FORMAT_KEY: 1, "units": [{"id": vorhanden.id + 999, "name": "Spezialeinheit",
|
|
"kind": "count", "factor": 1}]}
|
|
rep = import_master_data(db, data, mode="skip")
|
|
# Gleicher Name unter neuer id -> Eindeutigkeit verletzt -> uebersprungen.
|
|
assert rep["skipped"].get("units") == 1
|
|
assert db.query(Unit).filter_by(name="Spezialeinheit").count() == 1
|