Trixy7/plugins/fritzbox/tests/test_fritzbox_plugin.py

# -*- 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()