feat: WAN-Status, DSL-Stats, Portfreigaben, Firmware-Warnings, PPPoE-Interface

- get_system_config(): DSL-Leitungsqualität, Firmware-Status, DNS, NTP - get_device_warnings(): dsl_snr_low + firmware_update_available - get_port_forwards(): IPv6-Pinholes via WANIPv6Firewall1 (TCP6/UDP6) - get_wan_status(): robuste Service-Iteration, kein Crash bei "Auto"-Bitraten - get_interfaces(): pppoe0-Interface, Service-Guards für LAN/WAN, "connected"-Status - _to_float(): Helper für sichere TR-064 Feldkonvertierung - get_device_warnings() implementiert (war als Stub vorhanden) - Tests für alle neuen Methoden, PPPoE-Fixture Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-06-24 22:21:06 +02:00
parent 8eaf277504
commit 3b3469daa9
2 changed files with 557 additions and 76 deletions
@@ -62,6 +62,14 @@ from napalm_device_types.models import (
 _OPTIONAL_SERVICE_ERRORS = (FritzServiceError, FritzActionError, FritzConnectionException)


+def _to_float(value: Any, default: float = 0.0) -> float:
+    """Convert a TR-064 field to float, returning *default* for non-numeric values like 'Auto'."""
+    try:
+        return float(value or 0)
+    except (TypeError, ValueError):
+        return default
+
+
 class FritzBoxDriver(ResidentialGatewayDriver):
    """NAPALM driver for AVM FritzBox (read-only, TR-064)."""

@@ -226,47 +234,65 @@ class FritzBoxDriver(ResidentialGatewayDriver):
        ``WLANConfiguration{1,2,3}.GetInfo``.
        """
        interfaces: dict[str, dict[str, Any]] = {}
+        available = self.fc.services if self.fc else {}

+        if "LANEthernetInterfaceConfig1" in available:
            try:
                lan = self._call("LANEthernetInterfaceConfig1", "GetInfo")
                interfaces["lan"] = {
-                "is_up": lan.get("NewStatus", "") == "Up",
+                    "is_up": lan.get("NewStatus", "").lower() in ("up", "connected"),
                    "is_enabled": bool(lan.get("NewEnable", False)),
                    "description": "LAN",
                    "last_flapped": -1.0,
                    "mac_address": (lan.get("NewMACAddress") or "").lower(),
-                "speed": float(lan.get("NewMaxBitRate", 0) or 0),
+                    "speed": _to_float(lan.get("NewMaxBitRate")),
                    "mtu": 1500,
                }
            except _OPTIONAL_SERVICE_ERRORS:
                pass

+        if "WANCommonInterfaceConfig1" in available:
            try:
                link = self._call("WANCommonInterfaceConfig1", "GetCommonLinkProperties")
                interfaces["wan"] = {
-                "is_up": link.get("NewPhysicalLinkStatus", "") == "Up",
+                    "is_up": link.get("NewPhysicalLinkStatus", "").lower() in ("up", "connected"),
                    "is_enabled": True,
                    "description": link.get("NewWANAccessType", "WAN"),
                    "last_flapped": -1.0,
                    "mac_address": "",
-                "speed": float(link.get("NewLayer1DownstreamMaxBitRate", 0) or 0) / 1000,
+                    "speed": _to_float(link.get("NewLayer1DownstreamMaxBitRate")) / 1000,
                    "mtu": 1500,
                }
            except _OPTIONAL_SERVICE_ERRORS:
                pass

+        if "WANPPPConnection1" in available:
+            try:
+                status = self._call("WANPPPConnection1", "GetStatusInfo")
+                interfaces["pppoe0"] = {
+                    "is_up": status.get("NewConnectionStatus", "").lower() == "connected",
+                    "is_enabled": True,
+                    "description": "PPPoE",
+                    "last_flapped": -1.0,
+                    "mac_address": "",
+                    "speed": 0.0,
+                    "mtu": 1492,
+                }
+            except _OPTIONAL_SERVICE_ERRORS:
+                pass
+
        for service in self._wlan_services():
            try:
                wlan = self._call(service, "GetInfo")
            except _OPTIONAL_SERVICE_ERRORS:
                continue
            interfaces[service.lower()] = {
-                "is_up": wlan.get("NewStatus", "") == "Up",
+                "is_up": wlan.get("NewStatus", "").lower() in ("up", "connected"),
                "is_enabled": bool(wlan.get("NewEnable", False)),
                "description": wlan.get("NewSSID", ""),
                "last_flapped": -1.0,
                "mac_address": (wlan.get("NewBSSID") or "").lower(),
-                "speed": float(wlan.get("NewMaxBitRate", 0) or 0),
+                "speed": _to_float(wlan.get("NewMaxBitRate")),
                "mtu": 1500,
            }

@@ -277,7 +303,9 @@ class FritzBoxDriver(ResidentialGatewayDriver):

        WAN address comes from ``GetExternalIPAddress`` (and, if available,
        ``X_AVM-DE_GetExternalIPv6Address``) on the active WAN connection
-        service. The LAN address is the address used to reach the device
+        service. For PPPoE connections the IP is reported on ``pppoe0`` to
+        match the interface name produced by :meth:`get_interfaces`.
+        The LAN address is the address used to reach the device
        (TR-064 does not expose the router's own LAN IP generically).
        """
        result: dict[str, dict[str, Any]] = {}
@@ -286,16 +314,127 @@ class FritzBoxDriver(ResidentialGatewayDriver):
            service = self._wan_ip_service()
            ext_ip = self._call(service, "GetExternalIPAddress").get("NewExternalIPAddress", "")
            if ext_ip:
-                result["wan"] = {"ipv4": {ext_ip: {"prefix_length": 32}}}
+                wan_key = "pppoe0" if service == "WANPPPConnection1" else "wan"
+                result[wan_key] = {"ipv4": {ext_ip: {"prefix_length": 32}}}
                ipv6 = self._wan_external_ipv6(service)
                if ipv6:
-                    result["wan"]["ipv6"] = {ipv6: {"prefix_length": 64}}
+                    result[wan_key]["ipv6"] = {ipv6: {"prefix_length": 64}}
        except _OPTIONAL_SERVICE_ERRORS:
            pass

        result["lan"] = {"ipv4": {self.hostname: {"prefix_length": 24}}}
        return result

+    def get_system_config(self) -> dict[str, Any]:
+        """Return a system configuration snapshot for storage as ``system_snapshot``.
+
+        Collects DSL line statistics, firmware update status, upstream DNS
+        servers and NTP settings.  All fields are best-effort: missing
+        TR-064 services are silently skipped.
+        """
+        available = self.fc.services if self.fc else {}
+        result: dict[str, Any] = {}
+
+        if "WANDSLInterfaceConfig1" in available:
+            try:
+                info = self._call("WANDSLInterfaceConfig1", "GetInfo")
+                result["dsl_enabled"] = bool(info.get("NewEnable", False))
+                result["dsl_status"] = info.get("NewStatus", "")
+                result["dsl_downstream_rate"] = int(_to_float(info.get("NewDownstreamCurrRate")))
+                result["dsl_upstream_rate"] = int(_to_float(info.get("NewUpstreamCurrRate")))
+                result["dsl_downstream_max_rate"] = int(_to_float(info.get("NewDownstreamMaxRate")))
+                result["dsl_upstream_max_rate"] = int(_to_float(info.get("NewUpstreamMaxRate")))
+                # SNR margins and attenuation are stored as dB×10 by TR-064
+                result["dsl_downstream_noise_margin"] = int(_to_float(info.get("NewDownstreamNoiseMargin")))
+                result["dsl_upstream_noise_margin"] = int(_to_float(info.get("NewUpstreamNoiseMargin")))
+                result["dsl_downstream_attenuation"] = int(_to_float(info.get("NewDownstreamAttenuation")))
+                result["dsl_upstream_attenuation"] = int(_to_float(info.get("NewUpstreamAttenuation")))
+            except _OPTIONAL_SERVICE_ERRORS:
+                pass
+
+            try:
+                stats = self._call("WANDSLInterfaceConfig1", "GetStatisticsTotal")
+                result["dsl_crc_errors"] = int(_to_float(stats.get("NewCRCErrors")))
+                result["dsl_fec_errors"] = int(_to_float(stats.get("NewFECErrors")))
+                result["dsl_hec_errors"] = int(_to_float(stats.get("NewHECErrors")))
+            except _OPTIONAL_SERVICE_ERRORS:
+                pass
+
+        if "UserInterface1" in available:
+            try:
+                ui_info = self._call("UserInterface1", "X_AVM-DE_GetInfo")
+                result["firmware_update_available"] = bool(
+                    ui_info.get("NewX_AVM-DE_UpdateState", "") == "Available"
+                )
+                result["firmware_version"] = ui_info.get("NewX_AVM-DE_Version", "")
+            except _OPTIONAL_SERVICE_ERRORS:
+                try:
+                    ui_info = self._call("UserInterface1", "GetInfo")
+                    result["firmware_update_available"] = bool(
+                        ui_info.get("NewUpgradeAvailable", False)
+                    )
+                except _OPTIONAL_SERVICE_ERRORS:
+                    pass
+
+        if "LANHostConfigManagement1" in available:
+            try:
+                dns_resp = self._call("LANHostConfigManagement1", "GetDNSServers")
+                raw = dns_resp.get("NewDNSServers", "") or ""
+                result["dns_servers"] = [s.strip() for s in raw.split(",") if s.strip()]
+            except _OPTIONAL_SERVICE_ERRORS:
+                pass
+
+        if "Time1" in available:
+            try:
+                time_info = self._call("Time1", "GetInfo")
+                result["ntp_server"] = time_info.get("NewNTPServer1", "")
+                result["ntp_server2"] = time_info.get("NewNTPServer2", "")
+                result["timezone"] = time_info.get("NewCurrentLocalTime", "")
+            except _OPTIONAL_SERVICE_ERRORS:
+                pass
+
+        return result
+
+    def get_device_warnings(self) -> list[dict[str, Any]]:
+        """Return device warnings derived from DSL line quality and firmware state.
+
+        Called during the poll cycle via ``_collect_system_snapshots``.
+        Reads ``UserInterface1`` for firmware updates and
+        ``WANDSLInterfaceConfig1`` for line quality.
+        """
+        warnings: list[dict[str, Any]] = []
+        available = self.fc.services if self.fc else {}
+
+        if "UserInterface1" in available:
+            try:
+                ui_info = self._call("UserInterface1", "X_AVM-DE_GetInfo")
+                if ui_info.get("NewX_AVM-DE_UpdateState", "") == "Available":
+                    warnings.append({
+                        "code": "firmware_update_available",
+                        "severity": "info",
+                        "action": None,
+                        "meta": {"version": ui_info.get("NewX_AVM-DE_Version", "")},
+                    })
+            except _OPTIONAL_SERVICE_ERRORS:
+                pass
+
+        if "WANDSLInterfaceConfig1" in available:
+            try:
+                info = self._call("WANDSLInterfaceConfig1", "GetInfo")
+                # SNR margin is in dB×10; < 60 means < 6 dB — borderline line quality
+                snr_down = int(_to_float(info.get("NewDownstreamNoiseMargin")))
+                if 0 < snr_down < 60:
+                    warnings.append({
+                        "code": "dsl_snr_low",
+                        "severity": "warning",
+                        "action": None,
+                        "meta": {"snr_db_x10": snr_down},
+                    })
+            except _OPTIONAL_SERVICE_ERRORS:
+                pass
+
+        return warnings
+
    def get_arp_table(self, vrf: str = "") -> list[dict[str, Any]]:
        """Return the ARP table, derived from :meth:`get_hosts`.

@@ -323,46 +462,75 @@ class FritzBoxDriver(ResidentialGatewayDriver):
        """Return WAN/internet connection status.

        Calls ``WANCommonInterfaceConfig1.GetCommonLinkProperties`` and
-        ``GetAddonInfos`` for line/traffic stats, plus ``GetStatusInfo`` and
-        ``GetExternalIPAddress`` on the active WAN connection service.
+        ``GetAddonInfos`` for physical link info, then tries each WAN
+        connection service (WANIPConnection1, WANPPPConnection1) to get the
+        logical connection status and external IP.
        """
-        link = self._call("WANCommonInterfaceConfig1", "GetCommonLinkProperties")
+        available = self.fc.services if self.fc else {}

+        link: dict = {}
+        if "WANCommonInterfaceConfig1" in available:
+            try:
+                link = self._call("WANCommonInterfaceConfig1", "GetCommonLinkProperties")
+            except _OPTIONAL_SERVICE_ERRORS:
+                pass
+
+        addon: dict = {}
+        if "WANCommonInterfaceConfig1" in available:
            try:
                addon = self._call("WANCommonInterfaceConfig1", "GetAddonInfos")
            except _OPTIONAL_SERVICE_ERRORS:
-            addon = {}
+                pass

-        service = self._wan_ip_service()
-        status = self._call(service, "GetStatusInfo")
-        ext_ip = self._call(service, "GetExternalIPAddress").get("NewExternalIPAddress", "")
+        # Try each WAN connection service; PPPoE devices may only work with WANPPPConnection1
+        status: dict = {}
+        ext_ip = ""
+        ext_ipv6 = ""
+        wan_service_used = ""
+        for svc in self._WAN_IP_SERVICES:
+            if svc not in available:
+                continue
+            try:
+                status = self._call(svc, "GetStatusInfo")
+                ext_ip = self._call(svc, "GetExternalIPAddress").get("NewExternalIPAddress", "")
+                wan_service_used = svc
+                break
+            except (FritzArrayIndexError, FritzServiceError, FritzActionError, FritzConnectionException):
+                continue
+            except Exception:
+                continue
+
+        if wan_service_used:
+            ext_ipv6 = self._wan_external_ipv6(wan_service_used)

        result: WANStatusDict = {
            "connection_type": link.get("NewWANAccessType", ""),
            "is_connected": status.get("NewConnectionStatus", "") == "Connected",
            "external_ip": ext_ip,
-            "uptime": int(status.get("NewUptime", 0) or 0),
-            "bytes_sent": int(addon.get("NewTotalBytesSent", 0) or 0),
-            "bytes_received": int(addon.get("NewTotalBytesReceived", 0) or 0),
-            "max_bitrate_up": int(link.get("NewLayer1UpstreamMaxBitRate", 0) or 0) // 1000,
-            "max_bitrate_down": int(link.get("NewLayer1DownstreamMaxBitRate", 0) or 0) // 1000,
+            "uptime": int(_to_float(status.get("NewUptime"))),
+            "bytes_sent": int(_to_float(addon.get("NewTotalBytesSent"))),
+            "bytes_received": int(_to_float(addon.get("NewTotalBytesReceived"))),
+            "max_bitrate_up": int(_to_float(link.get("NewLayer1UpstreamMaxBitRate"))) // 1000,
+            "max_bitrate_down": int(_to_float(link.get("NewLayer1DownstreamMaxBitRate"))) // 1000,
            "link_status": link.get("NewPhysicalLinkStatus", ""),
        }

-        ipv6 = self._wan_external_ipv6(service)
-        if ipv6:
-            result["external_ipv6"] = ipv6
+        if ext_ipv6:
+            result["external_ipv6"] = ext_ipv6

        return result

    def get_port_forwards(self) -> list[PortForwardDict]:
-        """Return configured port forwarding rules.
+        """Return configured port forwarding rules (IPv4 + IPv6 pinholes).

        Iterates ``GetGenericPortMappingEntry`` on the active WAN connection
-        service until the device reports an out-of-range index.
+        service until the device reports an out-of-range index.  IPv6 pinholes
+        from ``WANIPv6Firewall1`` are appended with protocol ``TCP6``/``UDP6``.
        """
-        service = self._wan_ip_service()
        forwards: list[PortForwardDict] = []
+
+        try:
+            service = self._wan_ip_service()
            index = 0
            while True:
                try:
@@ -383,6 +551,37 @@ class FritzBoxDriver(ResidentialGatewayDriver):
                    forward["remote_host"] = remote_host
                forwards.append(forward)
                index += 1
+        except _OPTIONAL_SERVICE_ERRORS:
+            pass
+
+        available = self.fc.services if self.fc else {}
+        if "WANIPv6Firewall1" in available:
+            try:
+                fw = self._call("WANIPv6Firewall1", "GetFirewallStatus")
+                if fw.get("NewFirewallEnabled"):
+                    index = 0
+                    while True:
+                        try:
+                            ph = self._call(
+                                "WANIPv6Firewall1",
+                                "GetGenericPinholeEntry",
+                                NewPinholeIndex=index,
+                            )
+                        except (FritzArrayIndexError, FritzConnectionException, FritzActionError):
+                            break
+                        proto_num = int(ph.get("NewProtocol", 0) or 0)
+                        proto = "TCP6" if proto_num == 6 else ("UDP6" if proto_num == 17 else str(proto_num))
+                        forwards.append({
+                            "name": ph.get("NewPinholeDescription", ""),
+                            "protocol": proto,
+                            "external_port": int(ph.get("NewRemotePort", 0) or 0),
+                            "internal_ip": ph.get("NewInternalIPAddress", ""),
+                            "internal_port": int(ph.get("NewInternalPort", 0) or 0),
+                            "enabled": True,
+                        })
+                        index += 1
+            except _OPTIONAL_SERVICE_ERRORS:
+                pass

        return forwards

@@ -481,9 +680,9 @@ class FritzBoxDriver(ResidentialGatewayDriver):
                    "mac": (dev.get("NewAssociatedDeviceMACAddress") or "").lower(),
                    "ssid": ssid,
                    "radio": service,
-                    "signal": int(dev.get("NewX_AVM-DE_SignalStrength", 0) or 0),
+                    "signal": int(_to_float(dev.get("NewX_AVM-DE_SignalStrength"))),
                    "noise": 0,
-                    "tx_rate": float(dev.get("NewX_AVM-DE_Speed", 0) or 0),
+                    "tx_rate": _to_float(dev.get("NewX_AVM-DE_Speed")),
                    "rx_rate": 0.0,
                    "uptime": 0,
                }
