From 3b3469daa9ed2a190b2b8d5ee544dfc28ea95079 Mon Sep 17 00:00:00 2001 From: Christian Manivong Date: Wed, 24 Jun 2026 22:21:06 +0200 Subject: [PATCH] feat: WAN-Status, DSL-Stats, Portfreigaben, Firmware-Warnings, PPPoE-Interface MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 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 --- napalm_fritzbox/fritzbox.py | 347 ++++++++++++++++++++++++++++-------- tests/unit/test_driver.py | 286 ++++++++++++++++++++++++++++- 2 files changed, 557 insertions(+), 76 deletions(-) diff --git a/napalm_fritzbox/fritzbox.py b/napalm_fritzbox/fritzbox.py index f8e9d85..c5a491c 100644 --- a/napalm_fritzbox/fritzbox.py +++ b/napalm_fritzbox/fritzbox.py @@ -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,34 +234,52 @@ class FritzBoxDriver(ResidentialGatewayDriver): ``WLANConfiguration{1,2,3}.GetInfo``. """ interfaces: dict[str, dict[str, Any]] = {} + available = self.fc.services if self.fc else {} - try: - lan = self._call("LANEthernetInterfaceConfig1", "GetInfo") - interfaces["lan"] = { - "is_up": lan.get("NewStatus", "") == "Up", - "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), - "mtu": 1500, - } - except _OPTIONAL_SERVICE_ERRORS: - pass + if "LANEthernetInterfaceConfig1" in available: + try: + lan = self._call("LANEthernetInterfaceConfig1", "GetInfo") + interfaces["lan"] = { + "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": _to_float(lan.get("NewMaxBitRate")), + "mtu": 1500, + } + except _OPTIONAL_SERVICE_ERRORS: + pass - try: - link = self._call("WANCommonInterfaceConfig1", "GetCommonLinkProperties") - interfaces["wan"] = { - "is_up": link.get("NewPhysicalLinkStatus", "") == "Up", - "is_enabled": True, - "description": link.get("NewWANAccessType", "WAN"), - "last_flapped": -1.0, - "mac_address": "", - "speed": float(link.get("NewLayer1DownstreamMaxBitRate", 0) or 0) / 1000, - "mtu": 1500, - } - except _OPTIONAL_SERVICE_ERRORS: - pass + if "WANCommonInterfaceConfig1" in available: + try: + link = self._call("WANCommonInterfaceConfig1", "GetCommonLinkProperties") + interfaces["wan"] = { + "is_up": link.get("NewPhysicalLinkStatus", "").lower() in ("up", "connected"), + "is_enabled": True, + "description": link.get("NewWANAccessType", "WAN"), + "last_flapped": -1.0, + "mac_address": "", + "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: @@ -261,12 +287,12 @@ class FritzBoxDriver(ResidentialGatewayDriver): 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,66 +462,126 @@ 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 {} - try: - addon = self._call("WANCommonInterfaceConfig1", "GetAddonInfos") - except _OPTIONAL_SERVICE_ERRORS: - addon = {} + link: dict = {} + if "WANCommonInterfaceConfig1" in available: + try: + link = self._call("WANCommonInterfaceConfig1", "GetCommonLinkProperties") + except _OPTIONAL_SERVICE_ERRORS: + pass - service = self._wan_ip_service() - status = self._call(service, "GetStatusInfo") - ext_ip = self._call(service, "GetExternalIPAddress").get("NewExternalIPAddress", "") + addon: dict = {} + if "WANCommonInterfaceConfig1" in available: + try: + addon = self._call("WANCommonInterfaceConfig1", "GetAddonInfos") + except _OPTIONAL_SERVICE_ERRORS: + pass + + # 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] = [] - index = 0 - while True: - try: - entry = self._call(service, "GetGenericPortMappingEntry", NewPortMappingIndex=index) - except (FritzArrayIndexError, FritzConnectionException): - break - forward: PortForwardDict = { - "name": entry.get("NewPortMappingDescription", ""), - "protocol": entry.get("NewProtocol", ""), - "external_port": int(entry.get("NewExternalPort", 0) or 0), - "internal_ip": entry.get("NewInternalClient", ""), - "internal_port": int(entry.get("NewInternalPort", 0) or 0), - "enabled": bool(entry.get("NewPortMappingEnabled", False)), - } - remote_host = entry.get("NewRemoteHost", "") - if remote_host: - forward["remote_host"] = remote_host - forwards.append(forward) - index += 1 + try: + service = self._wan_ip_service() + index = 0 + while True: + try: + entry = self._call(service, "GetGenericPortMappingEntry", NewPortMappingIndex=index) + except (FritzArrayIndexError, FritzConnectionException): + break + + forward: PortForwardDict = { + "name": entry.get("NewPortMappingDescription", ""), + "protocol": entry.get("NewProtocol", ""), + "external_port": int(entry.get("NewExternalPort", 0) or 0), + "internal_ip": entry.get("NewInternalClient", ""), + "internal_port": int(entry.get("NewInternalPort", 0) or 0), + "enabled": bool(entry.get("NewPortMappingEnabled", False)), + } + remote_host = entry.get("NewRemoteHost", "") + if remote_host: + 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", diff --git a/tests/unit/test_driver.py b/tests/unit/test_driver.py index 61706a9..9f2edc4 100644 --- a/tests/unit/test_driver.py +++ b/tests/unit/test_driver.py @@ -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)