feat(zigbee): populate IEEE/Vendor/Model/LQI properties on approve and re-import
- Approval (single + bulk) of zigbee pending devices now writes IEEE, Vendor, Model, LQI into Node.properties so they show in the right panel. - Zigbee re-import refreshes properties on existing canvas Nodes and skips creating a pending row when the device was already approved — keeps approved devices out of pending/hidden modals. - Coordinator Node also receives the same properties on first creation and on re-import. - Remove the Parent Container selector from the add/edit node modal.
This commit is contained in:
@@ -528,6 +528,87 @@ async def test_bulk_approve_approves_devices(client: AsyncClient, headers, two_p
|
||||
assert pending_res.json() == []
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
async def zigbee_pending_device(db_session):
|
||||
device = PendingDevice(
|
||||
id=str(uuid.uuid4()),
|
||||
ip=None,
|
||||
mac=None,
|
||||
hostname=None,
|
||||
friendly_name="bulb_1",
|
||||
services=[],
|
||||
suggested_type="zigbee_enddevice",
|
||||
device_subtype="EndDevice",
|
||||
ieee_address="0xABCDEF",
|
||||
vendor="IKEA",
|
||||
model="TRADFRI",
|
||||
lqi=180,
|
||||
status="pending",
|
||||
discovery_source="zigbee",
|
||||
)
|
||||
db_session.add(device)
|
||||
await db_session.commit()
|
||||
await db_session.refresh(device)
|
||||
return device
|
||||
|
||||
|
||||
@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.ieee_address == "0xABCDEF"))
|
||||
).scalar_one()
|
||||
keys = {p["key"]: p["value"] for p in node.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.ieee_address == "0xABCDEF"))
|
||||
).scalar_one()
|
||||
keys = {p["key"]: p["value"] for p in node.properties}
|
||||
assert keys["IEEE"] == "0xABCDEF"
|
||||
assert keys["Vendor"] == "IKEA"
|
||||
assert keys["Model"] == "TRADFRI"
|
||||
assert keys["LQI"] == "180"
|
||||
assert node.check_method == "none"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_bulk_approve_sets_default_check_method(client: AsyncClient, headers, two_pending_devices, db_session):
|
||||
"""Approved devices with an IP must default to ping; otherwise scheduler skips them."""
|
||||
|
||||
@@ -388,6 +388,97 @@ async def test_persist_pending_import_replaces_links(db_session) -> None:
|
||||
assert (rows[0].source_ieee, rows[0].target_ieee) == ("0xCOORD", "0xR1")
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_persist_pending_import_sets_coordinator_properties(db_session) -> None:
|
||||
"""Coordinator Node is created with IEEE/Vendor/Model/LQI in properties."""
|
||||
from sqlalchemy import select
|
||||
|
||||
from app.api.routes.zigbee import _persist_pending_import
|
||||
from app.db.models import Node
|
||||
|
||||
nodes_with_meta = [dict(n) for n in _PENDING_NODES]
|
||||
nodes_with_meta[0]["vendor"] = "TI"
|
||||
nodes_with_meta[0]["model"] = "CC2652"
|
||||
|
||||
await _persist_pending_import(db_session, nodes_with_meta, _PENDING_EDGES)
|
||||
|
||||
coord = (
|
||||
await db_session.execute(select(Node).where(Node.ieee_address == "0xCOORD"))
|
||||
).scalar_one()
|
||||
keys = {p["key"]: p["value"] for p in coord.properties}
|
||||
assert keys == {"IEEE": "0xCOORD", "Vendor": "TI", "Model": "CC2652"}
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_persist_pending_import_skips_pending_for_approved_node(
|
||||
db_session,
|
||||
) -> None:
|
||||
"""A device already approved as a canvas Node must not reappear in pending.
|
||||
|
||||
Its properties must still be refreshed with the latest Vendor/Model/LQI.
|
||||
"""
|
||||
from sqlalchemy import select
|
||||
|
||||
from app.api.routes.zigbee import _persist_pending_import
|
||||
from app.db.models import Node, PendingDevice
|
||||
|
||||
# Simulate: router was approved earlier → exists as a canvas Node.
|
||||
approved = Node(
|
||||
label="router_1",
|
||||
type="zigbee_router",
|
||||
status="online",
|
||||
check_method="none",
|
||||
ieee_address="0xR1",
|
||||
services=[],
|
||||
properties=[],
|
||||
)
|
||||
db_session.add(approved)
|
||||
await db_session.commit()
|
||||
|
||||
bumped = [dict(n) for n in _PENDING_NODES]
|
||||
bumped[1]["lqi"] = 250 # new LQI from re-import
|
||||
await _persist_pending_import(db_session, bumped, _PENDING_EDGES)
|
||||
|
||||
# No PendingDevice row was created for the approved router.
|
||||
pendings = (
|
||||
await db_session.execute(
|
||||
select(PendingDevice).where(PendingDevice.ieee_address == "0xR1")
|
||||
)
|
||||
).scalars().all()
|
||||
assert pendings == []
|
||||
|
||||
# Node properties got refreshed.
|
||||
refreshed = (
|
||||
await db_session.execute(select(Node).where(Node.ieee_address == "0xR1"))
|
||||
).scalar_one()
|
||||
keys = {p["key"]: p["value"] for p in refreshed.properties}
|
||||
assert keys == {"IEEE": "0xR1", "Vendor": "TI", "Model": "CC2530", "LQI": "250"}
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_persist_pending_import_refreshes_existing_coordinator_properties(
|
||||
db_session,
|
||||
) -> None:
|
||||
from sqlalchemy import select
|
||||
|
||||
from app.api.routes.zigbee import _persist_pending_import
|
||||
from app.db.models import Node
|
||||
|
||||
await _persist_pending_import(db_session, _PENDING_NODES, _PENDING_EDGES)
|
||||
|
||||
bumped = [dict(n) for n in _PENDING_NODES]
|
||||
bumped[0]["vendor"] = "TI"
|
||||
bumped[0]["model"] = "CC2652"
|
||||
await _persist_pending_import(db_session, bumped, _PENDING_EDGES)
|
||||
|
||||
coord = (
|
||||
await db_session.execute(select(Node).where(Node.ieee_address == "0xCOORD"))
|
||||
).scalar_one()
|
||||
keys = {p["key"]: p["value"] for p in coord.properties}
|
||||
assert keys["Vendor"] == "TI"
|
||||
assert keys["Model"] == "CC2652"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_import_pending_requires_auth(client: AsyncClient) -> None:
|
||||
res = await client.post(
|
||||
|
||||
Reference in New Issue
Block a user