Pouzor
9b77e43ca1
Completes the split: `nodes` now holds only how a device is drawn on one canvas, and every device fact reaches the API from the inventory row. - The device columns are removed from `nodes` (SQLite table rebuild, the same shape as the existing device_inventory and canvas_state rebuilds). The drop is skipped, and logged, while any non-furniture node is still unlinked: those columns are the last copy of that node's facts. The backfill therefore reads them with raw SQL — by the time it runs, the model no longer declares them. - The status checker iterates devices, not nodes: one check per device however many canvases draw it, writing `status_live` / `last_seen` / `response_time_ms` on the row. `/ws/status` messages carry `device_id` and the node ids they light up. Hidden devices are not probed. - Readers repointed: scanner (last_scan lands on the row), proxmox (its node tier collapses into the inventory tier, keeping only the cluster handles), zigbee/zwave (one property refresh serves every canvas), rack inventory, liveview, stats, node dedupe. - `POST /scan/pending` merges into the row that already describes the host instead of minting a second one — one device is one row, whichever way it was documented. - Standalone keeps parity: the canvas blob gains `devices`, split on save and hydrated on load. A blob written before the split still reads. The rack inventory had a related bug: a mount that names a node explicitly printed the mount's device rather than the pinned node's. It now reads the node's own row. Tests that built a node with device columns are ported to the link; where a behaviour genuinely moved (properties refresh once on the row, last_scan is the device's) the assertion moved with it rather than being dropped. ha-relevant: yes
353 lines
12 KiB
Python
353 lines
12 KiB
Python
"""MAC-property builders and Zigbee/Z-Wave property population on approve."""
|
|
import uuid
|
|
|
|
import pytest
|
|
from httpx import AsyncClient
|
|
from sqlalchemy import select
|
|
|
|
from app.db.models import InventoryDevice, Node
|
|
from tests.scan.helpers import _add_design, _seed_zigbee_pending_pair
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_approve_zigbee_device_populates_properties(
|
|
client: AsyncClient, headers, zigbee_pending_device, db_session
|
|
):
|
|
"""Approving a zigbee device must populate IEEE/Vendor/Model/LQI in properties."""
|
|
from sqlalchemy import select
|
|
|
|
from app.db.models import Node as NodeModel
|
|
payload = {
|
|
"label": "bulb_1",
|
|
"type": "zigbee_enddevice",
|
|
"status": "online",
|
|
"services": [],
|
|
"check_method": "none",
|
|
}
|
|
res = await client.post(
|
|
f"/api/v1/scan/pending/{zigbee_pending_device.id}/approve",
|
|
json=payload,
|
|
headers=headers,
|
|
)
|
|
assert res.status_code == 200
|
|
node = (
|
|
await db_session.execute(select(NodeModel).where(NodeModel.device_id.is_not(None)))
|
|
).scalars().first()
|
|
assert node is not None
|
|
# The device facts belong to the inventory row the node draws.
|
|
device = await db_session.get(InventoryDevice, node.device_id)
|
|
assert device is not None
|
|
keys = {p["key"]: p["value"] for p in device.properties}
|
|
assert keys == {
|
|
"IEEE": "0xABCDEF",
|
|
"Vendor": "IKEA",
|
|
"Model": "TRADFRI",
|
|
"LQI": "180",
|
|
}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_bulk_approve_zigbee_populates_properties(
|
|
client: AsyncClient, headers, zigbee_pending_device, db_session
|
|
):
|
|
from sqlalchemy import select
|
|
|
|
from app.db.models import Node as NodeModel
|
|
res = await client.post(
|
|
"/api/v1/scan/pending/bulk-approve",
|
|
json={"device_ids": [zigbee_pending_device.id]},
|
|
headers=headers,
|
|
)
|
|
assert res.status_code == 200
|
|
node = (
|
|
await db_session.execute(select(NodeModel).where(NodeModel.device_id.is_not(None)))
|
|
).scalars().first()
|
|
assert node is not None
|
|
# The device facts belong to the inventory row the node draws.
|
|
device = await db_session.get(InventoryDevice, node.device_id)
|
|
assert device is not None
|
|
keys = {p["key"]: p["value"] for p in device.properties}
|
|
assert keys["IEEE"] == "0xABCDEF"
|
|
assert keys["Vendor"] == "IKEA"
|
|
assert keys["Model"] == "TRADFRI"
|
|
assert keys["LQI"] == "180"
|
|
assert device.check_method == "none"
|
|
|
|
|
|
def test_build_mac_property_returns_hidden_row():
|
|
from app.api.routes.scan import build_mac_property
|
|
|
|
assert build_mac_property("aa:bb:cc:dd:ee:ff") == [
|
|
{"key": "MAC", "value": "aa:bb:cc:dd:ee:ff", "icon": None, "visible": False}
|
|
]
|
|
|
|
|
|
def test_build_mac_property_empty_when_no_mac():
|
|
from app.api.routes.scan import build_mac_property
|
|
|
|
assert build_mac_property(None) == []
|
|
assert build_mac_property("") == []
|
|
|
|
|
|
def test_merge_mac_property_appends_when_absent():
|
|
from app.api.routes.scan import merge_mac_property
|
|
|
|
existing = [{"key": "Custom", "value": "x", "icon": None, "visible": True}]
|
|
merged = merge_mac_property(existing, "aa:bb:cc:dd:ee:ff")
|
|
assert {"key": "MAC", "value": "aa:bb:cc:dd:ee:ff", "icon": None, "visible": False} in merged
|
|
# Existing prop preserved untouched.
|
|
assert existing[0] in merged
|
|
|
|
|
|
def test_merge_mac_property_idempotent_and_preserves_visibility():
|
|
from app.api.routes.scan import merge_mac_property
|
|
|
|
existing = [{"key": "MAC", "value": "aa:bb:cc:dd:ee:ff", "icon": None, "visible": True}]
|
|
merged = merge_mac_property(existing, "aa:bb:cc:dd:ee:ff")
|
|
# No duplicate MAC row; user's visible=True choice kept.
|
|
macs = [p for p in merged if p["key"] == "MAC"]
|
|
assert len(macs) == 1
|
|
assert macs[0]["visible"] is True
|
|
|
|
|
|
def test_merge_mac_property_noop_without_mac():
|
|
from app.api.routes.scan import merge_mac_property
|
|
|
|
existing = [{"key": "Custom", "value": "x", "icon": None, "visible": True}]
|
|
assert merge_mac_property(existing, None) == existing
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_approve_device_does_not_duplicate_mac_property(
|
|
client: AsyncClient, headers, pending_device, db_session
|
|
):
|
|
"""If the approve payload already carries a MAC prop, don't add a second one."""
|
|
from sqlalchemy import select
|
|
|
|
from app.db.models import Node as NodeModel
|
|
res = await client.post(
|
|
f"/api/v1/scan/pending/{pending_device.id}/approve",
|
|
json={
|
|
"label": "My Server",
|
|
"type": "server",
|
|
"ip": "192.168.1.100",
|
|
"status": "unknown",
|
|
"services": [],
|
|
"properties": [
|
|
{"key": "MAC", "value": "aa:bb:cc:dd:ee:ff", "icon": None, "visible": True}
|
|
],
|
|
},
|
|
headers=headers,
|
|
)
|
|
assert res.status_code == 200
|
|
node = (
|
|
await db_session.execute(select(NodeModel).where(NodeModel.device_id.is_not(None)))
|
|
).scalars().first()
|
|
assert node is not None
|
|
# The device facts belong to the inventory row the node draws.
|
|
device = await db_session.get(InventoryDevice, node.device_id)
|
|
assert device is not None
|
|
mac_props = [p for p in device.properties if p["key"] == "MAC"]
|
|
assert len(mac_props) == 1
|
|
# User's visibility choice is preserved.
|
|
assert mac_props[0]["visible"] is True
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_approve_zigbee_creates_edge_when_other_endpoint_is_node(
|
|
client: AsyncClient, headers, db_session
|
|
):
|
|
from sqlalchemy import select
|
|
|
|
from app.db.models import Edge
|
|
|
|
coord, pending = await _seed_zigbee_pending_pair(db_session)
|
|
|
|
res = await client.post(
|
|
f"/api/v1/scan/pending/{pending.id}/approve",
|
|
json={
|
|
"label": "router_1",
|
|
"type": "zigbee_router",
|
|
"ip": None,
|
|
"status": "unknown",
|
|
"services": [],
|
|
},
|
|
headers=headers,
|
|
)
|
|
assert res.status_code == 200
|
|
data = res.json()
|
|
assert data["approved"] is True
|
|
assert data["edges_created"] == 1
|
|
|
|
edges = (await db_session.execute(select(Edge))).scalars().all()
|
|
assert len(edges) == 1
|
|
assert edges[0].source == coord.id
|
|
assert edges[0].target == data["node_id"]
|
|
assert edges[0].source_handle == "bottom"
|
|
# Bare side name (canonical stored form); renders at the top like before.
|
|
assert edges[0].target_handle == "top"
|
|
assert edges[0].type == "iot"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_approve_zigbee_skips_duplicate_edge(
|
|
client: AsyncClient, headers, db_session
|
|
):
|
|
"""Re-running the resolution does not create a second edge for the same pair."""
|
|
from sqlalchemy import select
|
|
|
|
from app.db.models import Edge, InventoryDevice, InventoryDeviceLink
|
|
|
|
coord, pending = await _seed_zigbee_pending_pair(db_session)
|
|
body = {"label": "router_1", "type": "zigbee_router", "ip": None, "status": "unknown", "services": []}
|
|
await client.post(f"/api/v1/scan/pending/{pending.id}/approve", json=body, headers=headers)
|
|
|
|
# Simulate a second pending row + link between same coord and a new device,
|
|
# but keep an existing edge in place to verify dedupe also handles
|
|
# the swapped-direction case.
|
|
new_pending = InventoryDevice(
|
|
ieee_address="0xR1B",
|
|
friendly_name="r1b",
|
|
suggested_type="zigbee_router",
|
|
status="pending",
|
|
discovery_source="zigbee",
|
|
)
|
|
db_session.add(new_pending)
|
|
db_session.add(
|
|
InventoryDeviceLink(source_ieee="0xCOORD", target_ieee="0xR1B", discovery_source="zigbee")
|
|
)
|
|
await db_session.commit()
|
|
res = await client.post(
|
|
f"/api/v1/scan/pending/{new_pending.id}/approve", json=body, headers=headers
|
|
)
|
|
assert res.json()["edges_created"] == 1 # only the new pair
|
|
edges = (await db_session.execute(select(Edge))).scalars().all()
|
|
assert len(edges) == 2 # original + new, no duplicate
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_approve_zigbee_skips_when_other_endpoint_still_pending(
|
|
client: AsyncClient, headers, db_session
|
|
):
|
|
"""Both endpoints pending → no edge yet, link row preserved for later."""
|
|
from sqlalchemy import select
|
|
|
|
from app.db.models import Edge, InventoryDevice, InventoryDeviceLink
|
|
|
|
a = InventoryDevice(
|
|
ieee_address="0xA",
|
|
friendly_name="a",
|
|
suggested_type="zigbee_router",
|
|
status="pending",
|
|
discovery_source="zigbee",
|
|
)
|
|
b = InventoryDevice(
|
|
ieee_address="0xB",
|
|
friendly_name="b",
|
|
suggested_type="zigbee_enddevice",
|
|
status="pending",
|
|
discovery_source="zigbee",
|
|
)
|
|
db_session.add_all([a, b])
|
|
db_session.add(
|
|
InventoryDeviceLink(source_ieee="0xA", target_ieee="0xB", discovery_source="zigbee")
|
|
)
|
|
await db_session.commit()
|
|
|
|
res = await client.post(
|
|
f"/api/v1/scan/pending/{a.id}/approve",
|
|
json={
|
|
"label": "a",
|
|
"type": "zigbee_router",
|
|
"ip": None,
|
|
"status": "unknown",
|
|
"services": [],
|
|
},
|
|
headers=headers,
|
|
)
|
|
assert res.status_code == 200
|
|
assert res.json()["edges_created"] == 0
|
|
|
|
edges = (await db_session.execute(select(Edge))).scalars().all()
|
|
assert edges == []
|
|
links = (await db_session.execute(select(InventoryDeviceLink))).scalars().all()
|
|
assert len(links) == 1 # preserved for later resolution
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_approve_zigbee_resolves_link_after_second_approval(
|
|
client: AsyncClient, headers, db_session
|
|
):
|
|
"""First approval keeps link (other endpoint pending); second approval
|
|
creates the edge. The link row is retained afterwards so the same pair can
|
|
be re-approved onto another canvas — it's topology, wiped only on reimport."""
|
|
from sqlalchemy import select
|
|
|
|
from app.db.models import Edge, InventoryDevice, InventoryDeviceLink
|
|
|
|
a = InventoryDevice(
|
|
ieee_address="0xA",
|
|
friendly_name="a",
|
|
suggested_type="zigbee_router",
|
|
status="pending",
|
|
discovery_source="zigbee",
|
|
)
|
|
b = InventoryDevice(
|
|
ieee_address="0xB",
|
|
friendly_name="b",
|
|
suggested_type="zigbee_enddevice",
|
|
status="pending",
|
|
discovery_source="zigbee",
|
|
)
|
|
db_session.add_all([a, b])
|
|
db_session.add(
|
|
InventoryDeviceLink(source_ieee="0xA", target_ieee="0xB", discovery_source="zigbee")
|
|
)
|
|
await db_session.commit()
|
|
|
|
body = {"label": "x", "type": "zigbee_router", "ip": None, "status": "unknown", "services": []}
|
|
await client.post(f"/api/v1/scan/pending/{a.id}/approve", json=body, headers=headers)
|
|
res = await client.post(f"/api/v1/scan/pending/{b.id}/approve", json=body, headers=headers)
|
|
assert res.json()["edges_created"] == 1
|
|
|
|
edges = (await db_session.execute(select(Edge))).scalars().all()
|
|
assert len(edges) == 1
|
|
links = (await db_session.execute(select(InventoryDeviceLink))).scalars().all()
|
|
assert len(links) == 1 # retained for re-approval onto other canvases
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_single_approve_zwave_sets_wireless_fields(client, headers, db_session):
|
|
active = await _add_design(db_session, "zwave")
|
|
dev = InventoryDevice(
|
|
id=str(uuid.uuid4()),
|
|
ieee_address="zwave-H-9",
|
|
friendly_name="Door Sensor",
|
|
suggested_type="zwave_enddevice",
|
|
vendor="Aeotec",
|
|
model="ZW120",
|
|
status="pending",
|
|
discovery_source="zwave",
|
|
)
|
|
db_session.add(dev)
|
|
await db_session.commit()
|
|
|
|
res = await client.post(
|
|
f"/api/v1/scan/pending/{dev.id}/approve",
|
|
json={"label": "Door Sensor", "type": "zwave_enddevice", "design_id": active},
|
|
headers=headers,
|
|
)
|
|
assert res.status_code == 200
|
|
|
|
node = (
|
|
await db_session.execute(select(Node).join(InventoryDevice, Node.device_id == InventoryDevice.id).where(InventoryDevice.ieee_address == "zwave-H-9"))
|
|
).scalar_one()
|
|
assert node.design_id == active
|
|
node = (await db_session.execute(select(Node))).scalars().first()
|
|
assert node is not None
|
|
device = await db_session.get(InventoryDevice, node.device_id)
|
|
assert device is not None
|
|
assert device.status_live == "online"
|
|
assert device.check_method == "none"
|
|
assert any(p["key"] == "Z-Wave ID" for p in device.properties)
|