fix: tolerate same device on multiple canvases in zigbee/zwave import

Zigbee and Z-Wave imports looked up canvas nodes by ieee_address with
scalar_one_or_none(), assuming one node per IEEE globally. A device placed
on two designs (one Node per canvas — a supported feature) made re-import
crash with MultipleResultsFound.

- Both mesh imports now refresh properties on every matching node instead
  of a single row (loop over .scalars().all()).
- approve_device guards against a true duplicate: same IEEE already on the
  SAME design reuses that node instead of inserting a second one.
- New node_dedupe service: loss-free repair keyed on (ieee, design_id).
  Collapses only genuine same-canvas duplicates (merges properties/services/
  missing fields, re-points edges + parent_id, drops self-loops/parallel
  edges). Cross-design placements preserved. Runs at start of both imports
  and bulk-approve. No-op on healthy DBs.
- IP correlation path already handled multiple nodes; left unchanged.

Tests: dedupe unit tests (collapse, cross-design preservation, edge/parent
re-point, idempotent), zigbee + zwave multi-canvas regression, approve
no-dupe guard.

ha-relevant: yes
This commit is contained in:
Pouzor
2026-07-03 01:52:56 +02:00
parent b0cf0deab0
commit 60383bee64
8 changed files with 485 additions and 32 deletions
+37 -1
View File
@@ -15,8 +15,12 @@ from app.db.database import AsyncSessionLocal, get_db
from app.db.models import Design, Edge, Node, PendingDevice, PendingDeviceLink, ScanRun from app.db.models import Design, Edge, Node, PendingDevice, PendingDeviceLink, ScanRun
from app.schemas.nodes import NodeCreate from app.schemas.nodes import NodeCreate
from app.schemas.scan import PendingDeviceResponse, ScanRunResponse from app.schemas.scan import PendingDeviceResponse, ScanRunResponse
from app.services.node_dedupe import dedupe_nodes_by_ieee
from app.services.scanner import DeepScanOptions, _valid_port_range, request_cancel, run_scan from app.services.scanner import DeepScanOptions, _valid_port_range, request_cancel, run_scan
from app.services.zigbee_service import build_zigbee_properties from app.services.zigbee_service import (
build_zigbee_properties,
merge_zigbee_properties,
)
from app.services.zwave_service import build_zwave_properties from app.services.zwave_service import build_zwave_properties
_ZIGBEE_TYPES = {"zigbee_coordinator", "zigbee_router", "zigbee_enddevice"} _ZIGBEE_TYPES = {"zigbee_coordinator", "zigbee_router", "zigbee_enddevice"}
@@ -308,6 +312,9 @@ async def bulk_approve_devices(
db: AsyncSession = Depends(get_db), db: AsyncSession = Depends(get_db),
_: str = Depends(get_current_user), _: str = Depends(get_current_user),
) -> dict[str, Any]: ) -> dict[str, Any]:
# Repair any legacy same-canvas duplicate nodes before placing more.
await dedupe_nodes_by_ieee(db)
# Target the design the user is on; fall back to the first design. # Target the design the user is on; fall back to the first design.
default_design_id = payload.design_id default_design_id = payload.design_id
if default_design_id is None: if default_design_id is None:
@@ -469,6 +476,35 @@ async def approve_device(
raise HTTPException(status_code=409, detail="Device already processed") raise HTTPException(status_code=409, detail="Device already processed")
device.status = "approved" device.status = "approved"
wireless = _is_wireless(node_data.type) wireless = _is_wireless(node_data.type)
# Guard against a true duplicate: same IEEE already placed on THIS design.
# (The same device on a *different* design is valid — one Node per canvas.)
if device.ieee_address is not None:
dup = (
await db.execute(
select(Node).where(
Node.ieee_address == device.ieee_address,
Node.design_id == node_design_id,
)
)
).scalars().first()
if dup is not None:
dup.properties = merge_zigbee_properties(
dup.properties,
_wireless_properties(
node_data.type, device.ieee_address,
device.vendor, device.model, device.lqi,
) if wireless else [],
)
edges = await _resolve_pending_links_for_ieee(db, device.ieee_address)
await db.commit()
return {
"approved": True,
"node_id": dup.id,
"edges_created": len(edges),
"edges": edges,
}
# Prefer the MAC discovered during the scan (stored on the pending device); # Prefer the MAC discovered during the scan (stored on the pending device);
# fall back to whatever the approve payload carried. # fall back to whatever the approve payload carried.
_mac = device.mac or node_data.mac _mac = device.mac or node_data.mac
+30 -16
View File
@@ -23,6 +23,7 @@ from app.schemas.zigbee import (
ZigbeeTestConnectionRequest, ZigbeeTestConnectionRequest,
ZigbeeTestConnectionResponse, ZigbeeTestConnectionResponse,
) )
from app.services.node_dedupe import dedupe_nodes_by_ieee
from app.services.zigbee_service import ( from app.services.zigbee_service import (
build_zigbee_properties, build_zigbee_properties,
fetch_networkmap, fetch_networkmap,
@@ -138,6 +139,10 @@ async def _persist_pending_import(
Coordinator auto-approves to a canvas Node. Other devices upsert by IEEE. Coordinator auto-approves to a canvas Node. Other devices upsert by IEEE.
All zigbee-source links are wiped and re-inserted from the new map. All zigbee-source links are wiped and re-inserted from the new map.
""" """
# Repair any pre-existing duplicate nodes (same IEEE) before upserting, so
# the by-IEEE lookups below resolve to a single row.
await dedupe_nodes_by_ieee(db)
# Determine target design (use first design as fallback) # Determine target design (use first design as fallback)
first_design = (await db.execute(select(Design).order_by(Design.created_at).limit(1))).scalar() first_design = (await db.execute(select(Design).order_by(Design.created_at).limit(1))).scalar()
default_design_id = first_design.id if first_design else None default_design_id = first_design.id if first_design else None
@@ -156,15 +161,22 @@ async def _persist_pending_import(
) )
if n.get("device_type") == "Coordinator": if n.get("device_type") == "Coordinator":
existing = await db.execute(select(Node).where(Node.ieee_address == ieee)) # The coordinator may sit on several canvases (one Node per design).
existing_node = existing.scalar_one_or_none() # Refresh props on every matching node, not just one.
if existing_node: existing_nodes = (
existing_node.properties = merge_zigbee_properties( await db.execute(
existing_node.properties, props select(Node).where(Node.ieee_address == ieee).order_by(Node.id)
) )
).scalars().all()
if existing_nodes:
for existing_node in existing_nodes:
existing_node.properties = merge_zigbee_properties(
existing_node.properties, props
)
first = existing_nodes[0]
coordinator_out = ZigbeeCoordinatorOut( coordinator_out = ZigbeeCoordinatorOut(
id=existing_node.id, id=first.id,
label=existing_node.label, label=first.label,
ieee_address=ieee, ieee_address=ieee,
) )
coordinator_existed = True coordinator_existed = True
@@ -188,16 +200,18 @@ async def _persist_pending_import(
continue continue
# If the device has already been approved as a canvas Node, refresh # If the device has already been approved as a canvas Node, refresh
# its properties and skip creating a pending row (keeps approved # its properties on every canvas it sits on and skip creating a pending
# devices out of pending/hidden modals on re-import). # row (keeps approved devices out of pending/hidden modals on re-import).
existing_node_q = await db.execute( existing_nodes = (
select(Node).where(Node.ieee_address == ieee) await db.execute(
) select(Node).where(Node.ieee_address == ieee).order_by(Node.id)
existing_node = existing_node_q.scalar_one_or_none()
if existing_node:
existing_node.properties = merge_zigbee_properties(
existing_node.properties, props
) )
).scalars().all()
if existing_nodes:
for existing_node in existing_nodes:
existing_node.properties = merge_zigbee_properties(
existing_node.properties, props
)
continue continue
result = await db.execute( result = await db.execute(
+30 -15
View File
@@ -23,6 +23,7 @@ from app.schemas.zwave import (
ZwaveTestConnectionRequest, ZwaveTestConnectionRequest,
ZwaveTestConnectionResponse, ZwaveTestConnectionResponse,
) )
from app.services.node_dedupe import dedupe_nodes_by_ieee
from app.services.zwave_service import ( from app.services.zwave_service import (
build_zwave_properties, build_zwave_properties,
fetch_zwave_network, fetch_zwave_network,
@@ -134,6 +135,10 @@ async def _persist_pending_import(
Coordinator auto-approves to a canvas Node. Other devices upsert by Z-Wave Coordinator auto-approves to a canvas Node. Other devices upsert by Z-Wave
identity. All zwave-source links are wiped and re-inserted from the new map. identity. All zwave-source links are wiped and re-inserted from the new map.
""" """
# Repair any pre-existing same-canvas duplicate nodes before upserting, so
# the by-IEEE lookups below resolve cleanly.
await dedupe_nodes_by_ieee(db)
first_design = (await db.execute(select(Design).order_by(Design.created_at).limit(1))).scalar() first_design = (await db.execute(select(Design).order_by(Design.created_at).limit(1))).scalar()
default_design_id = first_design.id if first_design else None default_design_id = first_design.id if first_design else None
@@ -149,15 +154,22 @@ async def _persist_pending_import(
props = build_zwave_properties(ieee, n.get("vendor"), n.get("model")) props = build_zwave_properties(ieee, n.get("vendor"), n.get("model"))
if n.get("type") == "zwave_coordinator": if n.get("type") == "zwave_coordinator":
existing = await db.execute(select(Node).where(Node.ieee_address == ieee)) # The coordinator may sit on several canvases (one Node per design).
existing_node = existing.scalar_one_or_none() # Refresh props on every matching node, not just one.
if existing_node: existing_nodes = (
existing_node.properties = merge_zwave_properties( await db.execute(
existing_node.properties, props select(Node).where(Node.ieee_address == ieee).order_by(Node.id)
) )
).scalars().all()
if existing_nodes:
for existing_node in existing_nodes:
existing_node.properties = merge_zwave_properties(
existing_node.properties, props
)
first = existing_nodes[0]
coordinator_out = ZwaveCoordinatorOut( coordinator_out = ZwaveCoordinatorOut(
id=existing_node.id, id=first.id,
label=existing_node.label, label=first.label,
ieee_address=ieee, ieee_address=ieee,
) )
coordinator_existed = True coordinator_existed = True
@@ -180,15 +192,18 @@ async def _persist_pending_import(
) )
continue continue
# Already approved as a canvas Node → refresh props, skip pending row. # Already approved as a canvas Node → refresh props on every canvas it
existing_node_q = await db.execute( # sits on, skip pending row.
select(Node).where(Node.ieee_address == ieee) existing_nodes = (
) await db.execute(
existing_node = existing_node_q.scalar_one_or_none() select(Node).where(Node.ieee_address == ieee).order_by(Node.id)
if existing_node:
existing_node.properties = merge_zwave_properties(
existing_node.properties, props
) )
).scalars().all()
if existing_nodes:
for existing_node in existing_nodes:
existing_node.properties = merge_zwave_properties(
existing_node.properties, props
)
continue continue
result = await db.execute( result = await db.execute(
+161
View File
@@ -0,0 +1,161 @@
"""Collapse *true* duplicate canvas nodes: same ``ieee_address`` **and** same
``design_id`` (i.e. the same device placed twice on the *same* canvas).
The same device legitimately appears on multiple canvases — placing a device on
another design creates a second :class:`Node` for that IEEE by design (see the
per-design guard in ``bulk_approve_devices``). Those cross-design rows are NOT
duplicates and must be preserved. Only two rows sharing both the IEEE *and* the
design are corrupt, and only those are collapsed here.
This module provides an idempotent, **loss-free** repair: per ``(ieee,
design_id)`` group with >1 node it keeps the oldest as canonical, merges the
extras' data into it (properties, missing scalar fields, services), re-points
every edge and ``parent_id`` reference onto the canonical node, then deletes the
extras. Edges are de-duplicated and self-loops dropped after re-pointing so no
dangling or redundant links remain.
"""
from __future__ import annotations
import logging
from typing import Any
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from app.db.models import Edge, Node
from app.services.zigbee_service import merge_zigbee_properties
logger = logging.getLogger(__name__)
# Scalar Node fields worth carrying over from a duplicate when the canonical
# node has no value. Positions, design, parent and identity fields are left on
# the canonical node untouched (that's the row the user actually placed).
_FILLABLE_FIELDS = (
"hostname",
"ip",
"mac",
"os",
"check_method",
"check_target",
"notes",
"cpu_count",
"cpu_model",
"ram_gb",
"disk_gb",
"custom_icon",
"last_seen",
"last_scan",
)
def _merge_services(
a: list[Any] | None, b: list[Any] | None
) -> list[Any]:
"""Union two service lists, de-duplicated, order-stable."""
out: list[Any] = list(a or [])
seen = {repr(s) for s in out}
for s in b or []:
if repr(s) not in seen:
out.append(s)
seen.add(repr(s))
return out
def _merge_into_canonical(canonical: Node, dup: Node) -> None:
"""Fold ``dup``'s data into ``canonical`` in place (no field lost)."""
canonical.properties = merge_zigbee_properties(
canonical.properties, dup.properties or []
)
canonical.services = _merge_services(canonical.services, dup.services)
for field in _FILLABLE_FIELDS:
if getattr(canonical, field, None) in (None, "") and getattr(dup, field, None) not in (None, ""):
setattr(canonical, field, getattr(dup, field))
# Prefer a human label over a bare IEEE/hex fallback.
if (not canonical.label or canonical.label == canonical.ieee_address) and dup.label:
canonical.label = dup.label
async def dedupe_nodes_by_ieee(db: AsyncSession) -> int:
"""Merge duplicate nodes sharing an ``ieee_address`` AND ``design_id``.
Returns the number of nodes removed. Idempotent: a no-op when every
``(ieee, design)`` pair maps to at most one node. Nodes with the same IEEE
on *different* designs are left untouched (valid cross-canvas placement).
Does not commit — the caller owns the transaction.
"""
rows = (
await db.execute(
select(Node)
.where(Node.ieee_address.is_not(None))
.order_by(Node.ieee_address, Node.created_at, Node.id)
)
).scalars().all()
groups: dict[tuple[str, str | None], list[Node]] = {}
for node in rows:
groups.setdefault((node.ieee_address, node.design_id), []).append(node) # type: ignore[arg-type]
removed = 0
for (ieee, _design), nodes in groups.items():
if len(nodes) < 2:
continue
canonical, *dups = nodes # oldest first (ordered above)
dup_ids = {d.id for d in dups}
for dup in dups:
_merge_into_canonical(canonical, dup)
# Re-point edges + parents, then drop self-loops / duplicates.
edges = (
await db.execute(
select(Edge).where(
Edge.source.in_(dup_ids) | Edge.target.in_(dup_ids)
)
)
).scalars().all()
for edge in edges:
if edge.source in dup_ids:
edge.source = canonical.id
if edge.target in dup_ids:
edge.target = canonical.id
# Re-point children whose parent was a duplicate.
children = (
await db.execute(select(Node).where(Node.parent_id.in_(dup_ids)))
).scalars().all()
for child in children:
child.parent_id = canonical.id
# Collapse self-loops and now-redundant parallel edges.
all_edges = (
await db.execute(
select(Edge).where(
(Edge.source == canonical.id) | (Edge.target == canonical.id)
)
)
).scalars().all()
seen_pairs: set[tuple[str, str, str]] = set()
for edge in all_edges:
if edge.source == edge.target:
await db.delete(edge)
continue
key = (edge.source, edge.target, edge.type)
if key in seen_pairs:
await db.delete(edge)
continue
seen_pairs.add(key)
await db.flush()
for dup in dups:
await db.delete(dup)
removed += 1
logger.info(
"Deduped IEEE %s: merged %d duplicate node(s) into %s",
ieee, len(dups), canonical.id,
)
if removed:
await db.flush()
return removed
+117
View File
@@ -0,0 +1,117 @@
"""Tests for the same-canvas node dedupe repair (app.services.node_dedupe)."""
import pytest
from sqlalchemy import select
from app.db.models import Design, Edge, Node
from app.services.node_dedupe import dedupe_nodes_by_ieee
async def _design(db, name="d1"):
d = Design(name=name)
db.add(d)
await db.flush()
return d
@pytest.mark.asyncio
async def test_collapses_same_ieee_same_design(db_session):
d = await _design(db_session)
keep = Node(
label="Sensor", type="zigbee_enddevice", design_id=d.id,
ieee_address="0xAAA", properties=[{"key": "IEEE", "value": "0xAAA", "visible": True}],
pos_x=100, pos_y=200,
)
db_session.add(keep)
await db_session.flush()
dup = Node(
label="Sensor", type="zigbee_enddevice", design_id=d.id,
ieee_address="0xAAA", ip="10.0.0.5",
properties=[{"key": "LQI", "value": "88", "visible": False}],
)
db_session.add(dup)
await db_session.flush()
removed = await dedupe_nodes_by_ieee(db_session)
assert removed == 1
nodes = (await db_session.execute(select(Node).where(Node.ieee_address == "0xAAA"))).scalars().all()
assert len(nodes) == 1
survivor = nodes[0]
assert survivor.id == keep.id # oldest kept
assert survivor.pos_x == 100 # canvas position preserved
assert survivor.ip == "10.0.0.5" # missing field filled from dup
keys = {p["key"] for p in survivor.properties}
assert keys == {"IEEE", "LQI"} # properties merged
@pytest.mark.asyncio
async def test_preserves_same_ieee_across_designs(db_session):
"""Same device on two canvases is valid — must NOT be merged."""
d1 = await _design(db_session, "d1")
d2 = await _design(db_session, "d2")
for d in (d1, d2):
db_session.add(Node(label="S", type="zigbee_enddevice", design_id=d.id, ieee_address="0xBBB"))
await db_session.flush()
removed = await dedupe_nodes_by_ieee(db_session)
assert removed == 0
nodes = (await db_session.execute(select(Node).where(Node.ieee_address == "0xBBB"))).scalars().all()
assert len(nodes) == 2
@pytest.mark.asyncio
async def test_repoints_edges_and_drops_dupes(db_session):
d = await _design(db_session)
keep = Node(label="A", type="server", design_id=d.id, ieee_address="0xCCC")
dup = Node(label="A", type="server", design_id=d.id, ieee_address="0xCCC")
other = Node(label="B", type="server", design_id=d.id)
db_session.add_all([keep, other])
await db_session.flush()
db_session.add(dup)
await db_session.flush()
# keep<->other and dup<->other (parallel after repoint), plus dup<->keep (self-loop).
db_session.add_all([
Edge(source=keep.id, target=other.id, type="ethernet", design_id=d.id),
Edge(source=dup.id, target=other.id, type="ethernet", design_id=d.id),
Edge(source=dup.id, target=keep.id, type="ethernet", design_id=d.id),
])
await db_session.flush()
removed = await dedupe_nodes_by_ieee(db_session)
assert removed == 1
edges = (await db_session.execute(select(Edge))).scalars().all()
# self-loop dropped, parallel edge collapsed -> a single keep<->other edge
assert len(edges) == 1
e = edges[0]
assert {e.source, e.target} == {keep.id, other.id}
@pytest.mark.asyncio
async def test_repoints_child_parent(db_session):
d = await _design(db_session)
keep = Node(label="Host", type="proxmox", design_id=d.id, ieee_address="0xDDD")
dup = Node(label="Host", type="proxmox", design_id=d.id, ieee_address="0xDDD")
db_session.add(keep)
await db_session.flush()
db_session.add(dup)
await db_session.flush()
child = Node(label="VM", type="vm", design_id=d.id, parent_id=dup.id)
db_session.add(child)
await db_session.flush()
await dedupe_nodes_by_ieee(db_session)
await db_session.refresh(child)
assert child.parent_id == keep.id
@pytest.mark.asyncio
async def test_idempotent_noop_when_unique(db_session):
d = await _design(db_session)
db_session.add(Node(label="X", type="server", design_id=d.id, ieee_address="0xEEE"))
await db_session.flush()
assert await dedupe_nodes_by_ieee(db_session) == 0
assert await dedupe_nodes_by_ieee(db_session) == 0
+38
View File
@@ -311,6 +311,44 @@ async def test_approve_device(client: AsyncClient, headers, pending_device):
assert inventory[0]["status"] == "approved" assert inventory[0]["status"] == "approved"
@pytest.mark.asyncio
async def test_approve_device_no_dupe_same_ieee_same_design(
client: AsyncClient, headers, db_session
):
"""Approving a device whose IEEE is already on the target design must reuse
the existing node, not create a duplicate (source of the crash bug)."""
design = Design(name="d1")
db_session.add(design)
await db_session.flush()
existing = Node(
label="sensor", type="zigbee_enddevice", ieee_address="0xZZZ",
services=[], design_id=design.id,
)
db_session.add(existing)
device = PendingDevice(
id=str(uuid.uuid4()), ieee_address="0xZZZ", suggested_type="zigbee_enddevice",
status="pending", discovery_source="zigbee",
)
db_session.add(device)
await db_session.commit()
res = await client.post(
f"/api/v1/scan/pending/{device.id}/approve",
json={
"label": "sensor", "type": "zigbee_enddevice",
"status": "online", "services": [], "design_id": design.id,
},
headers=headers,
)
assert res.status_code == 200
assert res.json()["node_id"] == existing.id # reused, not new
nodes = (
await db_session.execute(select(Node).where(Node.ieee_address == "0xZZZ"))
).scalars().all()
assert len(nodes) == 1 # no duplicate created
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_approve_nonexistent_device(client: AsyncClient, headers): async def test_approve_nonexistent_device(client: AsyncClient, headers):
node_payload = { node_payload = {
+37
View File
@@ -618,6 +618,43 @@ async def test_persist_pending_import_refreshes_existing_coordinator_properties(
assert by_key["Model"]["visible"] is False assert by_key["Model"]["visible"] is False
@pytest.mark.asyncio
async def test_persist_pending_import_device_on_multiple_canvases(
db_session,
) -> None:
"""Regression: a device approved onto TWO designs (one Node each) must not
crash re-import with MultipleResultsFound — props refresh on both nodes."""
from sqlalchemy import select
from app.api.routes.zigbee import _persist_pending_import
from app.db.models import Design, Node
d1 = Design(name="d1")
d2 = Design(name="d2")
db_session.add_all([d1, d2])
await db_session.flush()
for d in (d1, d2):
db_session.add(Node(
label="router_1", type="zigbee_router", status="online",
check_method="none", ieee_address="0xR1", services=[],
properties=[], design_id=d.id,
))
await db_session.commit()
bumped = [dict(n) for n in _PENDING_NODES]
bumped[1]["lqi"] = 240
# Must not raise.
await _persist_pending_import(db_session, bumped, _PENDING_EDGES)
nodes = (
await db_session.execute(select(Node).where(Node.ieee_address == "0xR1"))
).scalars().all()
assert len(nodes) == 2 # both canvas placements preserved
for n in nodes:
lqi = {p["key"]: p["value"] for p in n.properties}.get("LQI")
assert lqi == "240" # refreshed on every canvas
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_import_pending_requires_auth(client: AsyncClient) -> None: async def test_import_pending_requires_auth(client: AsyncClient) -> None:
res = await client.post( res = await client.post(
+35
View File
@@ -445,3 +445,38 @@ async def test_persist_keeps_hidden_hidden(db_session) -> None:
) )
).scalar_one() ).scalar_one()
assert still_hidden.status == "hidden" assert still_hidden.status == "hidden"
@pytest.mark.asyncio
async def test_persist_device_on_multiple_canvases(db_session) -> None:
"""Regression: a device on TWO designs (one Node each) must not crash
re-import with MultipleResultsFound — props refresh on both nodes."""
from sqlalchemy import select
from app.api.routes.zwave import _persist_pending_import
from app.db.models import Design, Node
d1 = Design(name="d1")
d2 = Design(name="d2")
db_session.add_all([d1, d2])
await db_session.flush()
for d in (d1, d2):
db_session.add(Node(
label="Wall Plug", type="zwave_router", status="online",
check_method="none", ieee_address="zwave-0xh-2", services=[],
properties=[], design_id=d.id,
))
await db_session.commit()
bumped = [dict(n) for n in _PENDING_NODES]
bumped[1]["model"] = "ZW200"
# Must not raise.
await _persist_pending_import(db_session, bumped, _PENDING_EDGES)
nodes = (
await db_session.execute(select(Node).where(Node.ieee_address == "zwave-0xh-2"))
).scalars().all()
assert len(nodes) == 2 # both canvas placements preserved
for n in nodes:
model = {p["key"]: p["value"] for p in n.properties}.get("Model")
assert model == "ZW200" # refreshed on every canvas