Pouzor
13420bead8
Approve (single + bulk) created the canvas Node under the first design instead of the design the user is viewing, so approved devices were invisible on the active canvas and got wiped on the next save — and a re-approve returned "0 approved" because the rows were already approved. - bulk-approve now accepts design_id; the UI sends the active design. - single approve already honoured design_id; the UI now sends it too. - generalize the wireless branch (status=online, mesh props, no ICMP check) to Z-Wave as well as Zigbee, using build_zwave_properties. ha-relevant: yes
1433 lines
48 KiB
Python
1433 lines
48 KiB
Python
"""Tests for scan routes: trigger, pending devices, approve/hide/ignore, stop."""
|
|
import uuid
|
|
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
|
|
|
|
|
|
# --- 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_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
|
|
# 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):
|
|
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.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"
|
|
assert edges[0].target_handle == "top-t"
|
|
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; second approval creates the edge."""
|
|
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 links == [] # consumed
|
|
|
|
|
|
# --- 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)
|