Preserve sub-component hierarchy in export/import

This commit is contained in:
2026-06-10 12:27:05 -05:00
parent d1fc80164f
commit d41fe7f4ab
2 changed files with 98 additions and 9 deletions
+28 -9
View File
@@ -1193,11 +1193,13 @@ def create_app(config_object="config"):
buf = io.StringIO() buf = io.StringIO()
w = csv.writer(buf) w = csv.writer(buf)
w.writerow(["id", "name", "battery_slots", "installed_count", w.writerow(["id", "name", "battery_slots", "installed_count",
"device_type", "battery_size", "location", "ha_entity_id", "notes"]) "device_type", "battery_size", "location", "ha_entity_id", "notes",
"parent_id", "parent_name"])
for d in rows: for d in rows:
w.writerow([d.id, d.name, d.battery_slots, d.installed_count(), w.writerow([d.id, d.name, d.battery_slots, d.installed_count(),
d.device_type or "", d.battery_size or "", d.device_type or "", d.battery_size or "",
d.location or "", d.ha_entity_id or "", d.notes or ""]) d.location or "", d.ha_entity_id or "", d.notes or "",
d.parent_id or "", d.parent.name if d.parent else ""])
return buf.getvalue() return buf.getvalue()
def _charge_logs_csv(): def _charge_logs_csv():
@@ -1296,7 +1298,9 @@ def create_app(config_object="config"):
{"id": d.id, "name": d.name, "battery_slots": d.battery_slots, {"id": d.id, "name": d.name, "battery_slots": d.battery_slots,
"installed_count": d.installed_count(), "device_type": d.device_type, "installed_count": d.installed_count(), "device_type": d.device_type,
"battery_size": d.battery_size, "location": d.location, "battery_size": d.battery_size, "location": d.location,
"ha_entity_id": d.ha_entity_id, "notes": d.notes} "ha_entity_id": d.ha_entity_id, "notes": d.notes,
"parent_id": d.parent_id,
"parent_name": d.parent.name if d.parent else None}
for d in devices for d in devices
], ],
"charge_logs": [ "charge_logs": [
@@ -1366,29 +1370,33 @@ def create_app(config_object="config"):
try: try:
# --- devices --- # --- devices ---
for d in data.get("devices", []): def _import_device(d, new_parent_id):
nonlocal devices_created, devices_skipped
old_id = d.get("id") old_id = d.get("id")
name = (d.get("name") or "").strip() name = (d.get("name") or "").strip()
if not name: if not name:
devices_skipped += 1 devices_skipped += 1
continue return
parent_key = new_parent_id if new_parent_id is not None else -1
existing = db.query(Device).filter( existing = db.query(Device).filter(
Device.parent_key == -1, Device.name == name Device.parent_key == parent_key, Device.name == name
).first() ).first()
if existing: if existing:
if old_id is not None: if old_id is not None:
device_id_map[old_id] = existing.id device_id_map[old_id] = existing.id
devices_skipped += 1 devices_skipped += 1
else: else:
slots = d.get("battery_slots")
new_dev = Device( new_dev = Device(
name = name, name = name,
battery_slots = d.get("battery_slots") or 1, battery_slots = slots if slots is not None else 1,
device_type = d.get("device_type") or None, device_type = d.get("device_type") or None,
battery_size = d.get("battery_size") or "", battery_size = d.get("battery_size") or None,
location = d.get("location") or None, location = d.get("location") or None,
ha_entity_id = d.get("ha_entity_id") or None, ha_entity_id = d.get("ha_entity_id") or None,
notes = d.get("notes") or None, notes = d.get("notes") or None,
parent_key = -1, parent_id = new_parent_id,
parent_key = parent_key,
) )
db.add(new_dev) db.add(new_dev)
db.flush() db.flush()
@@ -1396,6 +1404,17 @@ def create_app(config_object="config"):
device_id_map[old_id] = new_dev.id device_id_map[old_id] = new_dev.id
devices_created += 1 devices_created += 1
# Two passes so sub-components can resolve their parent's new id
# regardless of ordering in the payload. A sub whose parent is
# missing from the payload is imported as top-level.
device_rows = data.get("devices", [])
for d in device_rows:
if d.get("parent_id") is None:
_import_device(d, None)
for d in device_rows:
if d.get("parent_id") is not None:
_import_device(d, device_id_map.get(d.get("parent_id")))
# --- batteries --- # --- batteries ---
for b in data.get("batteries", []): for b in data.get("batteries", []):
old_id = b.get("id") old_id = b.get("id")
+70
View File
@@ -844,6 +844,76 @@ def test_full_roundtrip_export_import(client):
assert b"Import Results" in resp.data assert b"Import Results" in resp.data
def test_export_json_includes_parent_fields(client):
_setup_rc_car(client)
data = _json.loads(client.get("/export/all.json").data)
devs = {d["name"]: d for d in data["devices"]}
assert devs["RC Car Set"]["parent_id"] is None
assert devs["Remote"]["parent_id"] == devs["RC Car Set"]["id"]
assert devs["Remote"]["parent_name"] == "RC Car Set"
def test_import_rebuilds_hierarchy(client):
payload = _make_import_payload(devices=[
_dev(10, "Hub", battery_slots=0, battery_size=None),
_dev(11, "Probe", parent_id=10),
])
resp = _post_import(client, payload)
assert resp.status_code == 200
data = _json.loads(client.get("/export/all.json").data)
devs = {d["name"]: d for d in data["devices"]}
assert devs["Probe"]["parent_id"] == devs["Hub"]["id"]
assert devs["Probe"]["parent_name"] == "Hub"
# parent's 0 slots survive the import (previously coerced to 1)
assert devs["Hub"]["battery_slots"] == 0
def test_import_child_listed_before_parent(client):
payload = _make_import_payload(devices=[
_dev(11, "Probe", parent_id=10),
_dev(10, "Hub", battery_slots=0, battery_size=None),
])
resp = _post_import(client, payload)
assert resp.status_code == 200
data = _json.loads(client.get("/export/all.json").data)
devs = {d["name"]: d for d in data["devices"]}
assert devs["Probe"]["parent_id"] == devs["Hub"]["id"]
def test_import_subcomponent_missing_parent_becomes_top_level(client):
payload = _make_import_payload(devices=[_dev(11, "Probe", parent_id=99)])
resp = _post_import(client, payload)
assert resp.status_code == 200
data = _json.loads(client.get("/export/all.json").data)
devs = {d["name"]: d for d in data["devices"]}
assert devs["Probe"]["parent_id"] is None
def test_import_empty_battery_size_stored_as_null(client):
payload = _make_import_payload(devices=[_dev(10, "NoSize", battery_size="")])
resp = _post_import(client, payload)
assert resp.status_code == 200
data = _json.loads(client.get("/export/all.json").data)
devs = {d["name"]: d for d in data["devices"]}
assert devs["NoSize"]["battery_size"] is None
def test_roundtrip_import_preserves_hierarchy(client):
_setup_rc_car(client)
data = _json.loads(client.get("/export/all.json").data)
# rename everything so the import creates rows instead of skipping
for d in data["devices"]:
d["name"] = d["name"] + " v2"
buf = io.BytesIO(_json.dumps(data).encode())
resp = client.post("/import", data={"file": (buf, "export.json")},
content_type="multipart/form-data")
assert resp.status_code == 200
data2 = _json.loads(client.get("/export/all.json").data)
devs = {d["name"]: d for d in data2["devices"]}
assert devs["Remote v2"]["parent_id"] == devs["RC Car Set v2"]["id"]
assert devs["Car v2"]["parent_id"] == devs["RC Car Set v2"]["id"]
def test_device_detail_unassign_all(seeded_client): def test_device_detail_unassign_all(seeded_client):
client = seeded_client client = seeded_client
# install battery 1 into device 1 (2-slot AA device) # install battery 1 into device 1 (2-slot AA device)