# -*- coding: utf-8 -*- """ Tests fuer das Fritzbox Plugin. Testet Monitor-Klasse, Intent-Handler und Monitoring-Logik mit gemockter fritzconnection-Bibliothek. """ import asyncio import time from datetime import datetime from pathlib import Path from unittest.mock import AsyncMock, MagicMock, patch import pytest from trixy_core.nlp.handler import IntentReceivedData, IntentResult from plugins.fritzbox.monitor import ( FritzboxMonitor, TrafficSnapshot, NetworkDevice, ) from plugins.fritzbox.main import FritzboxPlugin # ============================================================================ # Fixtures # ============================================================================ def _make_traffic(upload_mbps: float = 1.0, download_mbps: float = 10.0) -> TrafficSnapshot: """Erstellt ein TrafficSnapshot mit Mbit/s Werten.""" return TrafficSnapshot( timestamp=datetime.now(), upload_bytes_sec=upload_mbps * 1_000_000 / 8, download_bytes_sec=download_mbps * 1_000_000 / 8, ) def _make_devices(count: int = 3) -> list[NetworkDevice]: """Erstellt eine Liste von Netzwerkgeraeten.""" return [ NetworkDevice( name=f"Device-{i}", ip=f"192.168.178.{10 + i}", mac=f"AA:BB:CC:DD:EE:{i:02X}", active=True, interface="LAN", ) for i in range(count) ] def _make_plugin(tmp_path, config_overrides=None) -> tuple[FritzboxPlugin, MagicMock]: """Erstellt ein FritzboxPlugin mit Mock-Application.""" mock_app = MagicMock() mock_app.events = MagicMock() mock_app.events.emit = AsyncMock() mock_app.events.trigger = AsyncMock() mock_app.events.register = MagicMock() mock_app.debug = True mock_app.services = MagicMock() mock_app.services.get_service = MagicMock(return_value=None) # Satellites fuer Benachrichtigungen mock_app.satellites = MagicMock() config = { "enabled": True, "fritzbox": { "host": "192.168.178.1", "port": 49000, "username": "admin", "password": "secret", "use_tls": False, }, "check_interval_seconds": 30, "known_devices": [], "alerts": { "high_upload": {"enabled": True, "threshold_mbit": 5.0, "message": "Hoher Upload: {rate} Mbit/s"}, "high_download": {"enabled": True, "threshold_mbit": 50.0, "message": "Hoher Download: {rate} Mbit/s"}, "new_device": {"enabled": True, "message": "Neues Geraet: {name} ({ip})"}, "device_offline": {"enabled": False, "watch_devices": [], "message": "Geraet offline: {name}"}, }, "notify_satellites": {"mode": "all", "aliases": [], "rooms": []}, "notification_sound": "notification.wav", } if config_overrides: config.update(config_overrides) plugin_path = tmp_path / "fritzbox" plugin_path.mkdir(exist_ok=True) plugin = FritzboxPlugin( application=mock_app, plugin_path=plugin_path, config=config, ) return plugin, mock_app def _make_data(**kwargs) -> IntentReceivedData: """Erstellt IntentReceivedData mit Defaults.""" defaults = { "intent": "fritzbox_status", "satellite_id": "sat-1", "room_id": "wohnzimmer", "is_authenticated": True, } defaults.update(kwargs) return IntentReceivedData(**defaults) # ============================================================================ # Tests: TrafficSnapshot # ============================================================================ class TestTrafficSnapshot: """Tests fuer TrafficSnapshot Datenklasse.""" def test_upload_mbit_conversion(self): """upload_mbit rechnet Bytes/s korrekt in Mbit/s um.""" snapshot = TrafficSnapshot( timestamp=datetime.now(), upload_bytes_sec=125000, # 1 Mbit/s download_bytes_sec=0, ) assert abs(snapshot.upload_mbit - 1.0) < 0.01 def test_download_mbit_conversion(self): """download_mbit rechnet Bytes/s korrekt in Mbit/s um.""" snapshot = TrafficSnapshot( timestamp=datetime.now(), upload_bytes_sec=0, download_bytes_sec=6250000, # 50 Mbit/s ) assert abs(snapshot.download_mbit - 50.0) < 0.01 def test_zero_traffic(self): """Null-Traffic gibt 0 Mbit/s zurueck.""" snapshot = TrafficSnapshot(timestamp=datetime.now()) assert snapshot.upload_mbit == 0.0 assert snapshot.download_mbit == 0.0 # ============================================================================ # Tests: NetworkDevice # ============================================================================ class TestNetworkDevice: """Tests fuer NetworkDevice Datenklasse.""" def test_creation(self): """NetworkDevice mit allen Feldern erstellen.""" dev = NetworkDevice( name="iPhone", ip="192.168.178.42", mac="AA:BB:CC:DD:EE:FF", active=True, interface="WLAN", ) assert dev.name == "iPhone" assert dev.active is True def test_default_interface(self): """NetworkDevice hat leeren Interface-Default.""" dev = NetworkDevice(name="Test", ip="1.2.3.4", mac="00:00:00:00:00:00", active=False) assert dev.interface == "" # ============================================================================ # Tests: FritzboxMonitor # ============================================================================ class TestFritzboxMonitor: """Tests fuer FritzboxMonitor (mit gemockter fritzconnection).""" def test_is_available_without_lib(self): """is_available gibt False wenn fritzconnection fehlt.""" with patch("plugins.fritzbox.monitor._HAS_FRITZ", False): monitor = FritzboxMonitor("192.168.178.1", 49000, "", "pass") assert monitor.is_available is False def test_is_available_with_lib(self): """is_available gibt True wenn fritzconnection vorhanden.""" with patch("plugins.fritzbox.monitor._HAS_FRITZ", True): monitor = FritzboxMonitor("192.168.178.1", 49000, "", "pass") assert monitor.is_available is True def test_initial_state(self): """Monitor ist initial nicht verbunden.""" monitor = FritzboxMonitor("192.168.178.1", 49000, "user", "pass") assert monitor.is_connected is False def test_connect_without_lib(self): """connect() gibt False wenn fritzconnection fehlt.""" with patch("plugins.fritzbox.monitor._HAS_FRITZ", False): monitor = FritzboxMonitor("192.168.178.1", 49000, "", "pass") assert monitor.connect() is False @patch("plugins.fritzbox.monitor._HAS_FRITZ", True) @patch("plugins.fritzbox.monitor.FritzConnection") @patch("plugins.fritzbox.monitor.FritzStatus") @patch("plugins.fritzbox.monitor.FritzHosts") def test_connect_success(self, mock_hosts, mock_status, mock_conn): """connect() setzt is_connected bei Erfolg.""" monitor = FritzboxMonitor("192.168.178.1", 49000, "user", "pass") assert monitor.connect() is True assert monitor.is_connected is True @patch("plugins.fritzbox.monitor._HAS_FRITZ", True) @patch("plugins.fritzbox.monitor.FritzConnection", side_effect=Exception("Timeout")) def test_connect_failure(self, mock_conn): """connect() gibt False bei Verbindungsfehler.""" monitor = FritzboxMonitor("192.168.178.1", 49000, "user", "pass") assert monitor.connect() is False assert monitor.is_connected is False def test_disconnect(self): """disconnect() setzt alles zurueck.""" monitor = FritzboxMonitor("192.168.178.1", 49000, "user", "pass") monitor._connected = True monitor._connection = MagicMock() monitor.disconnect() assert monitor.is_connected is False assert monitor._connection is None def test_get_traffic_not_connected(self): """get_traffic() ohne Verbindung gibt None.""" monitor = FritzboxMonitor("192.168.178.1", 49000, "user", "pass") assert monitor.get_traffic() is None def test_get_traffic_success(self): """get_traffic() liefert TrafficSnapshot.""" monitor = FritzboxMonitor("192.168.178.1", 49000, "user", "pass") mock_status = MagicMock() mock_status.transmission_rate = (125000, 6250000) # 1/50 Mbit monitor._status = mock_status result = monitor.get_traffic() assert result is not None assert abs(result.upload_mbit - 1.0) < 0.01 assert abs(result.download_mbit - 50.0) < 0.01 def test_get_traffic_exception(self): """get_traffic() gibt None bei Fehler.""" monitor = FritzboxMonitor("192.168.178.1", 49000, "user", "pass") mock_status = MagicMock() mock_status.transmission_rate = property(lambda self: (_ for _ in ()).throw(Exception("Fehler"))) type(mock_status).transmission_rate = property(lambda self: (_ for _ in ()).throw(Exception("Fehler"))) monitor._status = mock_status result = monitor.get_traffic() assert result is None def test_get_devices_not_connected(self): """get_devices() ohne Hosts gibt leere Liste.""" monitor = FritzboxMonitor("192.168.178.1", 49000, "user", "pass") assert monitor.get_devices() == [] def test_get_devices_success(self): """get_devices() liefert Geraete-Liste.""" monitor = FritzboxMonitor("192.168.178.1", 49000, "user", "pass") mock_hosts = MagicMock() mock_hosts.host_numbers = 2 mock_hosts.get_generic_host_entry = MagicMock(side_effect=[ {"NewHostName": "PC", "NewIPAddress": "192.168.178.10", "NewMACAddress": "AA:BB:CC:DD:EE:01", "NewActive": True, "NewInterfaceType": "LAN"}, {"NewHostName": "Phone", "NewIPAddress": "192.168.178.11", "NewMACAddress": "AA:BB:CC:DD:EE:02", "NewActive": False, "NewInterfaceType": "WLAN"}, ]) monitor._hosts = mock_hosts devices = monitor.get_devices() assert len(devices) == 2 assert devices[0].name == "PC" assert devices[0].active is True assert devices[1].active is False def test_get_active_devices(self): """get_active_devices() filtert inaktive Geraete.""" monitor = FritzboxMonitor("192.168.178.1", 49000, "user", "pass") mock_hosts = MagicMock() mock_hosts.host_numbers = 3 mock_hosts.get_generic_host_entry = MagicMock(side_effect=[ {"NewHostName": "A", "NewIPAddress": "1", "NewMACAddress": "M1", "NewActive": True}, {"NewHostName": "B", "NewIPAddress": "2", "NewMACAddress": "M2", "NewActive": False}, {"NewHostName": "C", "NewIPAddress": "3", "NewMACAddress": "M3", "NewActive": True}, ]) monitor._hosts = mock_hosts active = monitor.get_active_devices() assert len(active) == 2 # ============================================================================ # Tests: FritzboxPlugin — Intent-Handler # ============================================================================ class TestFritzboxIntents: """Tests fuer die 3 Fritzbox-Intent-Handler.""" @pytest.mark.asyncio async def test_handle_status_not_connected(self, tmp_path): """fritzbox_status bei nicht verbundener Fritzbox.""" plugin, _ = _make_plugin(tmp_path) plugin._monitor = None data = _make_data(intent="fritzbox_status") result = await plugin.handle_status(data) assert result.success is False assert "nicht verbunden" in result.response_text @pytest.mark.asyncio async def test_handle_status_connected(self, tmp_path): """fritzbox_status mit Traffic und Geraeten.""" plugin, _ = _make_plugin(tmp_path) mock_monitor = MagicMock() mock_monitor.is_connected = True mock_monitor.get_traffic = MagicMock(return_value=_make_traffic(2.5, 25.0)) mock_monitor.get_active_devices = MagicMock(return_value=_make_devices(5)) plugin._monitor = mock_monitor data = _make_data(intent="fritzbox_status") result = await plugin.handle_status(data) assert result.success is True assert "25.0" in result.response_text # Download assert "2.5" in result.response_text # Upload assert "5 Geraete" in result.response_text @pytest.mark.asyncio async def test_handle_status_no_traffic(self, tmp_path): """fritzbox_status ohne Traffic-Daten zeigt nur Geraete.""" plugin, _ = _make_plugin(tmp_path) mock_monitor = MagicMock() mock_monitor.is_connected = True mock_monitor.get_traffic = MagicMock(return_value=None) mock_monitor.get_active_devices = MagicMock(return_value=_make_devices(2)) plugin._monitor = mock_monitor data = _make_data(intent="fritzbox_status") result = await plugin.handle_status(data) assert result.success is True assert "2 Geraete" in result.response_text @pytest.mark.asyncio async def test_handle_devices_not_connected(self, tmp_path): """fritzbox_devices bei nicht verbundener Fritzbox.""" plugin, _ = _make_plugin(tmp_path) plugin._monitor = MagicMock() plugin._monitor.is_connected = False data = _make_data(intent="fritzbox_devices") result = await plugin.handle_devices(data) assert result.success is False @pytest.mark.asyncio async def test_handle_devices_with_list(self, tmp_path): """fritzbox_devices listet Geraete auf.""" plugin, _ = _make_plugin(tmp_path) mock_monitor = MagicMock() mock_monitor.is_connected = True mock_monitor.get_active_devices = MagicMock(return_value=_make_devices(3)) plugin._monitor = mock_monitor data = _make_data(intent="fritzbox_devices") result = await plugin.handle_devices(data) assert result.success is True assert "3 Geraete" in result.response_text assert "Device-0" in result.response_text @pytest.mark.asyncio async def test_handle_devices_empty(self, tmp_path): """fritzbox_devices ohne Geraete.""" plugin, _ = _make_plugin(tmp_path) mock_monitor = MagicMock() mock_monitor.is_connected = True mock_monitor.get_active_devices = MagicMock(return_value=[]) plugin._monitor = mock_monitor data = _make_data(intent="fritzbox_devices") result = await plugin.handle_devices(data) assert result.success is True assert "keine" in result.response_text.lower() @pytest.mark.asyncio async def test_handle_devices_more_than_10(self, tmp_path): """fritzbox_devices mit mehr als 10 Geraeten zeigt 'und X weitere'.""" plugin, _ = _make_plugin(tmp_path) mock_monitor = MagicMock() mock_monitor.is_connected = True mock_monitor.get_active_devices = MagicMock(return_value=_make_devices(15)) plugin._monitor = mock_monitor data = _make_data(intent="fritzbox_devices") result = await plugin.handle_devices(data) assert result.success is True assert "15 Geraete" in result.response_text assert "5 weitere" in result.response_text @pytest.mark.asyncio async def test_handle_traffic_not_connected(self, tmp_path): """fritzbox_traffic bei nicht verbundener Fritzbox.""" plugin, _ = _make_plugin(tmp_path) plugin._monitor = None data = _make_data(intent="fritzbox_traffic") result = await plugin.handle_traffic(data) assert result.success is False @pytest.mark.asyncio async def test_handle_traffic_success(self, tmp_path): """fritzbox_traffic zeigt Up-/Download-Raten.""" plugin, _ = _make_plugin(tmp_path) mock_monitor = MagicMock() mock_monitor.is_connected = True mock_monitor.get_traffic = MagicMock(return_value=_make_traffic(3.2, 45.7)) plugin._monitor = mock_monitor data = _make_data(intent="fritzbox_traffic") result = await plugin.handle_traffic(data) assert result.success is True assert "45.7" in result.response_text assert "3.2" in result.response_text @pytest.mark.asyncio async def test_handle_traffic_no_data(self, tmp_path): """fritzbox_traffic ohne Traffic-Daten.""" plugin, _ = _make_plugin(tmp_path) mock_monitor = MagicMock() mock_monitor.is_connected = True mock_monitor.get_traffic = MagicMock(return_value=None) plugin._monitor = mock_monitor data = _make_data(intent="fritzbox_traffic") result = await plugin.handle_traffic(data) assert result.success is False assert "keine" in result.response_text.lower() or "konnte" in result.response_text.lower() # ============================================================================ # Tests: Monitoring-Logik # ============================================================================ class TestMonitoringLogic: """Tests fuer die internen Monitoring-Methoden.""" @pytest.mark.asyncio async def test_check_traffic_upload_alert(self, tmp_path): """_check_traffic loest Upload-Warnung bei Ueberschreitung aus.""" plugin, _ = _make_plugin(tmp_path) plugin._notify = AsyncMock() plugin._last_upload_alert = 0.0 alerts = plugin.config["alerts"] traffic = _make_traffic(upload_mbps=10.0, download_mbps=1.0) await plugin._check_traffic(traffic, alerts) plugin._notify.assert_called_once() call_text = plugin._notify.call_args[0][0] assert "10.0" in call_text @pytest.mark.asyncio async def test_check_traffic_download_alert(self, tmp_path): """_check_traffic loest Download-Warnung bei Ueberschreitung aus.""" plugin, _ = _make_plugin(tmp_path) plugin._notify = AsyncMock() plugin._last_download_alert = 0.0 alerts = plugin.config["alerts"] traffic = _make_traffic(upload_mbps=0.1, download_mbps=100.0) await plugin._check_traffic(traffic, alerts) plugin._notify.assert_called_once() call_text = plugin._notify.call_args[0][0] assert "100.0" in call_text @pytest.mark.asyncio async def test_check_traffic_no_alert_below_threshold(self, tmp_path): """_check_traffic loest keine Warnung unter Schwellwert aus.""" plugin, _ = _make_plugin(tmp_path) plugin._notify = AsyncMock() alerts = plugin.config["alerts"] traffic = _make_traffic(upload_mbps=1.0, download_mbps=10.0) await plugin._check_traffic(traffic, alerts) plugin._notify.assert_not_called() @pytest.mark.asyncio async def test_check_traffic_cooldown(self, tmp_path): """_check_traffic respektiert Cooldown zwischen Warnungen.""" plugin, _ = _make_plugin(tmp_path) plugin._notify = AsyncMock() plugin._last_upload_alert = time.monotonic() # Gerade erst gewarnt alerts = plugin.config["alerts"] traffic = _make_traffic(upload_mbps=10.0, download_mbps=1.0) await plugin._check_traffic(traffic, alerts) plugin._notify.assert_not_called() @pytest.mark.asyncio async def test_check_devices_new_device(self, tmp_path): """_check_devices erkennt neues Geraet.""" plugin, _ = _make_plugin(tmp_path) plugin._notify = AsyncMock() plugin.save_config = MagicMock() plugin._known_macs = {"AA:BB:CC:DD:EE:00"} devices = _make_devices(2) alerts = plugin.config["alerts"] await plugin._check_devices(devices, alerts) # Device-0 ist bekannt (MAC AA:BB:CC:DD:EE:00), Device-1 ist neu assert plugin._notify.call_count == 1 assert "AA:BB:CC:DD:EE:01" in plugin._known_macs @pytest.mark.asyncio async def test_check_devices_all_known(self, tmp_path): """_check_devices bei nur bekannten Geraeten.""" plugin, _ = _make_plugin(tmp_path) plugin._notify = AsyncMock() devices = _make_devices(2) plugin._known_macs = {d.mac.upper() for d in devices} alerts = plugin.config["alerts"] await plugin._check_devices(devices, alerts) plugin._notify.assert_not_called() @pytest.mark.asyncio async def test_check_devices_offline_detection(self, tmp_path): """_check_devices erkennt offline-gegangenes Geraet.""" config_overrides = { "alerts": { "high_upload": {"enabled": False}, "high_download": {"enabled": False}, "new_device": {"enabled": False}, "device_offline": { "enabled": True, "watch_devices": ["AA:BB:CC:DD:EE:FF"], "message": "Geraet offline: {name}", }, }, } plugin, _ = _make_plugin(tmp_path, config_overrides=config_overrides) plugin._notify = AsyncMock() # Leere Geraete-Liste → AA:BB:CC:DD:EE:FF ist offline devices: list[NetworkDevice] = [] alerts = plugin.config["alerts"] await plugin._check_devices(devices, alerts) plugin._notify.assert_called_once() @pytest.mark.asyncio async def test_run_checks_not_connected(self, tmp_path): """_run_checks tut nichts wenn Monitor nicht verbunden.""" plugin, _ = _make_plugin(tmp_path) plugin._monitor = MagicMock() plugin._monitor.is_connected = False plugin._check_traffic = AsyncMock() plugin._check_devices = AsyncMock() await plugin._run_checks() plugin._check_traffic.assert_not_called() plugin._check_devices.assert_not_called() # ============================================================================ # Tests: Plugin Lifecycle # ============================================================================ class TestPluginLifecycle: """Tests fuer on_load und on_unload.""" @pytest.mark.asyncio async def test_on_load_no_password(self, tmp_path): """on_load ohne Passwort laedt keinen Monitor.""" config = { "enabled": True, "fritzbox": {"host": "192.168.178.1", "password": ""}, } plugin, _ = _make_plugin(tmp_path, config_overrides=config) await plugin.on_load() assert plugin._monitor is None @pytest.mark.asyncio async def test_on_unload_without_task(self, tmp_path): """on_unload ohne laufenden Task ist sicher.""" plugin, _ = _make_plugin(tmp_path) plugin._check_task = None plugin._monitor = None await plugin.on_unload() # Kein Fehler @pytest.mark.asyncio async def test_on_unload_cancels_task(self, tmp_path): """on_unload bricht laufenden Check-Task ab.""" plugin, _ = _make_plugin(tmp_path) async def fake_loop(): await asyncio.sleep(999) plugin._check_task = asyncio.create_task(fake_loop()) plugin._monitor = MagicMock() await plugin.on_unload() assert plugin._check_task.cancelled() or plugin._check_task.done() plugin._monitor.disconnect.assert_called_once()