feat: surface inventory timestamps on nodes
Add creation_date, last_scan and last_modify to the node inventory and the right-hand detail panel (last_seen already shown): - creation_date -> existing created_at column - last_modify -> existing updated_at column (bumped on any node change) - last_scan -> new column, stamped when a scan observes a node by IP/MAC - last_seen -> unchanged Backend: new nodes.last_scan column + idempotent migration, NodeResponse field, scanner stamps matching canvas nodes per scanned host. Frontend: NodeData fields + DetailPanel rows with UTC-safe timestamp formatting. ha-relevant: maybe
This commit is contained in:
@@ -51,6 +51,18 @@ async def test_save_canvas_creates_nodes_and_edges(client: AsyncClient, headers:
|
||||
assert canvas["viewport"] == {"x": 1, "y": 2, "zoom": 1.5}
|
||||
|
||||
|
||||
async def test_load_canvas_exposes_inventory_timestamps(client: AsyncClient, headers: dict):
|
||||
n1 = node_payload(label="Router", type="router")
|
||||
await client.post("/api/v1/canvas/save", json={"nodes": [n1], "edges": [], "viewport": {}}, headers=headers)
|
||||
|
||||
node = (await client.get("/api/v1/canvas", headers=headers)).json()["nodes"][0]
|
||||
# created_at / updated_at always set; last_seen / last_scan null until observed.
|
||||
assert node["created_at"] is not None
|
||||
assert node["updated_at"] is not None
|
||||
assert "last_seen" in node
|
||||
assert node["last_scan"] is None
|
||||
|
||||
|
||||
async def test_save_canvas_updates_existing_node(client: AsyncClient, headers: dict):
|
||||
n1 = node_payload(label="Old Label")
|
||||
await client.post("/api/v1/canvas/save", json={"nodes": [n1], "edges": [], "viewport": {}}, headers=headers)
|
||||
|
||||
@@ -105,6 +105,28 @@ async def test_legacy_canvas_migrates_into_default_design(legacy_engine):
|
||||
await engine.dispose()
|
||||
|
||||
|
||||
async def test_legacy_nodes_gain_last_scan_column(legacy_engine):
|
||||
"""A legacy nodes table (no last_scan) gains the column after init_db."""
|
||||
db_path, engine = legacy_engine
|
||||
await _build_legacy_schema(engine)
|
||||
|
||||
await database.init_db()
|
||||
|
||||
check = create_async_engine(f"sqlite+aiosqlite:///{db_path}")
|
||||
try:
|
||||
async with check.begin() as conn:
|
||||
cols = (await conn.exec_driver_sql("PRAGMA table_info(nodes)")).fetchall()
|
||||
assert "last_scan" in {c[1] for c in cols}
|
||||
# Existing rows backfill to NULL (never scanned yet).
|
||||
last_scan = (await conn.exec_driver_sql(
|
||||
"SELECT last_scan FROM nodes WHERE id='n1'"
|
||||
)).fetchone()
|
||||
assert last_scan[0] is None
|
||||
finally:
|
||||
await check.dispose()
|
||||
await engine.dispose()
|
||||
|
||||
|
||||
async def test_migration_is_idempotent(legacy_engine):
|
||||
"""Running init_db twice must not duplicate the design or drop any data."""
|
||||
db_path, engine = legacy_engine
|
||||
|
||||
@@ -404,6 +404,84 @@ async def test_run_scan_adds_nmap_devices_as_pending(mem_db):
|
||||
assert any(d.ip == "192.168.1.5" for d in devices)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_run_scan_stamps_last_scan_on_matching_node_by_ip(mem_db):
|
||||
"""A scan that sees a device matching a canvas node (by IP) stamps last_scan."""
|
||||
from app.services.scanner import run_scan
|
||||
|
||||
run_id = _make_run_id()
|
||||
async with mem_db() as session:
|
||||
session.add(_make_scan_run(run_id))
|
||||
session.add(Node(id="n1", type="server", label="NAS", ip="192.168.1.5"))
|
||||
await session.commit()
|
||||
|
||||
nmap_hosts = [{"ip": "192.168.1.5", "hostname": "nas.lan", "mac": None, "os": None, "open_ports": []}]
|
||||
|
||||
async with mem_db() as session:
|
||||
with patch("app.services.scanner._nmap_scan", return_value=nmap_hosts), \
|
||||
patch("app.services.scanner._mdns_discover", new_callable=AsyncMock, return_value=[]), \
|
||||
patch("app.api.routes.status.broadcast_scan_update", new_callable=AsyncMock):
|
||||
await run_scan(["192.168.1.0/24"], session, run_id)
|
||||
|
||||
async with mem_db() as session:
|
||||
node = await session.get(Node, "n1")
|
||||
|
||||
assert node is not None
|
||||
assert node.last_scan is not None
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_run_scan_stamps_last_scan_on_matching_node_by_mac(mem_db):
|
||||
"""A node with no IP but a matching MAC still gets last_scan stamped."""
|
||||
from app.services.scanner import run_scan
|
||||
|
||||
run_id = _make_run_id()
|
||||
async with mem_db() as session:
|
||||
session.add(_make_scan_run(run_id))
|
||||
session.add(Node(id="n2", type="iot", label="Sensor", mac="AA:BB:CC:DD:EE:FF"))
|
||||
await session.commit()
|
||||
|
||||
nmap_hosts = [{"ip": "192.168.1.9", "hostname": None, "mac": "AA:BB:CC:DD:EE:FF", "os": None, "open_ports": []}]
|
||||
|
||||
async with mem_db() as session:
|
||||
with patch("app.services.scanner._nmap_scan", return_value=nmap_hosts), \
|
||||
patch("app.services.scanner._mdns_discover", new_callable=AsyncMock, return_value=[]), \
|
||||
patch("app.api.routes.status.broadcast_scan_update", new_callable=AsyncMock):
|
||||
await run_scan(["192.168.1.0/24"], session, run_id)
|
||||
|
||||
async with mem_db() as session:
|
||||
node = await session.get(Node, "n2")
|
||||
|
||||
assert node is not None
|
||||
assert node.last_scan is not None
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_run_scan_leaves_last_scan_untouched_on_unmatched_node(mem_db):
|
||||
"""A node whose IP/MAC is not seen by the scan keeps last_scan = None."""
|
||||
from app.services.scanner import run_scan
|
||||
|
||||
run_id = _make_run_id()
|
||||
async with mem_db() as session:
|
||||
session.add(_make_scan_run(run_id))
|
||||
session.add(Node(id="n3", type="server", label="Other", ip="10.0.0.99"))
|
||||
await session.commit()
|
||||
|
||||
nmap_hosts = [{"ip": "192.168.1.5", "hostname": None, "mac": None, "os": None, "open_ports": []}]
|
||||
|
||||
async with mem_db() as session:
|
||||
with patch("app.services.scanner._nmap_scan", return_value=nmap_hosts), \
|
||||
patch("app.services.scanner._mdns_discover", new_callable=AsyncMock, return_value=[]), \
|
||||
patch("app.api.routes.status.broadcast_scan_update", new_callable=AsyncMock):
|
||||
await run_scan(["192.168.1.0/24"], session, run_id)
|
||||
|
||||
async with mem_db() as session:
|
||||
node = await session.get(Node, "n3")
|
||||
|
||||
assert node is not None
|
||||
assert node.last_scan is None
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_run_scan_mdns_only_device_added(mem_db):
|
||||
"""Devices found only by mDNS (not nmap) should appear in pending_devices."""
|
||||
|
||||
Reference in New Issue
Block a user