Prometheus exporter for HomeHub 4000 router

tests/conftest.py

@@ -0,0 +1,16 @@
+import tempfile
+from pathlib import Path
+
+import pytest
+
+from database import VirginMonitorDatabase
+
+
+@pytest.fixture
+def temp_db():
+    """Create a temporary database for testing."""
+    with tempfile.NamedTemporaryFile(suffix=".db", delete=False) as f:
+        db_path = f.name
+    db = VirginMonitorDatabase(db_path)
+    yield db
+    Path(db_path).unlink(missing_ok=True)

tests/test_database.py

@@ -0,0 +1,63 @@
+from datetime import datetime
+
+
+class TestSaveRouterLogs:
+    def test_inserts_new_entries(self, temp_db):
+        logs = [
+            {
+                "log_timestamp": "2024-01-15T10:30:00",
+                "level": "INF",
+                "module": "WIFI",
+                "message": "Device connected",
+            }
+        ]
+        inserted = temp_db.save_router_logs(logs)
+        assert inserted == 1
+
+    def test_deduplicates_entries(self, temp_db):
+        logs = [
+            {
+                "log_timestamp": "2024-01-15T10:30:00",
+                "level": "INF",
+                "module": "WIFI",
+                "message": "Device connected",
+            }
+        ]
+        temp_db.save_router_logs(logs)
+        inserted = temp_db.save_router_logs(logs)
+        assert inserted == 0
+
+    def test_empty_list_returns_zero(self, temp_db):
+        assert temp_db.save_router_logs([]) == 0
+
+
+class TestGetLogCounts:
+    def test_empty_database(self, temp_db):
+        counts = temp_db.get_log_counts()
+        assert counts["total"] == 0
+        assert counts["by_level"] == {}
+        assert counts["by_module"] == {}
+
+    def test_with_data(self, temp_db):
+        logs = [
+            {"log_timestamp": datetime.now().isoformat(), "level": "INF", "module": "WIFI", "message": "msg1"},
+            {"log_timestamp": datetime.now().isoformat(), "level": "INF", "module": "WIFI", "message": "msg2"},
+            {"log_timestamp": datetime.now().isoformat(), "level": "ERR", "module": "SYS", "message": "msg3"},
+        ]
+        temp_db.save_router_logs(logs)
+        counts = temp_db.get_log_counts()
+        assert counts["total"] == 3
+        assert counts["by_level"]["INF"] == 2
+        assert counts["by_level"]["ERR"] == 1
+        assert counts["by_module"]["WIFI"] == 2
+
+
+class TestGetDeviceLogRows:
+    def test_returns_logs_for_parsing(self, temp_db):
+        logs = [
+            {"log_timestamp": "2024-01-15T10:30:00", "level": "INF", "module": "WIFI", "message": "test message"},
+        ]
+        temp_db.save_router_logs(logs)
+        rows = temp_db.get_device_log_rows()
+        assert len(rows) == 1
+        assert rows[0]["message"] == "test message"

tests/test_parsing_utils.py

@@ -0,0 +1,32 @@
+from router_scraper import RouterScraper
+
+
+class TestParseInt:
+    def test_simple_number(self):
+        assert RouterScraper._parse_int("123") == 123
+
+    def test_with_commas(self):
+        assert RouterScraper._parse_int("1,234,567") == 1234567
+
+    def test_with_whitespace(self):
+        assert RouterScraper._parse_int("  456  ") == 456
+
+    def test_invalid_returns_zero(self):
+        assert RouterScraper._parse_int("not a number") == 0
+        assert RouterScraper._parse_int("") == 0
+
+
+class TestNormalizeInterfaceName:
+    def test_lan_ports(self):
+        scraper = RouterScraper.__new__(RouterScraper)
+        assert scraper._normalize_interface_name("LAN 1") == "lan_1"
+        assert scraper._normalize_interface_name("lan 2") == "lan_2"
+
+    def test_wifi_bands(self):
+        scraper = RouterScraper.__new__(RouterScraper)
+        assert scraper._normalize_interface_name("2.4 GHz") == "wifi_2.4ghz"
+        assert scraper._normalize_interface_name("5.0 GHz Radio 1") == "wifi_5ghz_radio1"
+
+    def test_unknown_returns_none(self):
+        scraper = RouterScraper.__new__(RouterScraper)
+        assert scraper._normalize_interface_name("Unknown Interface") is None

tests/test_router_status.py

@@ -0,0 +1,73 @@
+from datetime import datetime
+
+from router_status import parse_timestamp, parse_events, build_status
+
+
+class TestParseTimestamp:
+    def test_iso_format(self):
+        result = parse_timestamp("2024-01-15T10:30:00")
+        assert result == datetime(2024, 1, 15, 10, 30, 0)
+
+    def test_strptime_format(self):
+        result = parse_timestamp("2024-01-15 10:30:00")
+        assert result == datetime(2024, 1, 15, 10, 30, 0)
+
+    def test_invalid_returns_none(self):
+        assert parse_timestamp("not a date") is None
+        assert parse_timestamp("") is None
+        assert parse_timestamp(None) is None
+
+
+class TestParseEvents:
+    def test_extracts_connected_device(self):
+        rows = [
+            {
+                "log_timestamp": "2024-01-15T10:30:00",
+                "message": "<MyPhone [AA:BB:CC:DD:EE:FF]> has successfully connected to WiFi",
+            }
+        ]
+        events = list(parse_events(rows))
+        assert len(events) == 1
+        assert events[0]["name"] == "MyPhone"
+        assert events[0]["mac"] == "AA:BB:CC:DD:EE:FF"
+        assert events[0]["event"] == "connected"
+
+    def test_extracts_disconnected_device(self):
+        rows = [
+            {
+                "log_timestamp": "2024-01-15T10:30:00",
+                "message": "<Laptop [11:22:33:44:55:66]> was disconnected from WiFi",
+            }
+        ]
+        events = list(parse_events(rows))
+        assert len(events) == 1
+        assert events[0]["event"] == "disconnected"
+
+    def test_skips_missing_timestamp(self):
+        rows = [{"message": "<Device [AA:BB:CC:DD:EE:FF]> has successfully connected"}]
+        events = list(parse_events(rows))
+        assert len(events) == 0
+
+    def test_allows_missing_timestamp_when_flagged(self):
+        rows = [{"message": "<Device [AA:BB:CC:DD:EE:FF]> has successfully connected"}]
+        events = list(parse_events(rows, allow_missing_timestamp=True))
+        assert len(events) == 1
+
+
+class TestBuildStatus:
+    def test_tracks_connection(self):
+        events = [
+            {"name": "Phone", "mac": "AA:BB:CC:DD:EE:FF", "event": "connected", "time": datetime(2024, 1, 15, 10, 0)},
+        ]
+        status = build_status(events)
+        assert "AA:BB:CC:DD:EE:FF" in status
+        assert status["AA:BB:CC:DD:EE:FF"].is_connected is True
+
+    def test_tracks_disconnection(self):
+        events = [
+            {"name": "Phone", "mac": "AA:BB:CC:DD:EE:FF", "event": "connected", "time": datetime(2024, 1, 15, 10, 0)},
+            {"name": "Phone", "mac": "AA:BB:CC:DD:EE:FF", "event": "disconnected", "time": datetime(2024, 1, 15, 11, 0)},
+        ]
+        status = build_status(events)
+        assert status["AA:BB:CC:DD:EE:FF"].is_connected is False
+        assert status["AA:BB:CC:DD:EE:FF"].last_connected_time == datetime(2024, 1, 15, 10, 0)