haeaeae/homelable
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
fafbaeffcd05d532a409d0c8bd2d384efc333c67
homelable/backend/tests/test_inventory_sync.py
T
Pouzor 0c7fd3c127 fix(canvas): stop a canvas save from reverting a device edited elsewhere 
A canvas node carries a full copy of its Device Inventory row, hydrated when
the canvas loads. The save routed all of it back, so a save made for nothing
but a moved node rewrote the row from a snapshot that could be hours old —
silently reverting an edit made meanwhile in the inventory modal, on another
canvas, or by the scanner.

Diffing the payload against the row server-side cannot fix this: it can't tell
"I edited this" from "the row moved on since I loaded it". Only the client
holds the baseline.

- canvasStore keeps `factsBaseline` — the device facts as received — set on
  load, refreshed on save, rebased per field by `applyDeviceFacts`.
- `serializeNode` sends `changed_facts`: what this canvas actually edited.
  `link_facts(changed_fields=)` writes nothing outside that list, even where
  the values differ. Absent (older client, YAML import, MCP) keeps the previous
  full-write behaviour.
- `changed_facts()` additionally drops facts already equal to the row, so a
  no-op save writes nothing. Identity matching still uses the full payload.
- Live fields (status, last_seen…) bypass the filter — the status checker owns
  reachability, and a save only ever fills a row never checked.

An inventory edit also lands on the canvases already on screen:
`applyDeviceFacts` pushes the saved row onto every node drawing it, without
marking the canvas unsaved. A fact the canvas has edited but not saved is left
alone — work in progress wins locally and still saves.

ha-relevant: maybe
2026-08-14 18:02:02 +02:00

801 lines
32 KiB
Python
Raw Blame History

