Pouzor
5b5eabf5db
Approving the same zigbee/zwave/proxmox devices onto a second design placed the nodes but drew no edges. _resolve_pending_links_for_ieee deleted each pending_device_link after materializing its edge, so the first approve consumed the whole topology and later approves had nothing to resolve. Links are topology, not one-shot: every importer wipes+reinserts its link set on each import, so they can safely persist across approvals. - keep the link rows (drop both db.delete(link) calls) - scope resolution to the target design (Node.ieee_address + design_id) so a re-approve links that canvas's nodes, not another's - thread design_id through all three approve call sites Existing links deleted by the old code do not come back on their own; a re-import repopulates them. ha-relevant: maybe
1598 lines
55 KiB
Python
1598 lines
55 KiB
Python
"""Tests for scan routes: trigger, pending devices, approve/hide/ignore, stop."""
|
|
import uuid
|
|
from datetime import datetime, timezone
|
|
from unittest.mock import AsyncMock, patch
|
|
|
|
import pytest
|
|
from httpx import AsyncClient
|
|
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.db.models import Design, Node, PendingDevice, ScanRun
|
|
from app.services.scanner import _cancelled_runs, request_cancel, run_scan
|
|
|
|
|
|
@pytest.fixture
|
|
async def headers(client: AsyncClient):
|
|
res = await client.post("/api/v1/auth/login", json={"username": "admin", "password": "admin"})
|
|
token = res.json()["access_token"]
|
|
return {"Authorization": f"Bearer {token}"}
|
|
|
|
|
|
@pytest.fixture
|
|
async def pending_device(db_session):
|
|
import uuid
|
|
device = PendingDevice(
|
|
id=str(uuid.uuid4()),
|
|
ip="192.168.1.100",
|
|
mac="aa:bb:cc:dd:ee:ff",
|
|
hostname="my-server",
|
|
os="Linux",
|
|
services=[{"port": 22, "name": "ssh"}],
|
|
suggested_type="server",
|
|
status="pending",
|
|
)
|
|
db_session.add(device)
|
|
await db_session.commit()
|
|
await db_session.refresh(device)
|
|
return device
|
|
|
|
|
|
# --- _background_scan error handling ---
|
|
|
|
@pytest.fixture
|
|
async def mem_db():
|
|
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
|
|
|
|
from app.db.database import Base
|
|
engine = create_async_engine("sqlite+aiosqlite:///:memory:")
|
|
async with engine.begin() as conn:
|
|
await conn.run_sync(Base.metadata.create_all)
|
|
factory = async_sessionmaker(engine, expire_on_commit=False)
|
|
yield factory
|
|
await engine.dispose()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_background_scan_marks_run_failed_on_exception(mem_db):
|
|
"""If run_scan() raises, the ScanRun must transition running → failed and the
|
|
session rollback path must execute without a follow-on exception."""
|
|
from app.api.routes.scan import _background_scan
|
|
|
|
async with mem_db() as session:
|
|
run = ScanRun(status="running", ranges=["10.0.0.0/24"])
|
|
session.add(run)
|
|
await session.commit()
|
|
run_id = run.id
|
|
|
|
with (
|
|
patch("app.api.routes.scan.AsyncSessionLocal", mem_db),
|
|
patch(
|
|
"app.api.routes.scan.run_scan",
|
|
new_callable=AsyncMock,
|
|
side_effect=RuntimeError("boom"),
|
|
),
|
|
):
|
|
await _background_scan(run_id, ["10.0.0.0/24"])
|
|
|
|
async with mem_db() as session:
|
|
refreshed = await session.get(ScanRun, run_id)
|
|
assert refreshed is not None
|
|
assert refreshed.status == "failed"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_background_scan_leaves_non_running_status_alone(mem_db):
|
|
"""If the run was already stopped/cancelled before run_scan failed, _background_scan
|
|
must NOT overwrite that terminal status with 'failed'."""
|
|
from app.api.routes.scan import _background_scan
|
|
|
|
async with mem_db() as session:
|
|
run = ScanRun(status="cancelled", ranges=["10.0.0.0/24"])
|
|
session.add(run)
|
|
await session.commit()
|
|
run_id = run.id
|
|
|
|
with (
|
|
patch("app.api.routes.scan.AsyncSessionLocal", mem_db),
|
|
patch(
|
|
"app.api.routes.scan.run_scan",
|
|
new_callable=AsyncMock,
|
|
side_effect=RuntimeError("boom"),
|
|
),
|
|
):
|
|
await _background_scan(run_id, ["10.0.0.0/24"])
|
|
|
|
async with mem_db() as session:
|
|
refreshed = await session.get(ScanRun, run_id)
|
|
assert refreshed is not None
|
|
assert refreshed.status == "cancelled"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_background_scan_success_path_invokes_run_scan(mem_db):
|
|
from app.api.routes.scan import _background_scan
|
|
|
|
async with mem_db() as session:
|
|
run = ScanRun(status="running", ranges=["10.0.0.0/24"])
|
|
session.add(run)
|
|
await session.commit()
|
|
run_id = run.id
|
|
|
|
with (
|
|
patch("app.api.routes.scan.AsyncSessionLocal", mem_db),
|
|
patch("app.api.routes.scan.run_scan", new_callable=AsyncMock) as mock_run_scan,
|
|
):
|
|
from app.services.scanner import DeepScanOptions
|
|
await _background_scan(run_id, ["10.0.0.0/24"], DeepScanOptions())
|
|
mock_run_scan.assert_awaited_once()
|
|
|
|
|
|
# --- Trigger scan ---
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_trigger_scan_requires_auth(client: AsyncClient):
|
|
res = await client.post("/api/v1/scan/trigger")
|
|
assert res.status_code == 401
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_trigger_scan_creates_run(client: AsyncClient, headers):
|
|
with (
|
|
patch("app.api.routes.scan._background_scan", new_callable=AsyncMock),
|
|
patch("app.api.routes.scan.settings") as mock_settings,
|
|
):
|
|
mock_settings.scanner_ranges = ["192.168.1.0/24"]
|
|
res = await client.post("/api/v1/scan/trigger", headers=headers)
|
|
assert res.status_code == 200
|
|
data = res.json()
|
|
assert data["status"] == "running"
|
|
assert data["ranges"] == ["192.168.1.0/24"]
|
|
assert "id" in data
|
|
|
|
|
|
# --- Pending devices ---
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_pending_empty(client: AsyncClient, headers):
|
|
res = await client.get("/api/v1/scan/pending", headers=headers)
|
|
assert res.status_code == 200
|
|
assert res.json() == []
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_pending_returns_device(client: AsyncClient, headers, pending_device):
|
|
res = await client.get("/api/v1/scan/pending", headers=headers)
|
|
assert res.status_code == 200
|
|
data = res.json()
|
|
assert len(data) == 1
|
|
assert data[0]["ip"] == "192.168.1.100"
|
|
assert data[0]["hostname"] == "my-server"
|
|
# No matching node → not on any canvas.
|
|
assert data[0]["canvas_count"] == 0
|
|
|
|
|
|
# --- Canvas-presence correlation (canvas_count) ---
|
|
|
|
async def _add_design(db_session, name: str) -> 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, *, ip=None, ieee=None) -> Node:
|
|
return Node(
|
|
id=str(uuid.uuid4()), label="n", type="server", status="online",
|
|
ip=ip, ieee_address=ieee, services=[], pos_x=0.0, pos_y=0.0,
|
|
design_id=design_id,
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_canvas_count_counts_distinct_designs_by_ip(client, headers, db_session, pending_device):
|
|
# Same IP placed on two different canvases → canvas_count == 2.
|
|
d1 = await _add_design(db_session, "Home")
|
|
d2 = await _add_design(db_session, "Lab")
|
|
db_session.add(_node(d1, ip="192.168.1.100"))
|
|
db_session.add(_node(d2, ip="192.168.1.100"))
|
|
await db_session.commit()
|
|
|
|
res = await client.get("/api/v1/scan/pending", headers=headers)
|
|
data = res.json()
|
|
assert len(data) == 1
|
|
assert data[0]["canvas_count"] == 2
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_canvas_count_correlates_by_ieee(client, headers, db_session):
|
|
device = PendingDevice(
|
|
id=str(uuid.uuid4()), ieee_address="0x00124b001", discovery_source="zigbee",
|
|
suggested_type="zigbee_enddevice", services=[], status="pending",
|
|
)
|
|
db_session.add(device)
|
|
d1 = await _add_design(db_session, "Zigbee")
|
|
db_session.add(_node(d1, ieee="0x00124b001"))
|
|
await db_session.commit()
|
|
|
|
res = await client.get("/api/v1/scan/pending", headers=headers)
|
|
by_id = {d["id"]: d for d in res.json()}
|
|
assert by_id[device.id]["canvas_count"] == 1
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_canvas_count_ignores_nodes_without_design(client, headers, db_session, pending_device):
|
|
# A node with no design_id is not "on a canvas".
|
|
db_session.add(_node(None, ip="192.168.1.100"))
|
|
await db_session.commit()
|
|
|
|
res = await client.get("/api/v1/scan/pending", headers=headers)
|
|
assert res.json()[0]["canvas_count"] == 0
|
|
|
|
|
|
# --- Linked-node timestamps on the inventory response ---
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_pending_device_without_node_has_null_node_timestamps(client, headers, pending_device):
|
|
# No matching canvas node → node_* timestamps are all null; the device still
|
|
# carries its own discovered_at for the "Discovered" fallback on the tile.
|
|
data = (await client.get("/api/v1/scan/pending", headers=headers)).json()[0]
|
|
assert data["discovered_at"] is not None
|
|
assert data["node_created_at"] is None
|
|
assert data["node_last_scan"] is None
|
|
assert data["node_last_modified"] is None
|
|
assert data["node_last_seen"] is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_pending_device_exposes_linked_node_timestamps(client, headers, db_session, pending_device):
|
|
d1 = await _add_design(db_session, "Home")
|
|
node = _node(d1, ip="192.168.1.100")
|
|
node.last_scan = datetime(2026, 6, 1, 8, 30, tzinfo=timezone.utc)
|
|
node.last_seen = datetime(2026, 6, 25, 9, 15, tzinfo=timezone.utc)
|
|
db_session.add(node)
|
|
await db_session.commit()
|
|
|
|
data = (await client.get("/api/v1/scan/pending", headers=headers)).json()[0]
|
|
assert data["node_created_at"] is not None # defaulted on insert
|
|
assert data["node_last_modified"] is not None # updated_at defaulted on insert
|
|
assert data["node_last_scan"].startswith("2026-06-01")
|
|
assert data["node_last_seen"].startswith("2026-06-25")
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_node_timestamps_aggregate_across_matches(client, headers, db_session, pending_device):
|
|
# Two canvas nodes share the device IP: created_at takes the OLDEST,
|
|
# last_scan takes the NEWEST.
|
|
d1 = await _add_design(db_session, "Home")
|
|
d2 = await _add_design(db_session, "Lab")
|
|
older = _node(d1, ip="192.168.1.100")
|
|
older.created_at = datetime(2026, 1, 1, 0, 0, tzinfo=timezone.utc)
|
|
older.last_scan = datetime(2026, 3, 1, 0, 0, tzinfo=timezone.utc)
|
|
newer = _node(d2, ip="192.168.1.100")
|
|
newer.created_at = datetime(2026, 5, 1, 0, 0, tzinfo=timezone.utc)
|
|
newer.last_scan = datetime(2026, 6, 1, 0, 0, tzinfo=timezone.utc)
|
|
db_session.add_all([older, newer])
|
|
await db_session.commit()
|
|
|
|
data = (await client.get("/api/v1/scan/pending", headers=headers)).json()[0]
|
|
assert data["node_created_at"].startswith("2026-01-01") # oldest
|
|
assert data["node_last_scan"].startswith("2026-06-01") # newest
|
|
|
|
|
|
# --- Approve device ---
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_approve_device(client: AsyncClient, headers, pending_device):
|
|
node_payload = {
|
|
"label": "My Server",
|
|
"type": "server",
|
|
"ip": "192.168.1.100",
|
|
"hostname": "my-server",
|
|
"status": "unknown",
|
|
"services": [],
|
|
}
|
|
res = await client.post(
|
|
f"/api/v1/scan/pending/{pending_device.id}/approve",
|
|
json=node_payload,
|
|
headers=headers,
|
|
)
|
|
assert res.status_code == 200
|
|
data = res.json()
|
|
assert data["approved"] is True
|
|
assert "node_id" in data
|
|
|
|
# Approved devices stay in the inventory (status != "hidden") so they keep
|
|
# showing with an "In N canvas" badge — they are no longer dropped.
|
|
pending_res = await client.get("/api/v1/scan/pending", headers=headers)
|
|
inventory = pending_res.json()
|
|
assert len(inventory) == 1
|
|
assert inventory[0]["id"] == pending_device.id
|
|
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
|
|
async def test_approve_nonexistent_device(client: AsyncClient, headers):
|
|
node_payload = {
|
|
"label": "Ghost",
|
|
"type": "generic",
|
|
"ip": "10.0.0.1",
|
|
"status": "unknown",
|
|
"services": [],
|
|
}
|
|
res = await client.post(
|
|
"/api/v1/scan/pending/nonexistent-id/approve",
|
|
json=node_payload,
|
|
headers=headers,
|
|
)
|
|
assert res.status_code == 404
|
|
|
|
|
|
# --- Hide device ---
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_hide_device(client: AsyncClient, headers, pending_device):
|
|
res = await client.post(f"/api/v1/scan/pending/{pending_device.id}/hide", headers=headers)
|
|
assert res.status_code == 200
|
|
assert res.json()["hidden"] is True
|
|
|
|
# Should no longer appear in pending
|
|
pending_res = await client.get("/api/v1/scan/pending", headers=headers)
|
|
assert pending_res.json() == []
|
|
|
|
# Should appear in hidden
|
|
hidden_res = await client.get("/api/v1/scan/hidden", headers=headers)
|
|
assert len(hidden_res.json()) == 1
|
|
|
|
|
|
# --- Restore hidden device ---
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_restore_device(client: AsyncClient, headers, pending_device):
|
|
# Hide first
|
|
await client.post(f"/api/v1/scan/pending/{pending_device.id}/hide", headers=headers)
|
|
|
|
# Restore
|
|
res = await client.post(f"/api/v1/scan/pending/{pending_device.id}/restore", headers=headers)
|
|
assert res.status_code == 200
|
|
assert res.json()["restored"] is True
|
|
|
|
# Now back in pending, gone from hidden
|
|
pending_res = await client.get("/api/v1/scan/pending", headers=headers)
|
|
assert len(pending_res.json()) == 1
|
|
hidden_res = await client.get("/api/v1/scan/hidden", headers=headers)
|
|
assert hidden_res.json() == []
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_restore_device_rejects_non_hidden(client: AsyncClient, headers, pending_device):
|
|
res = await client.post(f"/api/v1/scan/pending/{pending_device.id}/restore", headers=headers)
|
|
assert res.status_code == 409
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_bulk_restore_devices(client: AsyncClient, headers, pending_device):
|
|
# Hide
|
|
await client.post(f"/api/v1/scan/pending/{pending_device.id}/hide", headers=headers)
|
|
|
|
res = await client.post(
|
|
"/api/v1/scan/pending/bulk-restore",
|
|
headers=headers,
|
|
json={"device_ids": [pending_device.id]},
|
|
)
|
|
assert res.status_code == 200
|
|
assert res.json()["restored"] == 1
|
|
assert res.json()["skipped"] == 0
|
|
|
|
pending_res = await client.get("/api/v1/scan/pending", headers=headers)
|
|
assert len(pending_res.json()) == 1
|
|
|
|
|
|
# --- Ignore device ---
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_ignore_device(client: AsyncClient, headers, pending_device):
|
|
res = await client.post(f"/api/v1/scan/pending/{pending_device.id}/ignore", headers=headers)
|
|
assert res.status_code == 200
|
|
assert res.json()["ignored"] is True
|
|
|
|
# Device should be gone from both pending and hidden
|
|
pending_res = await client.get("/api/v1/scan/pending", headers=headers)
|
|
assert pending_res.json() == []
|
|
hidden_res = await client.get("/api/v1/scan/hidden", headers=headers)
|
|
assert hidden_res.json() == []
|
|
|
|
|
|
# --- Scan runs ---
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_runs_empty(client: AsyncClient, headers):
|
|
res = await client.get("/api/v1/scan/runs", headers=headers)
|
|
assert res.status_code == 200
|
|
assert res.json() == []
|
|
|
|
|
|
# --- run_scan: re-scan updates existing pending devices ---
|
|
|
|
MOCK_HOST = {
|
|
"ip": "192.168.1.50",
|
|
"mac": "aa:bb:cc:dd:ee:ff",
|
|
"hostname": "myhost.lan",
|
|
"os": "Linux",
|
|
"open_ports": [{"port": 8096, "protocol": "tcp", "banner": "Jellyfin"}],
|
|
}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_run_scan_creates_new_pending_device(db_session: AsyncSession):
|
|
run_id = str(uuid.uuid4())
|
|
run = ScanRun(id=run_id, status="running", ranges=["192.168.1.0/24"])
|
|
db_session.add(run)
|
|
await db_session.commit()
|
|
|
|
with (
|
|
patch("app.services.scanner._nmap_scan", return_value=[MOCK_HOST]),
|
|
patch("app.api.routes.status.broadcast_scan_update", new_callable=AsyncMock),
|
|
):
|
|
await run_scan(["192.168.1.0/24"], db_session, run_id)
|
|
|
|
result = await db_session.execute(
|
|
select(PendingDevice).where(PendingDevice.ip == "192.168.1.50")
|
|
)
|
|
device = result.scalar_one_or_none()
|
|
assert device is not None
|
|
assert device.hostname == "myhost.lan"
|
|
assert any(s["port"] == 8096 for s in device.services)
|
|
assert device.suggested_type == "server"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_run_scan_keeps_stale_pending_for_canvas_nodes(db_session: AsyncSession):
|
|
"""Pending devices whose IP is already on a canvas are NOT purged — they stay
|
|
in the inventory and are surfaced with an "In N canvas" badge."""
|
|
node = Node(
|
|
id=str(uuid.uuid4()),
|
|
label="Existing Server",
|
|
type="server",
|
|
ip="192.168.1.50",
|
|
status="online",
|
|
services=[],
|
|
pos_x=0.0,
|
|
pos_y=0.0,
|
|
)
|
|
stale = PendingDevice(
|
|
id=str(uuid.uuid4()),
|
|
ip="192.168.1.50",
|
|
mac=None,
|
|
hostname=None,
|
|
os=None,
|
|
services=[],
|
|
suggested_type="generic",
|
|
status="pending",
|
|
)
|
|
db_session.add(node)
|
|
db_session.add(stale)
|
|
await db_session.commit()
|
|
|
|
run_id = str(uuid.uuid4())
|
|
run = ScanRun(id=run_id, status="running", ranges=["192.168.1.0/24"])
|
|
db_session.add(run)
|
|
await db_session.commit()
|
|
|
|
with (
|
|
patch("app.services.scanner._nmap_scan", return_value=[]),
|
|
patch("app.api.routes.status.broadcast_scan_update", new_callable=AsyncMock),
|
|
):
|
|
await run_scan(["192.168.1.0/24"], db_session, run_id)
|
|
|
|
result = await db_session.execute(
|
|
select(PendingDevice).where(PendingDevice.ip == "192.168.1.50")
|
|
)
|
|
assert result.scalar_one_or_none() is not None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_run_scan_records_ip_already_in_canvas(db_session: AsyncSession):
|
|
"""A scanned IP that already exists as a canvas Node still produces a pending
|
|
device (no longer suppressed)."""
|
|
node = Node(
|
|
id=str(uuid.uuid4()),
|
|
label="Existing Server",
|
|
type="server",
|
|
ip="192.168.1.50",
|
|
status="online",
|
|
services=[],
|
|
pos_x=0.0,
|
|
pos_y=0.0,
|
|
)
|
|
db_session.add(node)
|
|
await db_session.commit()
|
|
|
|
run_id = str(uuid.uuid4())
|
|
run = ScanRun(id=run_id, status="running", ranges=["192.168.1.0/24"])
|
|
db_session.add(run)
|
|
await db_session.commit()
|
|
|
|
with (
|
|
patch("app.services.scanner._nmap_scan", return_value=[MOCK_HOST]),
|
|
patch("app.api.routes.status.broadcast_scan_update", new_callable=AsyncMock),
|
|
):
|
|
await run_scan(["192.168.1.0/24"], db_session, run_id)
|
|
|
|
result = await db_session.execute(
|
|
select(PendingDevice).where(PendingDevice.ip == "192.168.1.50")
|
|
)
|
|
device = result.scalar_one_or_none()
|
|
assert device is not None
|
|
assert device.status == "pending"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_run_scan_refreshes_approved_device_without_duplicating(db_session: AsyncSession):
|
|
"""Re-scanning an already-approved device updates its row in place instead of
|
|
spawning a fresh pending duplicate, and keeps it approved."""
|
|
approved = PendingDevice(
|
|
id=str(uuid.uuid4()), ip="192.168.1.50", mac=None, hostname="old",
|
|
os=None, services=[], suggested_type="server", status="approved",
|
|
)
|
|
db_session.add(approved)
|
|
run_id = str(uuid.uuid4())
|
|
db_session.add(ScanRun(id=run_id, status="running", ranges=["192.168.1.0/24"]))
|
|
await db_session.commit()
|
|
|
|
with (
|
|
patch("app.services.scanner._nmap_scan", return_value=[MOCK_HOST]),
|
|
patch("app.api.routes.status.broadcast_scan_update", new_callable=AsyncMock),
|
|
):
|
|
await run_scan(["192.168.1.0/24"], db_session, run_id)
|
|
|
|
rows = (await db_session.execute(
|
|
select(PendingDevice).where(PendingDevice.ip == "192.168.1.50")
|
|
)).scalars().all()
|
|
assert len(rows) == 1
|
|
assert rows[0].status == "approved"
|
|
assert rows[0].hostname == "myhost.lan" # refreshed from the scan
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_run_scan_collapses_existing_duplicate_rows(db_session: AsyncSession):
|
|
"""Pre-existing duplicate inventory rows for one IP are collapsed to a single
|
|
row at scan start, even if the device is not re-discovered."""
|
|
for status in ("approved", "pending", "pending"):
|
|
db_session.add(PendingDevice(
|
|
id=str(uuid.uuid4()), ip="192.168.1.77", mac=None, hostname=None,
|
|
os=None, services=[], suggested_type="server", status=status,
|
|
))
|
|
run_id = str(uuid.uuid4())
|
|
db_session.add(ScanRun(id=run_id, status="running", ranges=["192.168.1.0/24"]))
|
|
await db_session.commit()
|
|
|
|
with (
|
|
patch("app.services.scanner._nmap_scan", return_value=[]),
|
|
patch("app.api.routes.status.broadcast_scan_update", new_callable=AsyncMock),
|
|
):
|
|
await run_scan(["192.168.1.0/24"], db_session, run_id)
|
|
|
|
rows = (await db_session.execute(
|
|
select(PendingDevice).where(PendingDevice.ip == "192.168.1.77")
|
|
)).scalars().all()
|
|
assert len(rows) == 1
|
|
assert rows[0].status == "approved" # approved row is the one kept
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_run_scan_skips_hidden_device(db_session: AsyncSession):
|
|
"""Devices previously hidden by the user must not re-appear in pending on re-scan."""
|
|
hidden = PendingDevice(
|
|
id=str(uuid.uuid4()),
|
|
ip="192.168.1.50",
|
|
mac=None,
|
|
hostname=None,
|
|
os=None,
|
|
services=[],
|
|
suggested_type="generic",
|
|
status="hidden",
|
|
)
|
|
db_session.add(hidden)
|
|
await db_session.commit()
|
|
|
|
run_id = str(uuid.uuid4())
|
|
run = ScanRun(id=run_id, status="running", ranges=["192.168.1.0/24"])
|
|
db_session.add(run)
|
|
await db_session.commit()
|
|
|
|
with (
|
|
patch("app.services.scanner._nmap_scan", return_value=[MOCK_HOST]),
|
|
patch("app.api.routes.status.broadcast_scan_update", new_callable=AsyncMock),
|
|
):
|
|
await run_scan(["192.168.1.0/24"], db_session, run_id)
|
|
|
|
result = await db_session.execute(
|
|
select(PendingDevice).where(
|
|
PendingDevice.ip == "192.168.1.50",
|
|
PendingDevice.status == "pending",
|
|
)
|
|
)
|
|
assert result.scalar_one_or_none() is None
|
|
|
|
|
|
# --- Stop scan ---
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_stop_scan_requires_auth(client: AsyncClient):
|
|
res = await client.post("/api/v1/scan/fake-id/stop")
|
|
assert res.status_code == 401
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_stop_scan_not_found(client: AsyncClient, headers):
|
|
import uuid as _uuid
|
|
res = await client.post(f"/api/v1/scan/{_uuid.uuid4()}/stop", headers=headers)
|
|
assert res.status_code == 404
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_stop_scan_not_running(client: AsyncClient, headers, db_session: AsyncSession):
|
|
run = ScanRun(id=str(uuid.uuid4()), status="done", ranges=["192.168.1.0/24"])
|
|
db_session.add(run)
|
|
await db_session.commit()
|
|
|
|
res = await client.post(f"/api/v1/scan/{run.id}/stop", headers=headers)
|
|
assert res.status_code == 409
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_stop_scan_success(client: AsyncClient, headers, db_session: AsyncSession):
|
|
run = ScanRun(id=str(uuid.uuid4()), status="running", ranges=["192.168.1.0/24"])
|
|
db_session.add(run)
|
|
await db_session.commit()
|
|
|
|
res = await client.post(f"/api/v1/scan/{run.id}/stop", headers=headers)
|
|
assert res.status_code == 200
|
|
assert res.json() == {"stopping": True}
|
|
# run_id added to cancel set
|
|
assert run.id in _cancelled_runs
|
|
# status flipped eagerly so the UI reacts without waiting for a checkpoint
|
|
await db_session.refresh(run)
|
|
assert run.status == "cancelled"
|
|
assert run.finished_at is not None
|
|
# cleanup for other tests
|
|
_cancelled_runs.discard(run.id)
|
|
|
|
|
|
# --- run_scan cancellation ---
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_run_scan_cancelled_marks_status(db_session: AsyncSession):
|
|
"""When cancel is requested before the scan starts, status becomes 'cancelled'."""
|
|
run_id = str(uuid.uuid4())
|
|
run = ScanRun(id=run_id, status="running", ranges=["192.168.1.0/24"])
|
|
db_session.add(run)
|
|
await db_session.commit()
|
|
|
|
request_cancel(run_id)
|
|
|
|
with (
|
|
patch("app.services.scanner._nmap_scan", return_value=[MOCK_HOST]) as mock_nmap,
|
|
patch("app.api.routes.status.broadcast_scan_update", new_callable=AsyncMock),
|
|
):
|
|
await run_scan(["192.168.1.0/24"], db_session, run_id)
|
|
# nmap should not have been called — cancelled before first range
|
|
mock_nmap.assert_not_called()
|
|
|
|
await db_session.refresh(run)
|
|
assert run.status == "cancelled"
|
|
assert run.finished_at is not None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_run_scan_cancelled_mid_scan_skips_remaining_cidrs(db_session: AsyncSession):
|
|
"""Cancel flag set after first CIDR is started prevents processing of the second CIDR."""
|
|
run_id = str(uuid.uuid4())
|
|
run = ScanRun(id=run_id, status="running", ranges=["10.0.0.0/24", "10.0.1.0/24"])
|
|
db_session.add(run)
|
|
await db_session.commit()
|
|
|
|
call_count = 0
|
|
|
|
def nmap_side_effect(target: str, port_spec: str | None = None, run_id: str | None = None):
|
|
nonlocal call_count
|
|
call_count += 1
|
|
# Signal cancellation after the first CIDR scan completes
|
|
if call_count == 1:
|
|
request_cancel(run_id)
|
|
return []
|
|
|
|
with (
|
|
patch("app.services.scanner._nmap_scan", side_effect=nmap_side_effect),
|
|
patch("app.api.routes.status.broadcast_scan_update", new_callable=AsyncMock),
|
|
):
|
|
await run_scan(["10.0.0.0/24", "10.0.1.0/24"], db_session, run_id)
|
|
|
|
assert call_count == 1 # second CIDR was skipped
|
|
await db_session.refresh(run)
|
|
assert run.status == "cancelled"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_run_scan_updates_existing_pending_device(db_session: AsyncSession):
|
|
"""Re-scanning the same IP updates services instead of creating a duplicate."""
|
|
# Pre-existing pending device with no services
|
|
existing = PendingDevice(
|
|
id=str(uuid.uuid4()),
|
|
ip="192.168.1.50",
|
|
mac=None,
|
|
hostname=None,
|
|
os=None,
|
|
services=[],
|
|
suggested_type="generic",
|
|
status="pending",
|
|
)
|
|
db_session.add(existing)
|
|
await db_session.commit()
|
|
|
|
run_id = str(uuid.uuid4())
|
|
run = ScanRun(id=run_id, status="running", ranges=["192.168.1.0/24"])
|
|
db_session.add(run)
|
|
await db_session.commit()
|
|
|
|
with (
|
|
patch("app.services.scanner._nmap_scan", return_value=[MOCK_HOST]),
|
|
patch("app.api.routes.status.broadcast_scan_update", new_callable=AsyncMock),
|
|
):
|
|
await run_scan(["192.168.1.0/24"], db_session, run_id)
|
|
|
|
# Should still be only one device
|
|
result = await db_session.execute(
|
|
select(PendingDevice).where(PendingDevice.ip == "192.168.1.50")
|
|
)
|
|
devices = list(result.scalars().all())
|
|
assert len(devices) == 1
|
|
device = devices[0]
|
|
# Services and hostname should be updated
|
|
assert device.hostname == "myhost.lan"
|
|
assert any(s["port"] == 8096 for s in device.services)
|
|
|
|
|
|
# --- Bulk approve ---
|
|
|
|
@pytest.fixture
|
|
async def two_pending_devices(db_session):
|
|
devices = []
|
|
for i in range(2):
|
|
d = PendingDevice(
|
|
id=str(uuid.uuid4()),
|
|
ip=f"192.168.1.{10 + i}",
|
|
mac=None,
|
|
hostname=f"host-{i}",
|
|
os=None,
|
|
services=[],
|
|
suggested_type="generic",
|
|
status="pending",
|
|
)
|
|
db_session.add(d)
|
|
devices.append(d)
|
|
await db_session.commit()
|
|
for d in devices:
|
|
await db_session.refresh(d)
|
|
return devices
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_bulk_approve_approves_devices(client: AsyncClient, headers, two_pending_devices):
|
|
ids = [d.id for d in two_pending_devices]
|
|
res = await client.post("/api/v1/scan/pending/bulk-approve", json={"device_ids": ids}, headers=headers)
|
|
assert res.status_code == 200
|
|
data = res.json()
|
|
assert data["approved"] == 2
|
|
assert len(data["node_ids"]) == 2
|
|
assert all(nid is not None for nid in data["node_ids"]), "node_ids must be non-null UUIDs"
|
|
assert len(data["device_ids"]) == 2
|
|
assert data["skipped"] == 0
|
|
# Approved devices stay in the inventory, now marked "approved".
|
|
pending_res = await client.get("/api/v1/scan/pending", headers=headers)
|
|
inventory = pending_res.json()
|
|
assert len(inventory) == 2
|
|
assert all(d["status"] == "approved" for d in inventory)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_bulk_approve_places_already_approved_device_on_another_design(
|
|
client: AsyncClient, headers, db_session, two_pending_devices
|
|
):
|
|
"""Regression: a device already approved (status='approved', e.g. placed on
|
|
another canvas) must still get a node on the design being approved onto.
|
|
|
|
Previously bulk-approve filtered status=='pending', so selecting an
|
|
already-approved device created no node — the user saw fewer nodes than
|
|
they selected."""
|
|
ids = [d.id for d in two_pending_devices]
|
|
design_a = await _add_design(db_session, "Canvas A")
|
|
design_b = await _add_design(db_session, "Canvas B")
|
|
|
|
# Approve both onto design A.
|
|
res_a = await client.post(
|
|
"/api/v1/scan/pending/bulk-approve",
|
|
json={"device_ids": ids, "design_id": design_a},
|
|
headers=headers,
|
|
)
|
|
assert res_a.json()["approved"] == 2
|
|
|
|
# Re-approve the same (now status='approved') devices onto design B.
|
|
res_b = await client.post(
|
|
"/api/v1/scan/pending/bulk-approve",
|
|
json={"device_ids": ids, "design_id": design_b},
|
|
headers=headers,
|
|
)
|
|
data_b = res_b.json()
|
|
assert data_b["approved"] == 2, "already-approved devices must place onto the new canvas"
|
|
assert data_b["skipped"] == 0
|
|
|
|
# Two nodes now exist on each design.
|
|
from app.db.models import Node as NodeModel
|
|
nodes_b = (
|
|
await db_session.execute(select(NodeModel).where(NodeModel.design_id == design_b))
|
|
).scalars().all()
|
|
assert len(nodes_b) == 2
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_bulk_approve_skips_device_already_on_target_design(
|
|
client: AsyncClient, headers, db_session, two_pending_devices
|
|
):
|
|
"""A device already on the target canvas (same ip) is not placed twice."""
|
|
ids = [d.id for d in two_pending_devices]
|
|
design = await _add_design(db_session, "Canvas")
|
|
# First device already sits on the canvas (matched by ip).
|
|
db_session.add(_node(design, ip="192.168.1.10"))
|
|
await db_session.commit()
|
|
|
|
res = await client.post(
|
|
"/api/v1/scan/pending/bulk-approve",
|
|
json={"device_ids": ids, "design_id": design},
|
|
headers=headers,
|
|
)
|
|
data = res.json()
|
|
assert data["approved"] == 1 # only the second device (192.168.1.11)
|
|
assert data["skipped"] == 1
|
|
|
|
from app.db.models import Node as NodeModel
|
|
nodes = (
|
|
await db_session.execute(select(NodeModel).where(NodeModel.design_id == design))
|
|
).scalars().all()
|
|
# The pre-existing node plus the one newly approved — no duplicate for .10.
|
|
assert len(nodes) == 2
|
|
assert sorted(n.ip for n in nodes) == ["192.168.1.10", "192.168.1.11"]
|
|
|
|
|
|
@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"
|
|
|
|
|
|
# --- MAC address propagation on approve (issue #168) ---
|
|
|
|
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_copies_mac_to_node_and_properties(
|
|
client: AsyncClient, headers, pending_device, db_session
|
|
):
|
|
"""Approving a scanned device must carry its MAC onto the node + properties."""
|
|
from sqlalchemy import select
|
|
|
|
from app.db.models import Node as NodeModel
|
|
# Payload intentionally omits mac — it must come from the pending device.
|
|
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": []},
|
|
headers=headers,
|
|
)
|
|
assert res.status_code == 200
|
|
node = (
|
|
await db_session.execute(select(NodeModel).where(NodeModel.ip == "192.168.1.100"))
|
|
).scalar_one()
|
|
assert node.mac == "aa:bb:cc:dd:ee:ff"
|
|
mac_props = [p for p in node.properties if p["key"] == "MAC"]
|
|
assert mac_props == [
|
|
{"key": "MAC", "value": "aa:bb:cc:dd:ee:ff", "icon": None, "visible": False}
|
|
]
|
|
|
|
|
|
@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.ip == "192.168.1.100"))
|
|
).scalar_one()
|
|
mac_props = [p for p in node.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_bulk_approve_copies_mac_to_node_and_properties(
|
|
client: AsyncClient, headers, db_session
|
|
):
|
|
"""Bulk approve must also propagate the scanned MAC to node + properties."""
|
|
from sqlalchemy import select
|
|
|
|
from app.db.models import Node as NodeModel
|
|
device = PendingDevice(
|
|
id=str(uuid.uuid4()),
|
|
ip="192.168.1.55",
|
|
mac="11:22:33:44:55:66",
|
|
hostname="host-mac",
|
|
services=[],
|
|
suggested_type="generic",
|
|
status="pending",
|
|
)
|
|
db_session.add(device)
|
|
await db_session.commit()
|
|
|
|
res = await client.post(
|
|
"/api/v1/scan/pending/bulk-approve",
|
|
json={"device_ids": [device.id]},
|
|
headers=headers,
|
|
)
|
|
assert res.status_code == 200
|
|
node = (
|
|
await db_session.execute(select(NodeModel).where(NodeModel.ip == "192.168.1.55"))
|
|
).scalar_one()
|
|
assert node.mac == "11:22:33:44:55:66"
|
|
mac_props = [p for p in node.properties if p["key"] == "MAC"]
|
|
assert mac_props == [
|
|
{"key": "MAC", "value": "11:22:33:44:55:66", "icon": None, "visible": False}
|
|
]
|
|
|
|
|
|
@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."""
|
|
from sqlalchemy import select
|
|
|
|
from app.db.models import Node as NodeModel
|
|
ids = [d.id for d in two_pending_devices]
|
|
res = await client.post("/api/v1/scan/pending/bulk-approve", json={"device_ids": ids}, headers=headers)
|
|
assert res.status_code == 200
|
|
nodes = (await db_session.execute(select(NodeModel))).scalars().all()
|
|
for n in nodes:
|
|
if n.ip:
|
|
assert n.check_method == "ping", f"node {n.id} created without check_method"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_approve_device_sets_default_check_method(client: AsyncClient, headers, pending_device, db_session):
|
|
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": "h", "type": "generic", "ip": "192.168.1.10", "status": "unknown", "services": []},
|
|
headers=headers,
|
|
)
|
|
assert res.status_code == 200
|
|
node = (await db_session.execute(select(NodeModel))).scalars().first()
|
|
assert node is not None
|
|
assert node.check_method == "ping"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_bulk_approve_skips_already_approved(client: AsyncClient, headers, two_pending_devices):
|
|
ids = [d.id for d in two_pending_devices]
|
|
# Approve first device individually first
|
|
await client.post(
|
|
f"/api/v1/scan/pending/{ids[0]}/approve",
|
|
json={"label": "h", "type": "generic", "ip": "192.168.1.10", "status": "unknown", "services": []},
|
|
headers=headers,
|
|
)
|
|
# Bulk approve both — first one is already approved (not pending), should be skipped
|
|
res = await client.post("/api/v1/scan/pending/bulk-approve", json={"device_ids": ids}, headers=headers)
|
|
assert res.status_code == 200
|
|
data = res.json()
|
|
assert data["approved"] == 1
|
|
assert data["skipped"] == 1
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_bulk_approve_requires_auth(client: AsyncClient, two_pending_devices):
|
|
ids = [d.id for d in two_pending_devices]
|
|
res = await client.post("/api/v1/scan/pending/bulk-approve", json={"device_ids": ids})
|
|
assert res.status_code == 401
|
|
|
|
|
|
# --- Bulk hide ---
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_bulk_hide_hides_devices(client: AsyncClient, headers, two_pending_devices):
|
|
ids = [d.id for d in two_pending_devices]
|
|
res = await client.post("/api/v1/scan/pending/bulk-hide", json={"device_ids": ids}, headers=headers)
|
|
assert res.status_code == 200
|
|
data = res.json()
|
|
assert data["hidden"] == 2
|
|
assert data["skipped"] == 0
|
|
# Should appear in hidden list
|
|
hidden_res = await client.get("/api/v1/scan/hidden", headers=headers)
|
|
assert len(hidden_res.json()) == 2
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_bulk_hide_skips_non_pending(client: AsyncClient, headers, two_pending_devices):
|
|
ids = [d.id for d in two_pending_devices]
|
|
# Hide first device individually first
|
|
await client.post(f"/api/v1/scan/pending/{ids[0]}/hide", headers=headers)
|
|
# Bulk hide both — first is already hidden (not pending anymore)
|
|
res = await client.post("/api/v1/scan/pending/bulk-hide", json={"device_ids": ids}, headers=headers)
|
|
assert res.status_code == 200
|
|
data = res.json()
|
|
assert data["hidden"] == 1
|
|
assert data["skipped"] == 1
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_bulk_hide_requires_auth(client: AsyncClient, two_pending_devices):
|
|
ids = [d.id for d in two_pending_devices]
|
|
res = await client.post("/api/v1/scan/pending/bulk-hide", json={"device_ids": ids})
|
|
assert res.status_code == 401
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Approve auto-creates Edges from pending_device_links (Zigbee flow)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
async def _seed_zigbee_pending_pair(db_session):
|
|
"""Create a coordinator Node + a pending device + a link between them."""
|
|
from app.db.models import Node, PendingDevice, PendingDeviceLink
|
|
|
|
coord = Node(
|
|
label="Coordinator",
|
|
type="zigbee_coordinator",
|
|
status="unknown",
|
|
ieee_address="0xCOORD",
|
|
)
|
|
db_session.add(coord)
|
|
|
|
pending = PendingDevice(
|
|
ieee_address="0xR1",
|
|
friendly_name="router_1",
|
|
suggested_type="zigbee_router",
|
|
device_subtype="Router",
|
|
status="pending",
|
|
discovery_source="zigbee",
|
|
)
|
|
db_session.add(pending)
|
|
|
|
db_session.add(
|
|
PendingDeviceLink(
|
|
source_ieee="0xCOORD",
|
|
target_ieee="0xR1",
|
|
discovery_source="zigbee",
|
|
)
|
|
)
|
|
await db_session.commit()
|
|
return coord, pending
|
|
|
|
|
|
@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, PendingDevice, PendingDeviceLink
|
|
|
|
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 = PendingDevice(
|
|
ieee_address="0xR1B",
|
|
friendly_name="r1b",
|
|
suggested_type="zigbee_router",
|
|
status="pending",
|
|
discovery_source="zigbee",
|
|
)
|
|
db_session.add(new_pending)
|
|
db_session.add(
|
|
PendingDeviceLink(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, PendingDevice, PendingDeviceLink
|
|
|
|
a = PendingDevice(
|
|
ieee_address="0xA",
|
|
friendly_name="a",
|
|
suggested_type="zigbee_router",
|
|
status="pending",
|
|
discovery_source="zigbee",
|
|
)
|
|
b = PendingDevice(
|
|
ieee_address="0xB",
|
|
friendly_name="b",
|
|
suggested_type="zigbee_enddevice",
|
|
status="pending",
|
|
discovery_source="zigbee",
|
|
)
|
|
db_session.add_all([a, b])
|
|
db_session.add(
|
|
PendingDeviceLink(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(PendingDeviceLink))).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, PendingDevice, PendingDeviceLink
|
|
|
|
a = PendingDevice(
|
|
ieee_address="0xA",
|
|
friendly_name="a",
|
|
suggested_type="zigbee_router",
|
|
status="pending",
|
|
discovery_source="zigbee",
|
|
)
|
|
b = PendingDevice(
|
|
ieee_address="0xB",
|
|
friendly_name="b",
|
|
suggested_type="zigbee_enddevice",
|
|
status="pending",
|
|
discovery_source="zigbee",
|
|
)
|
|
db_session.add_all([a, b])
|
|
db_session.add(
|
|
PendingDeviceLink(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(PendingDeviceLink))).scalars().all()
|
|
assert len(links) == 1 # retained for re-approval onto other canvases
|
|
|
|
|
|
# --- Deep scan: trigger overrides + config persistence ---
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_resolve_deep_scan_falls_back_to_settings():
|
|
from app.api.routes.scan import TriggerScanRequest, _resolve_deep_scan
|
|
|
|
with patch("app.api.routes.scan.settings") as mock_settings:
|
|
mock_settings.scanner_http_ranges = ["7000-7100"]
|
|
mock_settings.scanner_http_probe_enabled = True
|
|
mock_settings.scanner_http_verify_tls = False
|
|
# Empty payload → all values come from settings defaults
|
|
ds = _resolve_deep_scan(TriggerScanRequest())
|
|
assert ds.http_ranges == ["7000-7100"]
|
|
assert ds.http_probe_enabled is True
|
|
assert ds.verify_tls is False
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_resolve_deep_scan_override_wins():
|
|
from app.api.routes.scan import TriggerScanRequest, _resolve_deep_scan
|
|
|
|
with patch("app.api.routes.scan.settings") as mock_settings:
|
|
mock_settings.scanner_http_ranges = []
|
|
mock_settings.scanner_http_probe_enabled = False
|
|
mock_settings.scanner_http_verify_tls = False
|
|
ds = _resolve_deep_scan(
|
|
TriggerScanRequest(http_ranges=["9000"], http_probe_enabled=True, verify_tls=True)
|
|
)
|
|
assert ds.http_ranges == ["9000"]
|
|
assert ds.http_probe_enabled is True
|
|
assert ds.verify_tls is True
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_trigger_scan_passes_deep_scan_options(client: AsyncClient, headers):
|
|
captured = {}
|
|
|
|
async def fake_bg(run_id, ranges, deep_scan):
|
|
captured["deep_scan"] = deep_scan
|
|
|
|
with (
|
|
patch("app.api.routes.scan._background_scan", new=fake_bg),
|
|
patch("app.api.routes.scan.settings") as mock_settings,
|
|
):
|
|
mock_settings.scanner_ranges = ["192.168.1.0/24"]
|
|
mock_settings.scanner_http_ranges = []
|
|
mock_settings.scanner_http_probe_enabled = False
|
|
mock_settings.scanner_http_verify_tls = False
|
|
res = await client.post(
|
|
"/api/v1/scan/trigger",
|
|
json={"http_probe_enabled": True, "http_ranges": ["8000-8100"]},
|
|
headers=headers,
|
|
)
|
|
assert res.status_code == 200
|
|
assert captured["deep_scan"].http_probe_enabled is True
|
|
assert captured["deep_scan"].http_ranges == ["8000-8100"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_trigger_scan_rejects_invalid_port_range(client: AsyncClient, headers):
|
|
with patch("app.api.routes.scan.settings") as mock_settings:
|
|
mock_settings.scanner_ranges = ["192.168.1.0/24"]
|
|
res = await client.post(
|
|
"/api/v1/scan/trigger",
|
|
json={"http_ranges": ["70000-80000"]},
|
|
headers=headers,
|
|
)
|
|
assert res.status_code == 422
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_scan_config_includes_deep_scan(client: AsyncClient, headers):
|
|
with patch("app.api.routes.scan.settings") as mock_settings:
|
|
mock_settings.scanner_ranges = ["192.168.1.0/24"]
|
|
mock_settings.scanner_http_ranges = ["8000-8100"]
|
|
mock_settings.scanner_http_probe_enabled = True
|
|
mock_settings.scanner_http_verify_tls = False
|
|
res = await client.get("/api/v1/scan/config", headers=headers)
|
|
assert res.status_code == 200
|
|
data = res.json()
|
|
assert data["http_ranges"] == ["8000-8100"]
|
|
assert data["http_probe_enabled"] is True
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_scan_config_persists_deep_scan(client: AsyncClient, headers):
|
|
saved = {}
|
|
|
|
with patch("app.api.routes.scan.settings") as mock_settings:
|
|
mock_settings.scanner_ranges = ["192.168.1.0/24"]
|
|
mock_settings.scanner_http_ranges = []
|
|
mock_settings.scanner_http_probe_enabled = False
|
|
mock_settings.scanner_http_verify_tls = False
|
|
mock_settings.save_overrides = lambda: saved.update(
|
|
http_ranges=mock_settings.scanner_http_ranges,
|
|
probe=mock_settings.scanner_http_probe_enabled,
|
|
)
|
|
res = await client.post(
|
|
"/api/v1/scan/config",
|
|
json={
|
|
"ranges": ["192.168.1.0/24"],
|
|
"http_ranges": ["9000-9100"],
|
|
"http_probe_enabled": True,
|
|
"verify_tls": True,
|
|
},
|
|
headers=headers,
|
|
)
|
|
assert res.status_code == 200
|
|
assert saved == {"http_ranges": ["9000-9100"], "probe": True}
|
|
|
|
|
|
# --- Z-Wave approve: active design targeting + wireless props (regression) ---
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_bulk_approve_targets_requested_design(client, headers, db_session):
|
|
"""bulk-approve must place nodes on the design_id sent by the UI, not the
|
|
first design — otherwise approved devices land on the wrong canvas."""
|
|
first = await _add_design(db_session, "Default") # first design (fallback)
|
|
active = await _add_design(db_session, "zwave") # the design the user is on
|
|
dev = PendingDevice(
|
|
id=str(uuid.uuid4()),
|
|
ieee_address="zwave-H-2",
|
|
friendly_name="Living Room Plug",
|
|
suggested_type="zwave_router",
|
|
device_subtype="Router",
|
|
vendor="Aeotec",
|
|
model="ZW096",
|
|
status="pending",
|
|
discovery_source="zwave",
|
|
)
|
|
db_session.add(dev)
|
|
await db_session.commit()
|
|
|
|
res = await client.post(
|
|
"/api/v1/scan/pending/bulk-approve",
|
|
json={"device_ids": [dev.id], "design_id": active},
|
|
headers=headers,
|
|
)
|
|
assert res.status_code == 200
|
|
assert res.json()["approved"] == 1
|
|
|
|
node = (
|
|
await db_session.execute(select(Node).where(Node.ieee_address == "zwave-H-2"))
|
|
).scalar_one()
|
|
assert node.design_id == active
|
|
assert node.design_id != first
|
|
# Z-Wave device → online + Z-Wave property rows, no ICMP check.
|
|
assert node.status == "online"
|
|
assert node.check_method == "none"
|
|
assert {p["key"] for p in node.properties} == {"Z-Wave ID", "Vendor", "Model"}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_single_approve_zwave_sets_wireless_fields(client, headers, db_session):
|
|
active = await _add_design(db_session, "zwave")
|
|
dev = PendingDevice(
|
|
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).where(Node.ieee_address == "zwave-H-9"))
|
|
).scalar_one()
|
|
assert node.design_id == active
|
|
assert node.status == "online"
|
|
assert node.check_method == "none"
|
|
assert any(p["key"] == "Z-Wave ID" for p in node.properties)
|