141 lines
4.9 KiB
Python
141 lines
4.9 KiB
Python
import ipaddress
|
|
import os
|
|
import unittest
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
from unittest.mock import AsyncMock, patch
|
|
|
|
from httpx import ASGITransport, AsyncClient
|
|
from sqlalchemy import delete
|
|
|
|
TEST_DB_PATH = Path(__file__).with_name("test_ignis_discovery.db")
|
|
if TEST_DB_PATH.exists():
|
|
TEST_DB_PATH.unlink()
|
|
|
|
MASTER_KEY = "master-secret-for-discovery-tests"
|
|
os.environ["IGNIS_API_KEY"] = MASTER_KEY
|
|
os.environ["IGNIS_DATABASE_URL"] = f"sqlite+aiosqlite:///{TEST_DB_PATH}"
|
|
os.environ["IGNIS_SYNC_DATABASE_URL"] = f"sqlite:///{TEST_DB_PATH}"
|
|
|
|
import main # noqa: E402
|
|
from app.core.database import async_session, init_db # noqa: E402
|
|
from app.core.discovery import DiscoveryService, InterfaceSubnet # noqa: E402
|
|
from app.core.state import state_manager # noqa: E402
|
|
from app.models.device import GroupModel # noqa: E402
|
|
|
|
|
|
class DiscoveryBehaviorTests(unittest.IsolatedAsyncioTestCase):
|
|
async def asyncSetUp(self):
|
|
os.environ["IGNIS_API_KEY"] = MASTER_KEY
|
|
await init_db()
|
|
await self._reset_database()
|
|
state_manager.devices.clear()
|
|
state_manager.groups.clear()
|
|
state_manager._missing_scan_counts.clear()
|
|
self.client = AsyncClient(
|
|
transport=ASGITransport(app=main.app),
|
|
base_url="http://testserver",
|
|
)
|
|
|
|
async def asyncTearDown(self):
|
|
await self.client.aclose()
|
|
state_manager.devices.clear()
|
|
state_manager.groups.clear()
|
|
state_manager._missing_scan_counts.clear()
|
|
|
|
async def _reset_database(self):
|
|
async with async_session() as session:
|
|
await session.execute(delete(GroupModel))
|
|
await session.commit()
|
|
|
|
def _headers(self) -> dict[str, str]:
|
|
return {"X-API-Key": MASTER_KEY}
|
|
|
|
def test_auto_subnets_prefer_non_vpn_private_interfaces(self):
|
|
service = DiscoveryService()
|
|
candidates = [
|
|
InterfaceSubnet(
|
|
name="wg0",
|
|
address=ipaddress.IPv4Address("10.8.0.2"),
|
|
network=ipaddress.IPv4Network("10.8.0.0/24"),
|
|
),
|
|
InterfaceSubnet(
|
|
name="wlan0",
|
|
address=ipaddress.IPv4Address("192.168.0.25"),
|
|
network=ipaddress.IPv4Network("192.168.0.0/24"),
|
|
),
|
|
InterfaceSubnet(
|
|
name="docker0",
|
|
address=ipaddress.IPv4Address("172.17.0.1"),
|
|
network=ipaddress.IPv4Network("172.17.0.0/16"),
|
|
),
|
|
InterfaceSubnet(
|
|
name="enp3s0",
|
|
address=ipaddress.IPv4Address("192.168.1.20"),
|
|
network=ipaddress.IPv4Network("192.168.0.0/23"),
|
|
),
|
|
]
|
|
|
|
with patch.dict(os.environ, {}, clear=True):
|
|
with patch.object(service, "_interface_subnets", return_value=candidates):
|
|
subnets = service._get_target_subnets()
|
|
|
|
self.assertEqual(subnets, ["192.168.0.0/24", "192.168.1.0/24"])
|
|
|
|
async def test_manual_rescan_updates_and_removes_devices_immediately(self):
|
|
state_manager.devices["stale-device"] = SimpleNamespace(
|
|
id="stale-device",
|
|
ip="192.168.0.10",
|
|
name="Old Lamp",
|
|
room="Office",
|
|
)
|
|
|
|
with patch.object(
|
|
main.discovery_service,
|
|
"scan_network",
|
|
AsyncMock(
|
|
return_value=[
|
|
{
|
|
"mac": "fresh-device",
|
|
"ip": "192.168.0.20",
|
|
"state": {"on": True, "dimming": 100, "temp": 4100},
|
|
}
|
|
]
|
|
),
|
|
):
|
|
response = await self.client.post(
|
|
"/devices/rescan",
|
|
headers=self._headers(),
|
|
)
|
|
|
|
self.assertEqual(response.status_code, 200)
|
|
payload = response.json()
|
|
self.assertEqual(payload["status"], "ok")
|
|
self.assertEqual(payload["found"], 1)
|
|
self.assertEqual(payload["added"], 1)
|
|
self.assertEqual(payload["updated"], 0)
|
|
self.assertEqual(payload["removed_offline"], 1)
|
|
self.assertEqual(payload["online"], 1)
|
|
self.assertEqual(list(state_manager.devices.keys()), ["fresh-device"])
|
|
|
|
def test_background_cleanup_requires_multiple_misses(self):
|
|
state_manager.update_device({"mac": "dev-1", "ip": "192.168.0.10"})
|
|
|
|
first = state_manager.apply_discovery_snapshot(
|
|
[],
|
|
remove_missing=True,
|
|
missing_threshold=2,
|
|
)
|
|
self.assertEqual(first.removed_offline, 0)
|
|
self.assertEqual(first.pending_removal, 1)
|
|
self.assertIn("dev-1", state_manager.devices)
|
|
|
|
second = state_manager.apply_discovery_snapshot(
|
|
[],
|
|
remove_missing=True,
|
|
missing_threshold=2,
|
|
)
|
|
self.assertEqual(second.removed_offline, 1)
|
|
self.assertEqual(second.pending_removal, 0)
|
|
self.assertNotIn("dev-1", state_manager.devices)
|