"""The node ↔ Device Inventory link: matching, merging, and the 3.3.0 backfill.
The inventory row owns the device facts; a node owns how the device is drawn.
These tests pin the rules that make one device end up as one row even when it
was drawn on several canvases.
"""
import json
import uuid
from datetime import datetime, timedelta, timezone
import pytest
from httpx import AsyncClient
from sqlalchemy import select, text
from app.db.models import Design, InventoryDevice, Node
from app.services.inventory_sync import (
backfill_node_devices,
changed_facts,
find_device_for,
merge_properties,
merge_services,
)
def _now(offset_minutes: int = 0) -> datetime:
return datetime(2026, 1, 1, 12, 0, tzinfo=timezone.utc) + timedelta(minutes=offset_minutes)
async def _design(db_session, name="Net") -> str:
design = Design(id=str(uuid.uuid4()), name=name)
db_session.add(design)
await db_session.commit()
return design.id
def _node(design_id: str, **kwargs) -> Node:
payload = {
"id": str(uuid.uuid4()),
"label": "n",
"type": "server",
"pos_x": 0.0,
"pos_y": 0.0,
"design_id": design_id,
}
payload.update(kwargs)
return Node(**payload)
# --- pure merge rules -------------------------------------------------------
class TestMergeRules:
def test_properties_union_on_key_with_incoming_winning(self):
base = [{"key": "Rack", "value": "A1", "icon": None, "visible": True}]
incoming = [
{"key": "rack", "value": "B2", "icon": None, "visible": False},
{"key": "Owner", "value": "me", "icon": None, "visible": True},
]
out = merge_properties(base, incoming)
assert [p["value"] for p in out] == ["B2", "me"]
# Existing keys keep their position — a user's ordering survives.
assert out[0]["key"] == "Rack"
assert out[0]["visible"] is False
def test_properties_keep_established_value_when_incoming_is_blank(self):
out = merge_properties(
[{"key": "Rack", "value": "A1", "icon": "Server", "visible": True}],
[{"key": "Rack", "value": "", "icon": None, "visible": True}],
)
assert out[0]["value"] == "A1"
assert out[0]["icon"] == "Server"
def test_services_union_on_port_protocol_and_name(self):
base = [{"port": 22, "protocol": "tcp", "service_name": "ssh"}]
incoming = [
{"port": 22, "protocol": "tcp", "service_name": "SSH", "icon": "Terminal"},
{"port": 80, "protocol": "tcp", "service_name": "http"},
]
out = merge_services(base, incoming)
assert len(out) == 2
assert out[0]["icon"] == "Terminal"
assert out[1]["port"] == 80
class TestChangedFacts:
"""What a save is allowed to write: its edit, not its whole snapshot."""
def test_an_unchanged_snapshot_writes_nothing(self):
device = InventoryDevice(
id="d-1",
label="NAS",
type="nas",
hostname="nas.lan",
ip="10.0.0.5",
notes="in the garage",
status_live="online",
properties=[{"key": "Rack", "value": "A1", "icon": None, "visible": True}],
services=[{"port": 22, "protocol": "tcp", "service_name": "ssh"}],
)
facts = {
"label": "NAS",
"type": "nas",
"hostname": "nas.lan",
"ip": "10.0.0.5",
"notes": "in the garage",
"status": "online",
"properties": [{"key": "Rack", "value": "A1", "icon": None, "visible": True}],
"services": [{"port": 22, "protocol": "tcp", "service_name": "ssh"}],
}
assert changed_facts(device, facts) == {}
def test_only_the_edited_field_survives(self):
device = InventoryDevice(id="d-1", hostname="nas.lan", ip="10.0.0.5", notes="old")
facts = {"hostname": "nas.lan", "ip": "10.0.0.5", "notes": "new"}
assert changed_facts(device, facts) == {"notes": "new"}
def test_a_blank_incoming_value_is_not_a_change(self):
# A blank never clears an established value, so it is not an edit either.
device = InventoryDevice(id="d-1", hostname="nas.lan", notes="keep me")
assert changed_facts(device, {"hostname": "", "notes": None}) == {}
def test_a_changed_list_is_sent_whole(self):
device = InventoryDevice(
id="d-1",
properties=[{"key": "Rack", "value": "A1", "icon": None, "visible": True}],
services=[{"port": 22, "protocol": "tcp", "service_name": "ssh"}],
)
facts = {
"properties": [],
"services": [{"port": 22, "protocol": "tcp", "service_name": "ssh"}],
}
# Replace semantics need the full list, and the untouched one stays out.
assert changed_facts(device, facts) == {"properties": []}
def test_live_status_only_fills_a_row_never_checked(self):
unknown = InventoryDevice(id="d-1", status_live="unknown")
assert changed_facts(unknown, {"status": "online"}) == {"status": "online"}
checked = InventoryDevice(id="d-2", status_live="offline")
# The status checker owns reachability; a stale canvas must not reset it.
assert changed_facts(checked, {"status": "online"}) == {}
# --- matching ---------------------------------------------------------------
class TestFindDeviceFor:
@pytest.mark.asyncio
async def test_prefers_ieee_over_ip_and_mac(self, db_session):
by_ip = InventoryDevice(id="d-ip", ip="10.0.0.5")
by_ieee = InventoryDevice(id="d-ieee", ieee_address="0xAAA")
db_session.add_all([by_ip, by_ieee])
await db_session.commit()
found = await find_device_for(db_session, ip="10.0.0.5", mac=None, ieee="0xAAA")
assert found is not None and found.id == "d-ieee"
@pytest.mark.asyncio
async def test_matches_one_address_out_of_a_comma_list(self, db_session):
device = InventoryDevice(id="d-1", ip="10.0.0.5, fd00::1")
db_session.add(device)
await db_session.commit()
found = await find_device_for(db_session, ip="fd00::1", mac=None, ieee=None)
assert found is not None and found.id == "d-1"
@pytest.mark.asyncio
async def test_does_not_match_a_longer_address_with_the_same_prefix(self, db_session):
db_session.add(InventoryDevice(id="d-1", ip="10.0.0.40"))
await db_session.commit()
assert await find_device_for(db_session, ip="10.0.0.4", mac=None, ieee=None) is None
@pytest.mark.asyncio
async def test_matches_a_hidden_row_rather_than_minting_a_second(self, db_session):
db_session.add(InventoryDevice(id="d-hidden", ip="10.0.0.5", status="hidden"))
await db_session.commit()
found = await find_device_for(db_session, ip="10.0.0.5", mac=None, ieee=None)
assert found is not None and found.id == "d-hidden"
# --- the backfill -----------------------------------------------------------
class TestBackfill:
"""The 3.3.0 migration path.
A pre-3.3.0 database still has the device columns on `nodes`; the backfill
reads them with raw SQL (the model no longer declares them) and folds each
node's view into an inventory row. These tests recreate that shape.
"""
async def _legacy_nodes_table(self, db_session) -> None:
"""Re-add the pre-3.3.0 device columns to `nodes`."""
for column, sql_type in (
("hostname", "VARCHAR"), ("ip", "VARCHAR"), ("mac", "VARCHAR"), ("os", "VARCHAR"),
("status", "VARCHAR"), ("check_method", "VARCHAR"), ("check_target", "VARCHAR"),
("services", "JSON"), ("notes", "TEXT"), ("cpu_count", "INTEGER"),
("cpu_model", "VARCHAR"), ("ram_gb", "FLOAT"), ("disk_gb", "FLOAT"),
("show_hardware", "BOOLEAN"), ("properties", "JSON"), ("ieee_address", "VARCHAR"),
("last_seen", "DATETIME"), ("last_scan", "DATETIME"), ("response_time_ms", "INTEGER"),
):
await db_session.execute(text(f"ALTER TABLE nodes ADD COLUMN {column} {sql_type}"))
await db_session.commit()
async def _legacy_node(self, db_session, design_id: str, *, updated_at=None, **facts) -> str:
node_id = str(uuid.uuid4())
columns = {
"id": node_id,
"label": facts.pop("label", "n"),
"type": facts.pop("type", "server"),
"design_id": design_id,
"pos_x": 0.0,
"pos_y": 0.0,
"container_mode": 0,
"show_port_numbers": 0,
"bottom_handles": 1,
"top_handles": 1,
"left_handles": 0,
"right_handles": 0,
"created_at": _now(0),
"updated_at": updated_at or _now(0),
}
for key in ("services", "properties"):
if key in facts:
facts[key] = json.dumps(facts[key])
columns.update(facts)
names = ", ".join(columns)
binds = ", ".join(f":{c}" for c in columns)
await db_session.execute(text(f"INSERT INTO nodes ({names}) VALUES ({binds})"), columns)
await db_session.commit()
return node_id
@pytest.mark.asyncio
async def test_does_nothing_when_the_columns_are_already_gone(self, db_session):
"""A second boot: there is no legacy data left to read."""
design = await _design(db_session)
db_session.add(_node(design))
await db_session.commit()
assert await backfill_node_devices(db_session) == {"linked": 0, "created": 0, "merged": 0}
@pytest.mark.asyncio
async def test_links_a_node_to_its_existing_row(self, db_session):
await self._legacy_nodes_table(db_session)
design = await _design(db_session)
db_session.add(InventoryDevice(id="d-1", ip="10.0.0.5", hostname="nas"))
await db_session.commit()
node_id = await self._legacy_node(db_session, design, ip="10.0.0.5", label="NAS")
stats = await backfill_node_devices(db_session)
await db_session.commit()
assert stats == {"linked": 1, "created": 0, "merged": 1}
node = await db_session.get(Node, node_id)
assert node is not None and node.device_id == "d-1"
@pytest.mark.asyncio
async def test_creates_a_row_for_a_node_no_scan_ever_saw(self, db_session):
await self._legacy_nodes_table(db_session)
design = await _design(db_session)
node_id = await self._legacy_node(
db_session, design, label="Dumb switch", type="switch", notes="under the desk"
)
stats = await backfill_node_devices(db_session)
await db_session.commit()
assert stats == {"linked": 1, "created": 1, "merged": 0}
node = await db_session.get(Node, node_id)
assert node is not None
device = await db_session.get(InventoryDevice, node.device_id)
assert device is not None
assert device.label == "Dumb switch"
assert device.type == "switch"
assert device.notes == "under the desk"
assert device.status == "approved"
assert device.discovery_sources == ["canvas"]
@pytest.mark.asyncio
async def test_two_canvases_one_device_merge_into_one_row(self, db_session):
"""The whole point: the same host drawn twice becomes one inventory row."""
await self._legacy_nodes_table(db_session)
design_a = await _design(db_session, "A")
design_b = await _design(db_session, "B")
older = await self._legacy_node(
db_session, design_a,
ip="10.0.0.5", label="nas-old", hostname="nas.lan", notes="older note",
properties=[{"key": "Rack", "value": "A1", "icon": None, "visible": True}],
services=[{"port": 22, "protocol": "tcp", "service_name": "ssh"}],
updated_at=_now(0),
)
newer = await self._legacy_node(
db_session, design_b,
ip="10.0.0.5", label="nas-new", os="TrueNAS",
properties=[{"key": "Owner", "value": "me", "icon": None, "visible": True}],
services=[{"port": 80, "protocol": "tcp", "service_name": "http"}],
updated_at=_now(30),
)
stats = await backfill_node_devices(db_session)
await db_session.commit()
assert stats["linked"] == 2
node_a = await db_session.get(Node, older)
node_b = await db_session.get(Node, newer)
assert node_a is not None and node_b is not None
assert node_a.device_id == node_b.device_id
device = await db_session.get(InventoryDevice, node_a.device_id)
assert device is not None
# Scalars: most recent edit wins, blanks never wipe an established value.
assert device.label == "nas-new"
assert device.hostname == "nas.lan"
assert device.os == "TrueNAS"
assert device.notes == "older note"
# Cumulative fields: union of both canvases.
assert {p["key"] for p in device.properties} == {"Rack", "Owner"}
assert {s["port"] for s in device.services} == {22, 80}
@pytest.mark.asyncio
async def test_leaves_canvas_furniture_alone(self, db_session):
await self._legacy_nodes_table(db_session)
design = await _design(db_session)
for kind in ("group", "groupRect", "text"):
await self._legacy_node(db_session, design, type=kind, label=kind)
stats = await backfill_node_devices(db_session)
await db_session.commit()
assert stats["linked"] == 0
assert (await db_session.execute(select(InventoryDevice))).scalars().all() == []
@pytest.mark.asyncio
async def test_is_a_no_op_on_a_second_run(self, db_session):
await self._legacy_nodes_table(db_session)
design = await _design(db_session)
await self._legacy_node(db_session, design, ip="10.0.0.5")
first = await backfill_node_devices(db_session)
await db_session.commit()
second = await backfill_node_devices(db_session)
await db_session.commit()
assert first["linked"] == 1
assert second == {"linked": 0, "created": 0, "merged": 0}
rows = (await db_session.execute(select(InventoryDevice))).scalars().all()
assert len(rows) == 1
# --- routes -----------------------------------------------------------------
class TestRoutesKeepTheLinkInStep:
@pytest.mark.asyncio
async def test_creating_a_node_joins_the_matching_row(self, client: AsyncClient, headers, db_session):
db_session.add(InventoryDevice(id="d-1", ip="10.0.0.5", hostname="nas", os="TrueNAS"))
await db_session.commit()
res = await client.post(
"/api/v1/nodes",
json={"type": "nas", "label": "NAS", "ip": "10.0.0.5", "force": True},
headers=headers,
)
assert res.status_code == 201
body = res.json()
assert body["device_id"] == "d-1"
# Hydrated from the row: the node never carried an OS.
assert body["os"] == "TrueNAS"
@pytest.mark.asyncio
async def test_creating_furniture_creates_no_row(self, client: AsyncClient, headers, db_session):
res = await client.post(
"/api/v1/nodes", json={"type": "groupRect", "label": "Zone"}, headers=headers
)
assert res.status_code == 201
assert res.json()["device_id"] is None
rows = (await db_session.execute(select(InventoryDevice))).scalars().all()
assert rows == []
@pytest.mark.asyncio
async def test_editing_a_node_writes_through_to_the_row(self, client: AsyncClient, headers, db_session):
db_session.add(InventoryDevice(id="d-1", ip="10.0.0.5", hostname="nas"))
await db_session.commit()
node_id = (
await client.post(
"/api/v1/nodes",
json={"type": "nas", "label": "NAS", "ip": "10.0.0.5", "force": True},
headers=headers,
)
).json()["id"]
res = await client.patch(
f"/api/v1/nodes/{node_id}",
json={"hostname": "nas2.lan", "notes": "moved to the garage"},
headers=headers,
)
assert res.status_code == 200
device = await db_session.get(InventoryDevice, "d-1")
await db_session.refresh(device)
assert device.hostname == "nas2.lan"
assert device.notes == "moved to the garage"
@pytest.mark.asyncio
async def test_patching_one_list_leaves_the_other_alone(self, client: AsyncClient, headers, db_session):
"""Replace applies list by list — an unsent list is not blanked."""
db_session.add(
InventoryDevice(
id="d-1",
ip="10.0.0.5",
services=[{"port": 22, "protocol": "tcp", "service_name": "ssh"}],
properties=[{"key": "Rack", "value": "A1", "icon": None, "visible": True}],
)
)
await db_session.commit()
node_id = (
await client.post(
"/api/v1/nodes",
json={"type": "nas", "label": "NAS", "ip": "10.0.0.5", "force": True},
headers=headers,
)
).json()["id"]
# properties only: services must survive untouched.
res = await client.patch(
f"/api/v1/nodes/{node_id}",
json={"properties": [{"key": "Owner", "value": "me", "icon": None, "visible": True}]},
headers=headers,
)
assert res.status_code == 200
device = await db_session.get(InventoryDevice, "d-1")
await db_session.refresh(device)
assert [p["key"] for p in device.properties] == ["Owner"]
assert [s["port"] for s in device.services] == [22]
# services only: the properties just written must survive in turn.
res = await client.patch(
f"/api/v1/nodes/{node_id}",
json={"services": [{"port": 443, "protocol": "tcp", "service_name": "https"}]},
headers=headers,
)
assert res.status_code == 200
await db_session.refresh(device)
assert [s["port"] for s in device.services] == [443]
assert [p["key"] for p in device.properties] == ["Owner"]
@pytest.mark.asyncio
async def test_patching_a_list_empty_still_clears_it(self, client: AsyncClient, headers, db_session):
"""An explicit empty list is a deletion, not an omission."""
db_session.add(
InventoryDevice(
id="d-1",
ip="10.0.0.5",
properties=[{"key": "Rack", "value": "A1", "icon": None, "visible": True}],
)
)
await db_session.commit()
node_id = (
await client.post(
"/api/v1/nodes",
json={"type": "nas", "label": "NAS", "ip": "10.0.0.5", "force": True},
headers=headers,
)
).json()["id"]
res = await client.patch(f"/api/v1/nodes/{node_id}", json={"properties": []}, headers=headers)
assert res.status_code == 200
device = await db_session.get(InventoryDevice, "d-1")
await db_session.refresh(device)
assert device.properties == []
@pytest.mark.asyncio
async def test_a_canvas_save_does_not_revert_an_inventory_edit(
self, client: AsyncClient, headers, db_session
):
"""Regression: a save carries this canvas' edit, not its stale snapshot.
The canvas payload holds a full copy of the device, hydrated when it
loaded. Editing the device elsewhere and then saving the still-open
canvas — for nothing but a moved node — must not roll the row back.
"""
db_session.add(
InventoryDevice(
id="d-1",
ip="10.0.0.5",
label="NAS",
type="nas",
notes="old note",
status="approved",
properties=[{"key": "Rack", "value": "A1", "icon": None, "visible": True}],
)
)
await db_session.commit()
design_id = (await client.post("/api/v1/designs", json={"name": "A"}, headers=headers)).json()["id"]
# What the canvas loaded and still holds in memory.
stale = {
"id": str(uuid.uuid4()),
"type": "nas",
"label": "NAS",
"status": "unknown",
"pos_x": 0,
"pos_y": 0,
"ip": "10.0.0.5",
"notes": "old note",
"properties": [{"key": "Rack", "value": "A1", "icon": None, "visible": True}],
}
res = await client.post(
"/api/v1/canvas/save",
json={"design_id": design_id, "nodes": [stale], "edges": [], "viewport": {}},
headers=headers,
)
assert res.status_code == 200
# Meanwhile, in the Device Inventory modal.
res = await client.patch(
"/api/v1/scan/pending/d-1",
json={
"notes": "moved to the loft",
"properties": [
{"key": "Rack", "value": "A1", "icon": None, "visible": True},
{"key": "Owner", "value": "me", "icon": None, "visible": True},
],
},
headers=headers,
)
assert res.status_code == 200
# The canvas saves again — only the node position changed, so it claims
# no device edit at all.
res = await client.post(
"/api/v1/canvas/save",
json={
"design_id": design_id,
"nodes": [{**stale, "pos_x": 240, "changed_facts": []}],
"edges": [],
"viewport": {},
},
headers=headers,
)
assert res.status_code == 200
device = await db_session.get(InventoryDevice, "d-1")
await db_session.refresh(device)
assert device.notes == "moved to the loft"
assert [p["key"] for p in device.properties] == ["Rack", "Owner"]
@pytest.mark.asyncio
async def test_a_canvas_save_still_writes_what_that_canvas_changed(
self, client: AsyncClient, headers, db_session
):
"""The narrowing must not cost the write-through: an edit still lands."""
db_session.add(InventoryDevice(id="d-1", ip="10.0.0.5", label="NAS", type="nas", notes="old note"))
await db_session.commit()
design_id = (await client.post("/api/v1/designs", json={"name": "A"}, headers=headers)).json()["id"]
node = {
"id": str(uuid.uuid4()),
"type": "nas",
"label": "NAS",
"status": "unknown",
"pos_x": 0,
"pos_y": 0,
"ip": "10.0.0.5",
"notes": "old note",
}
await client.post(
"/api/v1/canvas/save",
json={"design_id": design_id, "nodes": [node], "edges": [], "viewport": {}},
headers=headers,
)
res = await client.post(
"/api/v1/canvas/save",
json={
"design_id": design_id,
"nodes": [
{
**node,
"notes": "edited on the canvas",
"hostname": "nas.lan",
"changed_facts": ["notes", "hostname"],
}
],
"edges": [],
"viewport": {},
},
headers=headers,
)
assert res.status_code == 200
device = await db_session.get(InventoryDevice, "d-1")
await db_session.refresh(device)
assert device.notes == "edited on the canvas"
assert device.hostname == "nas.lan"
@pytest.mark.asyncio
async def test_an_edit_writes_without_dragging_the_rest_of_the_snapshot(
self, client: AsyncClient, headers, db_session
):
"""One edited fact lands; the stale fields beside it stay out of the write."""
db_session.add(
InventoryDevice(id="d-1", ip="10.0.0.5", label="NAS", type="nas", notes="old note")
)
await db_session.commit()
design_id = (await client.post("/api/v1/designs", json={"name": "A"}, headers=headers)).json()["id"]
node = {
"id": str(uuid.uuid4()),
"type": "nas",
"label": "NAS",
"status": "unknown",
"pos_x": 0,
"pos_y": 0,
"ip": "10.0.0.5",
"notes": "old note",
}
await client.post(
"/api/v1/canvas/save",
json={"design_id": design_id, "nodes": [node], "edges": [], "viewport": {}},
headers=headers,
)
await client.patch("/api/v1/scan/pending/d-1", json={"notes": "moved to the loft"}, headers=headers)
# The canvas renames the node; its `notes` copy is stale but unedited.
await client.post(
"/api/v1/canvas/save",
json={
"design_id": design_id,
"nodes": [{**node, "label": "Big NAS", "changed_facts": ["label"]}],
"edges": [],
"viewport": {},
},
headers=headers,
)
device = await db_session.get(InventoryDevice, "d-1")
await db_session.refresh(device)
assert device.label == "Big NAS"
assert device.notes == "moved to the loft"
@pytest.mark.asyncio
async def test_a_client_that_tracks_no_changes_still_writes_its_facts(
self, client: AsyncClient, headers, db_session
):
"""Backward compatible: an import or older client omits the field entirely."""
db_session.add(InventoryDevice(id="d-1", ip="10.0.0.5", notes="old note"))
await db_session.commit()
design_id = (await client.post("/api/v1/designs", json={"name": "A"}, headers=headers)).json()["id"]
await client.post(
"/api/v1/canvas/save",
json={
"design_id": design_id,
"nodes": [
{
"id": str(uuid.uuid4()),
"type": "nas",
"label": "NAS",
"status": "unknown",
"pos_x": 0,
"pos_y": 0,
"ip": "10.0.0.5",
"notes": "from the import",
}
],
"edges": [],
"viewport": {},
},
headers=headers,
)
device = await db_session.get(InventoryDevice, "d-1")
await db_session.refresh(device)
assert device.notes == "from the import"
@pytest.mark.asyncio
async def test_the_second_canvas_sees_the_first_canvas_edit(self, client: AsyncClient, headers, db_session):
"""No per-node overrides: one device, one set of facts, every canvas."""
db_session.add(InventoryDevice(id="d-1", ip="10.0.0.5"))
await db_session.commit()
design_a = (await client.post("/api/v1/designs", json={"name": "A"}, headers=headers)).json()["id"]
design_b = (await client.post("/api/v1/designs", json={"name": "B"}, headers=headers)).json()["id"]
node_a = (
await client.post(
"/api/v1/nodes",
json={"type": "nas", "label": "NAS", "ip": "10.0.0.5", "design_id": design_a, "force": True},
headers=headers,
)
).json()["id"]
await client.post(
"/api/v1/nodes",
json={"type": "nas", "label": "NAS", "ip": "10.0.0.5", "design_id": design_b, "force": True},
headers=headers,
)
await client.patch(f"/api/v1/nodes/{node_a}", json={"ip": "10.0.0.9"}, headers=headers)
canvas_b = (await client.get(f"/api/v1/canvas?design_id={design_b}", headers=headers)).json()
assert canvas_b["nodes"][0]["ip"] == "10.0.0.9"
@pytest.mark.asyncio
async def test_the_wire_shape_the_mcp_server_reads_is_unchanged(
self, client: AsyncClient, headers, db_session
):
"""`mcp/app/tools.py` reads the facts flat on the node — keep it that way.
The MCP server is a thin proxy over these routes: `_slim_canvas` keeps
`NODE_KEEP`, `_slim_node_summary` keeps label/type/status, and `get_node`
looks a node up by label. None of it knows the inventory row exists, so
the split must not show on the wire.
"""
db_session.add(
InventoryDevice(
id="d-1",
ip="10.0.0.5",
hostname="nas.lan",
mac="aa:bb:cc:dd:ee:ff",
os="TrueNAS",
notes="in the garage",
cpu_count=8,
cpu_model="Xeon",
ram_gb=32.0,
disk_gb=4000.0,
services=[{"port": 22, "protocol": "tcp", "service_name": "ssh"}],
properties=[{"key": "Rack", "value": "A1", "icon": None, "visible": True}],
status_live="online",
)
)
await db_session.commit()
design_id = (await client.post("/api/v1/designs", json={"name": "Net"}, headers=headers)).json()["id"]
created = (
await client.post(
"/api/v1/nodes",
json={
"type": "nas", "label": "NAS", "ip": "10.0.0.5",
"design_id": design_id, "force": True,
},
headers=headers,
)
).json()
# mcp `_slim_canvas` NODE_KEEP, plus what `_slim_node_summary` reads.
expected = {
"type": "nas",
"label": "NAS",
"ip": "10.0.0.5",
"hostname": "nas.lan",
"mac": "aa:bb:cc:dd:ee:ff",
"os": "TrueNAS",
"status": "online",
"notes": "in the garage",
"cpu_count": 8,
"cpu_model": "Xeon",
"ram_gb": 32.0,
"disk_gb": 4000.0,
}
for source in (
created,
(await client.get(f"/api/v1/nodes/{created['id']}", headers=headers)).json(),
(await client.get("/api/v1/nodes", headers=headers)).json()[0],
(await client.get("/api/v1/canvas", headers=headers)).json()["nodes"][0],
# mcp `get_node` falls back to a label lookup when given no id.
(await client.get("/api/v1/nodes?label=NAS", headers=headers)).json()[0],
):
assert {k: source[k] for k in expected} == expected
assert source["services"] == [{"port": 22, "protocol": "tcp", "service_name": "ssh"}]
assert source["properties"] == [
{"key": "Rack", "value": "A1", "icon": None, "visible": True}
]
assert "parent_id" in source
@pytest.mark.asyncio
async def test_deleting_a_node_keeps_the_device(self, client: AsyncClient, headers, db_session):
db_session.add(InventoryDevice(id="d-1", ip="10.0.0.5"))
await db_session.commit()
node_id = (
await client.post(
"/api/v1/nodes",
json={"type": "nas", "label": "NAS", "ip": "10.0.0.5", "force": True},
headers=headers,
)
).json()["id"]
assert (await client.delete(f"/api/v1/nodes/{node_id}", headers=headers)).status_code == 204
assert await db_session.get(InventoryDevice, "d-1") is not None
@pytest.mark.asyncio
async def test_approving_a_device_links_instead_of_copying(self, client: AsyncClient, headers, db_session):
db_session.add(
InventoryDevice(id="d-1", ip="10.0.0.5", hostname="nas", suggested_type="nas", status="pending")
)
await db_session.commit()
res = await client.post(
"/api/v1/scan/pending/d-1/approve",
json={"type": "nas", "label": "NAS", "ip": "10.0.0.5", "status": "unknown"},
headers=headers,
)
assert res.status_code == 200
node = await db_session.get(Node, res.json()["node_id"])
assert node is not None and node.device_id == "d-1"
Reference in New Issue View Git Blame Copy Permalink