128 lines
5.5 KiB
Python
128 lines
5.5 KiB
Python
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
from librenet_scanner.actions import normalize_mac
|
|
from librenet_scanner.comparison import compare_hosts
|
|
from librenet_scanner.intelligence import canonical_service_name, enrich_host
|
|
from librenet_scanner.models import Host, PortInfo
|
|
from librenet_scanner.storage import HistoryStore
|
|
|
|
|
|
class IntelligenceTests(unittest.TestCase):
|
|
def test_canonical_proxmox_service(self):
|
|
self.assertEqual(canonical_service_name(8006, "tcp", "wpl-analytics"), "Proxmox VE")
|
|
|
|
def test_proxmox_classification(self):
|
|
host = Host("192.168.5.1", hostname="pve01.local", ports=[PortInfo(8006, service="wpl-analytics")])
|
|
enrich_host(host)
|
|
self.assertEqual(host.device_type, "Hyperviseur Proxmox")
|
|
self.assertEqual(host.ports[0].service, "Proxmox VE")
|
|
|
|
def test_synology_classification(self):
|
|
host = Host("192.168.5.11", hostname="syno.maison.bro", ports=[PortInfo(5001)])
|
|
enrich_host(host)
|
|
self.assertEqual(host.device_type, "NAS Synology")
|
|
|
|
def test_pbs_classification(self):
|
|
host = Host("192.168.5.12", hostname="pbs01.local.raspot.in", ports=[PortInfo(8007, service="ajp13")])
|
|
enrich_host(host)
|
|
self.assertEqual(host.device_type, "Proxmox Backup Server")
|
|
self.assertEqual(host.ports[0].service, "Proxmox Backup Server")
|
|
|
|
def test_wol_mac_normalization(self):
|
|
self.assertEqual(normalize_mac("aa-bb-cc-dd-ee-ff"), "AA:BB:CC:DD:EE:FF")
|
|
with self.assertRaises(ValueError):
|
|
normalize_mac("not-a-mac")
|
|
|
|
|
|
class ComparisonTests(unittest.TestCase):
|
|
def test_new_modified_disappeared(self):
|
|
previous = [
|
|
Host("10.0.0.1", hostname="srv", ports=[PortInfo(22, service="SSH")]),
|
|
Host("10.0.0.2", hostname="old"),
|
|
]
|
|
current = [
|
|
Host("10.0.0.1", hostname="srv", ports=[PortInfo(22, service="SSH"), PortInfo(80, service="HTTP")]),
|
|
Host("10.0.0.3", hostname="new"),
|
|
]
|
|
result = compare_hosts(current, previous)
|
|
by_ip = {h.ip: h for h in result}
|
|
self.assertEqual(by_ip["10.0.0.1"].change_status, "Modifié")
|
|
self.assertIn("80/tcp", by_ip["10.0.0.1"].change_detail)
|
|
self.assertEqual(by_ip["10.0.0.3"].change_status, "Nouveau")
|
|
self.assertEqual(by_ip["10.0.0.2"].change_status, "Disparu")
|
|
self.assertEqual(by_ip["10.0.0.2"].status, "down")
|
|
|
|
def test_ip_move_by_mac(self):
|
|
previous = [Host("10.0.0.10", mac="AA:BB:CC:DD:EE:FF")]
|
|
current = [Host("10.0.0.20", mac="AA:BB:CC:DD:EE:FF")]
|
|
result = compare_hosts(current, previous)
|
|
self.assertEqual(len(result), 1)
|
|
self.assertEqual(result[0].change_status, "IP modifiée")
|
|
self.assertEqual(result[0].previous_ip, "10.0.0.10")
|
|
|
|
def test_first_scan_is_reference(self):
|
|
current = [Host("10.0.0.1")]
|
|
result = compare_hosts(current, None)
|
|
self.assertEqual(result[0].change_status, "")
|
|
|
|
|
|
class StorageTests(unittest.TestCase):
|
|
def test_history_roundtrip(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
store = HistoryStore(Path(tmp) / "history.sqlite3")
|
|
host = Host(
|
|
"192.168.1.2",
|
|
hostname="pve01.local",
|
|
mac="AA:BB:CC:DD:EE:01",
|
|
ports=[PortInfo(22, service="SSH"), PortInfo(8006, service="Proxmox VE")],
|
|
)
|
|
scan_id = store.save_scan("192.168.1.0/24", "Standard", [host], "standard-test-v2")
|
|
latest = store.latest_scan("192.168.1.0/24", "Standard", "standard-test-v2")
|
|
self.assertIsNotNone(latest)
|
|
self.assertEqual(int(latest["id"]), scan_id)
|
|
loaded = store.load_scan_hosts(scan_id)
|
|
self.assertEqual(len(loaded), 1)
|
|
self.assertEqual(loaded[0].hostname, "pve01.local")
|
|
self.assertEqual(loaded[0].device_type, "Hyperviseur Proxmox")
|
|
|
|
def test_v01_database_is_migrated_without_becoming_v02_baseline(self):
|
|
import sqlite3
|
|
from contextlib import closing
|
|
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
db = Path(tmp) / "history.sqlite3"
|
|
with closing(sqlite3.connect(db)) as conn:
|
|
conn.executescript(
|
|
"""
|
|
CREATE TABLE scans (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
created_at TEXT NOT NULL,
|
|
target TEXT NOT NULL,
|
|
profile TEXT NOT NULL,
|
|
host_count INTEGER NOT NULL
|
|
);
|
|
CREATE TABLE scan_hosts (
|
|
scan_id INTEGER NOT NULL,
|
|
ip TEXT NOT NULL,
|
|
hostname TEXT,
|
|
mac TEXT,
|
|
vendor TEXT,
|
|
os_name TEXT,
|
|
ports_json TEXT NOT NULL
|
|
);
|
|
INSERT INTO scans(created_at, target, profile, host_count)
|
|
VALUES ('2026-08-21T12:00:00+02:00', '192.168.5.0/24', 'Standard', 1);
|
|
INSERT INTO scan_hosts(scan_id, ip, hostname, mac, vendor, os_name, ports_json)
|
|
VALUES (1, '192.168.5.1', 'pve01.local', '', '', '', '[]');
|
|
"""
|
|
)
|
|
store = HistoryStore(db)
|
|
self.assertIsNone(store.latest_scan("192.168.5.0/24", "Standard", "standard-v2:test"))
|
|
self.assertEqual(len(store.recent_scans()), 1)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|