@@ -543,7 +742,7 @@ class FritzBoxDriver(ResidentialGatewayDriver):
            except _OPTIONAL_SERVICE_ERRORS:
                continue

-            channel = int(info.get("NewChannel", 0) or 0)
+            channel = int(_to_float(info.get("NewChannel")))
            radios[service] = {
                "enabled": bool(info.get("NewEnable", False)),
                "band": "2.4GHz" if channel <= 14 else "5GHz",
@@ -10,7 +10,7 @@ from fritzconnection.core.exceptions import (
 )
 from napalm.base.exceptions import ConnectionClosedException, ConnectionException

-from napalm_fritzbox.fritzbox import FritzBoxDriver
+from napalm_fritzbox.fritzbox import FritzBoxDriver, _to_float

 # ---------------------------------------------------------------------------
 # Sample TR-064 responses
@@ -112,6 +112,71 @@ WLAN2_INFO = {

 WLAN2_TOTAL_ASSOCIATIONS = {"NewTotalAssociations": 0}

+LAN_INFO_AUTO_RATE = {
+    "NewEnable": True,
+    "NewStatus": "Up",
+    "NewMACAddress": "AA:BB:CC:DD:EE:01",
+    "NewMaxBitRate": "Auto",
+}
+
+PPP_STATUS_INFO = {
+    "NewConnectionStatus": "Connected",
+    "NewLastConnectionError": "ERROR_NONE",
+    "NewUptime": 7200,
+}
+
+PPP_EXTERNAL_IP = {"NewExternalIPAddress": "203.0.113.99"}
+
+DSL_INFO = {
+    "NewEnable": True,
+    "NewStatus": "Up",
+    "NewDownstreamCurrRate": 100000,
+    "NewUpstreamCurrRate": 40000,
+    "NewDownstreamMaxRate": 120000,
+    "NewUpstreamMaxRate": 50000,
+    "NewDownstreamNoiseMargin": 120,  # 12.0 dB
+    "NewUpstreamNoiseMargin": 80,     # 8.0 dB
+    "NewDownstreamAttenuation": 150,  # 15.0 dB
+    "NewUpstreamAttenuation": 90,     # 9.0 dB
+}
+
+DSL_STATS = {
+    "NewCRCErrors": 5,
+    "NewFECErrors": 12,
+    "NewHECErrors": 0,
+}
+
+UI_INFO = {
+    "NewX_AVM-DE_UpdateState": "Available",
+    "NewX_AVM-DE_Version": "08.20",
+}
+
+UI_INFO_NO_UPDATE = {
+    "NewX_AVM-DE_UpdateState": "None",
+    "NewX_AVM-DE_Version": "08.20",
+}
+
+DNS_SERVERS = {"NewDNSServers": "1.1.1.1,8.8.8.8"}
+
+TIME_INFO = {
+    "NewNTPServer1": "ntp.example.com",
+    "NewNTPServer2": "",
+    "NewCurrentLocalTime": "2026-06-24T16:00:00+02:00",
+}
+
+IPV6_FIREWALL_STATUS = {"NewFirewallEnabled": True, "NewInboundPinholeAllowed": True}
+
+IPV6_PINHOLES = [
+    {
+        "NewPinholeDescription": "SSH6",
+        "NewProtocol": 6,
+        "NewRemotePort": 0,
+        "NewInternalIPAddress": "2001:db8::10",
+        "NewInternalPort": 22,
+        "NewLeaseTime": 0,
+    }
+]
+

 # ---------------------------------------------------------------------------
 # call_action dispatch
@@ -124,11 +189,19 @@ _SIMPLE_RESPONSES = {
    ("WANCommonInterfaceConfig1", "GetAddonInfos"): WAN_ADDON_INFOS,
    ("WANIPConnection1", "GetStatusInfo"): WAN_STATUS_INFO,
    ("WANIPConnection1", "GetExternalIPAddress"): WAN_EXTERNAL_IP,
+    ("WANPPPConnection1", "GetStatusInfo"): PPP_STATUS_INFO,
+    ("WANPPPConnection1", "GetExternalIPAddress"): PPP_EXTERNAL_IP,
    ("Hosts1", "GetHostNumberOfEntries"): HOSTS_COUNT,
    ("WLANConfiguration1", "GetInfo"): WLAN1_INFO,
    ("WLANConfiguration1", "GetTotalAssociations"): WLAN1_TOTAL_ASSOCIATIONS,
    ("WLANConfiguration2", "GetInfo"): WLAN2_INFO,
    ("WLANConfiguration2", "GetTotalAssociations"): WLAN2_TOTAL_ASSOCIATIONS,
+    ("WANDSLInterfaceConfig1", "GetInfo"): DSL_INFO,
+    ("WANDSLInterfaceConfig1", "GetStatisticsTotal"): DSL_STATS,
+    ("UserInterface1", "X_AVM-DE_GetInfo"): UI_INFO,
+    ("LANHostConfigManagement1", "GetDNSServers"): DNS_SERVERS,
+    ("Time1", "GetInfo"): TIME_INFO,
+    ("WANIPv6Firewall1", "GetFirewallStatus"): IPV6_FIREWALL_STATUS,
 }


@@ -145,6 +218,12 @@ def _call_action(service, action, **kwargs):
            raise FritzArrayIndexError("index out of range")
        return PORT_FORWARDS[index]

+    if (service, action) == ("WANIPv6Firewall1", "GetGenericPinholeEntry"):
+        index = kwargs["NewPinholeIndex"]
+        if index >= len(IPV6_PINHOLES):
+            raise FritzArrayIndexError("index out of range")
+        return IPV6_PINHOLES[index]
+
    if (service, action) == ("WLANConfiguration1", "GetGenericAssociatedDeviceInfo"):
        index = kwargs["NewAssociatedDeviceIndex"]
        if index >= 1:
@@ -167,7 +246,7 @@ def _call_action(service, action, **kwargs):

@pytest.fixture
 def driver():
-    """Return a driver instance with a mocked FritzConnection."""
+    """Return a driver instance with a mocked FritzConnection (WANIPConnection1 / cable)."""
    mock_fc = MagicMock()
    mock_fc.services = {
        "DeviceInfo1": MagicMock(),
@@ -177,6 +256,31 @@ def driver():
        "Hosts1": MagicMock(),
        "WLANConfiguration1": MagicMock(),
        "WLANConfiguration2": MagicMock(),
+        "WANDSLInterfaceConfig1": MagicMock(),
+        "UserInterface1": MagicMock(),
+        "LANHostConfigManagement1": MagicMock(),
+        "Time1": MagicMock(),
+        "WANIPv6Firewall1": MagicMock(),
+    }
+    mock_fc.call_action.side_effect = _call_action
+
+    drv = FritzBoxDriver(hostname="fritz.box", username="user", password="pass")
+    drv.fc = mock_fc
+    yield drv
+
+
+@pytest.fixture
+def pppoe_driver():
+    """Return a driver instance simulating a PPPoE FritzBox (WANPPPConnection1)."""
+    mock_fc = MagicMock()
+    mock_fc.services = {
+        "DeviceInfo1": MagicMock(),
+        "LANEthernetInterfaceConfig1": MagicMock(),
+        "WANCommonInterfaceConfig1": MagicMock(),
+        "WANPPPConnection1": MagicMock(),
+        "Hosts1": MagicMock(),
+        "WLANConfiguration1": MagicMock(),
+        "WLANConfiguration2": MagicMock(),
    }
    mock_fc.call_action.side_effect = _call_action

@@ -462,3 +566,181 @@ class TestGetRadioStatus:
        assert radio["enabled"] is False
        assert radio["band"] == "2.4GHz"
        assert radio["channel"] == 1
+
+
+# ---------------------------------------------------------------------------
+# _to_float helper
+# ---------------------------------------------------------------------------
+
+
+class TestToFloat:
+    def test_numeric_string(self):
+        assert _to_float("1000") == 1000.0
+
+    def test_auto_string_returns_default(self):
+        assert _to_float("Auto") == 0.0
+
+    def test_none_returns_default(self):
+        assert _to_float(None) == 0.0
+
+    def test_int(self):
+        assert _to_float(250000000) == 250000000.0
+
+    def test_custom_default(self):
+        assert _to_float("bogus", default=-1.0) == -1.0
+
+
+# ---------------------------------------------------------------------------
+# PPPoE interface (WANPPPConnection1)
+# ---------------------------------------------------------------------------
+
+
+class TestPPPoEInterface:
+    def test_pppoe0_present_when_ppp_service_available(self, pppoe_driver):
+        interfaces = pppoe_driver.get_interfaces()
+        assert "pppoe0" in interfaces
+
+    def test_pppoe0_is_up(self, pppoe_driver):
+        assert pppoe_driver.get_interfaces()["pppoe0"]["is_up"] is True
+
+    def test_pppoe0_mtu(self, pppoe_driver):
+        assert pppoe_driver.get_interfaces()["pppoe0"]["mtu"] == 1492
+
+    def test_pppoe0_not_present_without_ppp_service(self, driver):
+        # driver fixture uses WANIPConnection1 only — no PPPoE
+        assert "pppoe0" not in driver.get_interfaces()
+
+    def test_pppoe_ip_on_pppoe0_key(self, pppoe_driver):
+        result = pppoe_driver.get_interfaces_ip()
+        assert "pppoe0" in result
+        assert "203.0.113.99" in result["pppoe0"]["ipv4"]
+
+    def test_cable_ip_on_wan_key(self, driver):
+        result = driver.get_interfaces_ip()
+        assert "wan" in result
+        assert "203.0.113.7" in result["wan"]["ipv4"]
+
+
+# ---------------------------------------------------------------------------
+# Auto/non-numeric MaxBitRate in get_interfaces
+# ---------------------------------------------------------------------------
+
+
+class TestAutoMaxBitRate:
+    def test_lan_speed_zero_when_auto(self, driver):
+        driver.fc.call_action.side_effect = lambda svc, act, **kw: (
+            LAN_INFO_AUTO_RATE if (svc, act) == ("LANEthernetInterfaceConfig1", "GetInfo")
+            else _call_action(svc, act, **kw)
+        )
+        interfaces = driver.get_interfaces()
+        assert interfaces["lan"]["speed"] == 0.0
+
+
+# ---------------------------------------------------------------------------
+# get_device_warnings
+# ---------------------------------------------------------------------------
+
+
+class TestGetDeviceWarnings:
+    def test_returns_empty_list(self, driver):
+        # UI_INFO has UpdateState=Available → one warning expected
+        warnings = driver.get_device_warnings()
+        codes = [w["code"] for w in warnings]
+        assert "firmware_update_available" in codes
+
+    def test_no_firmware_warning_when_no_update(self, driver):
+        driver.fc.call_action.side_effect = lambda svc, act, **kw: (
+            UI_INFO_NO_UPDATE if (svc, act) == ("UserInterface1", "X_AVM-DE_GetInfo")
+            else _call_action(svc, act, **kw)
+        )
+        warnings = driver.get_device_warnings()
+        assert not any(w["code"] == "firmware_update_available" for w in warnings)
+
+    def test_dsl_snr_low_warning(self, driver):
+        low_snr = dict(DSL_INFO, NewDownstreamNoiseMargin=40)  # 4.0 dB < 6 dB threshold
+        driver.fc.call_action.side_effect = lambda svc, act, **kw: (
+            low_snr if (svc, act) == ("WANDSLInterfaceConfig1", "GetInfo")
+            else _call_action(svc, act, **kw)
+        )
+        warnings = driver.get_device_warnings()
+        assert any(w["code"] == "dsl_snr_low" for w in warnings)
+
+    def test_no_dsl_warning_when_snr_ok(self, driver):
+        warnings = driver.get_device_warnings()
+        assert not any(w["code"] == "dsl_snr_low" for w in warnings)
+
+
+# ---------------------------------------------------------------------------
+# get_system_config (DSL stats, firmware, DNS, NTP)
+# ---------------------------------------------------------------------------
+
+
+class TestGetSystemConfig:
+    def test_dsl_downstream_rate(self, driver):
+        cfg = driver.get_system_config()
+        assert cfg["dsl_downstream_rate"] == 100000
+
+    def test_dsl_upstream_rate(self, driver):
+        assert driver.get_system_config()["dsl_upstream_rate"] == 40000
+
+    def test_dsl_snr_margin(self, driver):
+        cfg = driver.get_system_config()
+        assert cfg["dsl_downstream_noise_margin"] == 120
+        assert cfg["dsl_upstream_noise_margin"] == 80
+
+    def test_dsl_attenuation(self, driver):
+        cfg = driver.get_system_config()
+        assert cfg["dsl_downstream_attenuation"] == 150
+
+    def test_dsl_crc_errors(self, driver):
+        assert driver.get_system_config()["dsl_crc_errors"] == 5
+
+    def test_dsl_fec_errors(self, driver):
+        assert driver.get_system_config()["dsl_fec_errors"] == 12
+
+    def test_firmware_update_available(self, driver):
+        assert driver.get_system_config()["firmware_update_available"] is True
+
+    def test_firmware_not_available_when_none(self, driver):
+        driver.fc.call_action.side_effect = lambda svc, act, **kw: (
+            UI_INFO_NO_UPDATE if (svc, act) == ("UserInterface1", "X_AVM-DE_GetInfo")
+            else _call_action(svc, act, **kw)
+        )
+        assert driver.get_system_config()["firmware_update_available"] is False
+
+    def test_dns_servers_parsed(self, driver):
+        cfg = driver.get_system_config()
+        assert cfg["dns_servers"] == ["1.1.1.1", "8.8.8.8"]
+
+    def test_ntp_server(self, driver):
+        assert driver.get_system_config()["ntp_server"] == "ntp.example.com"
+
+    def test_empty_when_no_dsl_service(self, driver):
+        del driver.fc.services["WANDSLInterfaceConfig1"]
+        cfg = driver.get_system_config()
+        assert "dsl_downstream_rate" not in cfg
+
+
+# ---------------------------------------------------------------------------
+# get_port_forwards — IPv6 pinholes
+# ---------------------------------------------------------------------------
+
+
+class TestIPv6Pinholes:
+    def test_ipv6_pinhole_appended(self, driver):
+        forwards = driver.get_port_forwards()
+        protocols = [f["protocol"] for f in forwards]
+        assert "TCP6" in protocols
+
+    def test_ipv6_pinhole_fields(self, driver):
+        forwards = driver.get_port_forwards()
+        ssh6 = next(f for f in forwards if f["protocol"] == "TCP6")
+        assert ssh6["name"] == "SSH6"
+        assert ssh6["internal_ip"] == "2001:db8::10"
+        assert ssh6["internal_port"] == 22
+        assert ssh6["enabled"] is True
+
+    def test_no_pinholes_without_ipv6_service(self, driver):
+        del driver.fc.services["WANIPv6Firewall1"]
+        forwards = driver.get_port_forwards()
+        assert all(f["protocol"] not in ("TCP6", "UDP6") for f in forwards)