test: add coverage for none check method, re-scan update, icon registry
This commit is contained in:
@@ -1,10 +1,14 @@
|
||||
"""Tests for scan routes: trigger, pending devices, approve/hide/ignore."""
|
||||
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 PendingDevice
|
||||
from app.db.models import PendingDevice, ScanRun
|
||||
from app.services.scanner import run_scan
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
@@ -158,3 +162,77 @@ 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_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)
|
||||
|
||||
@@ -7,6 +7,22 @@ from app.services.status_checker import _tcp_connect, check_node
|
||||
|
||||
# --- check_node dispatcher ---
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_check_node_none_always_online():
|
||||
result = await check_node("none", None, None)
|
||||
assert result["status"] == "online"
|
||||
assert result["response_time_ms"] is None
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_check_node_none_always_online_ignores_host():
|
||||
"""'none' returns online immediately without any network call."""
|
||||
with patch("app.services.status_checker._ping", new_callable=AsyncMock) as mock_ping:
|
||||
result = await check_node("none", None, "192.168.1.1")
|
||||
mock_ping.assert_not_called()
|
||||
assert result["status"] == "online"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_check_node_unknown_without_host():
|
||||
result = await check_node("ping", None, None)
|
||||
|
||||
Reference in New Issue
Block a user