"""Unit tests for FritzBoxDriver — no real device required.""" from unittest.mock import MagicMock, patch import pytest from fritzconnection.core.exceptions import ( FritzActionError, FritzArrayIndexError, FritzServiceError, ) from napalm.base.exceptions import ConnectionClosedException, ConnectionException from napalm_fritzbox.fritzbox import FritzBoxDriver # --------------------------------------------------------------------------- # Sample TR-064 responses # --------------------------------------------------------------------------- DEVICE_INFO = { "NewModelName": "FRITZ!Box 7590", "NewSoftwareVersion": "7.57", "NewSerialNumber": "1234567890ABC", "NewUpTime": 123456, } LAN_INFO = { "NewEnable": True, "NewStatus": "Up", "NewMACAddress": "AA:BB:CC:DD:EE:01", "NewMaxBitRate": 1000, } WAN_LINK_PROPERTIES = { "NewWANAccessType": "DSL", "NewPhysicalLinkStatus": "Up", "NewLayer1UpstreamMaxBitRate": 40000000, "NewLayer1DownstreamMaxBitRate": 250000000, } WAN_ADDON_INFOS = { "NewTotalBytesSent": 123456789, "NewTotalBytesReceived": 987654321, } WAN_STATUS_INFO = { "NewConnectionStatus": "Connected", "NewUptime": 345600, } WAN_EXTERNAL_IP = {"NewExternalIPAddress": "203.0.113.7"} HOSTS_COUNT = {"NewHostNumberOfEntries": 2} HOST_ENTRIES = [ { "NewMACAddress": "11:22:33:44:55:66", "NewIPAddress": "192.168.1.50", "NewHostName": "laptop", "NewInterfaceType": "WLAN", "NewActive": True, "NewLeaseTimeRemaining": 3600, }, { "NewMACAddress": "AA:11:22:33:44:55", "NewIPAddress": "192.168.1.51", "NewHostName": "desktop", "NewInterfaceType": "LAN", "NewActive": False, }, ] PORT_FORWARDS = [ { "NewPortMappingDescription": "Web", "NewProtocol": "TCP", "NewExternalPort": 443, "NewInternalClient": "192.168.1.10", "NewInternalPort": 443, "NewPortMappingEnabled": True, "NewRemoteHost": "", } ] WLAN1_INFO = { "NewEnable": True, "NewStatus": "Up", "NewSSID": "MyWifi", "NewBSSID": "AA:BB:CC:DD:EE:02", "NewMaxBitRate": 866, "NewChannel": 36, "NewBeaconType": "11i", } WLAN1_TOTAL_ASSOCIATIONS = {"NewTotalAssociations": 1} WLAN1_ASSOCIATED_DEVICE = { "NewAssociatedDeviceMACAddress": "11:22:33:44:55:66", "NewAssociatedDeviceIPAddress": "192.168.1.50", "NewX_AVM-DE_SignalStrength": 80, "NewX_AVM-DE_Speed": 866, } WLAN2_INFO = { "NewEnable": False, "NewStatus": "Down", "NewSSID": "MyWifi-Guest", "NewBSSID": "AA:BB:CC:DD:EE:03", "NewMaxBitRate": 0, "NewChannel": 1, "NewBeaconType": "11i", } WLAN2_TOTAL_ASSOCIATIONS = {"NewTotalAssociations": 0} # --------------------------------------------------------------------------- # call_action dispatch # --------------------------------------------------------------------------- _SIMPLE_RESPONSES = { ("DeviceInfo1", "GetInfo"): DEVICE_INFO, ("LANEthernetInterfaceConfig1", "GetInfo"): LAN_INFO, ("WANCommonInterfaceConfig1", "GetCommonLinkProperties"): WAN_LINK_PROPERTIES, ("WANCommonInterfaceConfig1", "GetAddonInfos"): WAN_ADDON_INFOS, ("WANIPConnection1", "GetStatusInfo"): WAN_STATUS_INFO, ("WANIPConnection1", "GetExternalIPAddress"): WAN_EXTERNAL_IP, ("Hosts1", "GetHostNumberOfEntries"): HOSTS_COUNT, ("WLANConfiguration1", "GetInfo"): WLAN1_INFO, ("WLANConfiguration1", "GetTotalAssociations"): WLAN1_TOTAL_ASSOCIATIONS, ("WLANConfiguration2", "GetInfo"): WLAN2_INFO, ("WLANConfiguration2", "GetTotalAssociations"): WLAN2_TOTAL_ASSOCIATIONS, } def _call_action(service, action, **kwargs): if (service, action) == ("Hosts1", "GetGenericHostEntry"): index = kwargs["NewIndex"] if index >= len(HOST_ENTRIES): raise FritzArrayIndexError("index out of range") return HOST_ENTRIES[index] if (service, action) == ("WANIPConnection1", "GetGenericPortMappingEntry"): index = kwargs["NewPortMappingIndex"] if index >= len(PORT_FORWARDS): raise FritzArrayIndexError("index out of range") return PORT_FORWARDS[index] if (service, action) == ("WLANConfiguration1", "GetGenericAssociatedDeviceInfo"): index = kwargs["NewAssociatedDeviceIndex"] if index >= 1: raise FritzArrayIndexError("index out of range") return WLAN1_ASSOCIATED_DEVICE if (service, action) == ("WLANConfiguration2", "GetGenericAssociatedDeviceInfo"): raise FritzArrayIndexError("index out of range") if action == "X_AVM-DE_GetExternalIPv6Address": raise FritzActionError("action not supported") return _SIMPLE_RESPONSES[(service, action)] # --------------------------------------------------------------------------- # Fixtures # --------------------------------------------------------------------------- @pytest.fixture def driver(): """Return a driver instance with a mocked FritzConnection.""" mock_fc = MagicMock() mock_fc.services = { "DeviceInfo1": MagicMock(), "LANEthernetInterfaceConfig1": MagicMock(), "WANCommonInterfaceConfig1": MagicMock(), "WANIPConnection1": MagicMock(), "Hosts1": MagicMock(), "WLANConfiguration1": MagicMock(), "WLANConfiguration2": MagicMock(), } mock_fc.call_action.side_effect = _call_action drv = FritzBoxDriver(hostname="fritz.box", username="user", password="pass") drv.fc = mock_fc yield drv # --------------------------------------------------------------------------- # open() / close() / is_alive() # --------------------------------------------------------------------------- class TestOpenClose: def test_open_raises_connection_exception_on_error(self): drv = FritzBoxDriver("unreachable.invalid", "u", "p") with patch("napalm_fritzbox.fritzbox.FritzConnection", side_effect=OSError("boom")): with pytest.raises(ConnectionException): drv.open() def test_open_sets_fc_on_success(self): drv = FritzBoxDriver("fritz.box", "u", "p") mock_fc = MagicMock() mock_fc.call_action.return_value = DEVICE_INFO with patch("napalm_fritzbox.fritzbox.FritzConnection", return_value=mock_fc): drv.open() assert drv.fc is mock_fc def test_close_clears_fc(self, driver): driver.close() assert driver.fc is None def test_close_is_idempotent(self, driver): driver.close() driver.close() class TestIsAlive: def test_returns_false_when_no_fc(self): drv = FritzBoxDriver("fritz.box", "u", "p") assert drv.is_alive() == {"is_alive": False} def test_returns_true_on_successful_connection(self, driver): with patch("napalm_fritzbox.fritzbox.socket.create_connection") as mock_conn: mock_conn.return_value.__enter__ = MagicMock(return_value=None) mock_conn.return_value.__exit__ = MagicMock(return_value=False) result = driver.is_alive() assert result == {"is_alive": True} def test_returns_false_on_socket_error(self, driver): with patch( "napalm_fritzbox.fritzbox.socket.create_connection", side_effect=OSError("refused"), ): result = driver.is_alive() assert result == {"is_alive": False} # --------------------------------------------------------------------------- # _call() # --------------------------------------------------------------------------- class TestCall: def test_raises_when_not_connected(self): drv = FritzBoxDriver("fritz.box", "u", "p") with pytest.raises(ConnectionClosedException): drv._call("DeviceInfo1", "GetInfo") # --------------------------------------------------------------------------- # get_facts() # --------------------------------------------------------------------------- class TestGetFacts: def test_vendor_is_avm(self, driver): assert driver.get_facts()["vendor"] == "AVM" def test_model_parsed(self, driver): assert driver.get_facts()["model"] == "FRITZ!Box 7590" def test_os_version_parsed(self, driver): assert driver.get_facts()["os_version"] == "7.57" def test_serial_number_parsed(self, driver): assert driver.get_facts()["serial_number"] == "1234567890ABC" def test_uptime_parsed(self, driver): assert driver.get_facts()["uptime"] == 123456 def test_interface_list_populated(self, driver): facts = driver.get_facts() assert "lan" in facts["interface_list"] assert "wan" in facts["interface_list"] # --------------------------------------------------------------------------- # get_interfaces() / get_interfaces_ip() # --------------------------------------------------------------------------- class TestGetInterfaces: def test_lan_interface_present(self, driver): interfaces = driver.get_interfaces() assert interfaces["lan"]["is_up"] is True assert interfaces["lan"]["mac_address"] == "aa:bb:cc:dd:ee:01" def test_wan_interface_present(self, driver): interfaces = driver.get_interfaces() assert interfaces["wan"]["is_up"] is True assert interfaces["wan"]["description"] == "DSL" def test_wlan_interfaces_present(self, driver): interfaces = driver.get_interfaces() assert interfaces["wlanconfiguration1"]["description"] == "MyWifi" assert interfaces["wlanconfiguration2"]["is_enabled"] is False class TestGetInterfacesIp: def test_wan_external_ip(self, driver): result = driver.get_interfaces_ip() assert result["wan"]["ipv4"] == {"203.0.113.7": {"prefix_length": 32}} def test_lan_address_falls_back_to_hostname(self, driver): result = driver.get_interfaces_ip() assert "fritz.box" in result["lan"]["ipv4"] # --------------------------------------------------------------------------- # get_hosts() / get_arp_table() # --------------------------------------------------------------------------- class TestGetHosts: def test_returns_all_hosts(self, driver): assert len(driver.get_hosts()) == 2 def test_active_host_fields(self, driver): host = driver.get_hosts()[0] assert host["mac"] == "11:22:33:44:55:66" assert host["ip"] == "192.168.1.50" assert host["hostname"] == "laptop" assert host["interface_type"] == "WLAN" assert host["is_active"] is True assert host["lease_time_remaining"] == 3600 def test_inactive_host_has_no_lease(self, driver): host = driver.get_hosts()[1] assert host["is_active"] is False assert "lease_time_remaining" not in host class TestGetArpTable: def test_only_active_hosts_with_mac_and_ip(self, driver): arp_table = driver.get_arp_table() assert len(arp_table) == 1 assert arp_table[0]["mac"] == "11:22:33:44:55:66" assert arp_table[0]["ip"] == "192.168.1.50" # --------------------------------------------------------------------------- # get_wan_status() # --------------------------------------------------------------------------- class TestGetWanStatus: def test_connection_type(self, driver): assert driver.get_wan_status()["connection_type"] == "DSL" def test_is_connected(self, driver): assert driver.get_wan_status()["is_connected"] is True def test_external_ip(self, driver): assert driver.get_wan_status()["external_ip"] == "203.0.113.7" def test_uptime(self, driver): assert driver.get_wan_status()["uptime"] == 345600 def test_traffic_counters(self, driver): status = driver.get_wan_status() assert status["bytes_sent"] == 123456789 assert status["bytes_received"] == 987654321 def test_max_bitrates_in_kbit(self, driver): status = driver.get_wan_status() assert status["max_bitrate_up"] == 40000 assert status["max_bitrate_down"] == 250000 def test_no_ipv6_when_unsupported(self, driver): assert "external_ipv6" not in driver.get_wan_status() # --------------------------------------------------------------------------- # get_port_forwards() / get_nat_translations() # --------------------------------------------------------------------------- class TestGetPortForwards: def test_returns_all_rules(self, driver): assert len(driver.get_port_forwards()) == 1 def test_rule_fields(self, driver): forward = driver.get_port_forwards()[0] assert forward["name"] == "Web" assert forward["protocol"] == "TCP" assert forward["external_port"] == 443 assert forward["internal_ip"] == "192.168.1.10" assert forward["internal_port"] == 443 assert forward["enabled"] is True def test_empty_remote_host_not_included(self, driver): assert "remote_host" not in driver.get_port_forwards()[0] class TestGetNatTranslations: def test_derived_from_enabled_port_forwards(self, driver): translations = driver.get_nat_translations() assert len(translations) == 1 assert translations[0]["protocol"] == "tcp" assert translations[0]["inside_local"] == "192.168.1.10:443" assert translations[0]["inside_global"] == "203.0.113.7:443" # --------------------------------------------------------------------------- # get_vpn_tunnels() # --------------------------------------------------------------------------- class TestGetVpnTunnels: def test_returns_empty_dict(self, driver): assert driver.get_vpn_tunnels() == {} # --------------------------------------------------------------------------- # get_wireless_clients() / get_ssids() / get_radio_status() # --------------------------------------------------------------------------- class TestGetWirelessClients: def test_returns_associated_clients(self, driver): clients = driver.get_wireless_clients() assert len(clients) == 1 client = clients[0] assert client["mac"] == "11:22:33:44:55:66" assert client["ssid"] == "MyWifi" assert client["radio"] == "WLANConfiguration1" assert client["signal"] == 80 assert client["ip"] == "192.168.1.50" class TestGetSsids: def test_returns_ssid_for_each_radio(self, driver): ssids = driver.get_ssids() assert "MyWifi" in ssids assert "MyWifi-Guest" in ssids def test_ssid_fields(self, driver): ssid = driver.get_ssids()["MyWifi"] assert ssid["enabled"] is True assert ssid["radio"] == "WLANConfiguration1" assert ssid["bssid"] == "aa:bb:cc:dd:ee:02" assert ssid["clients"] == 1 def test_disabled_ssid_has_zero_clients(self, driver): assert driver.get_ssids()["MyWifi-Guest"]["clients"] == 0 class TestGetRadioStatus: def test_returns_entry_per_radio(self, driver): radios = driver.get_radio_status() assert "WLANConfiguration1" in radios assert "WLANConfiguration2" in radios def test_5ghz_radio_enabled(self, driver): radio = driver.get_radio_status()["WLANConfiguration1"] assert radio["enabled"] is True assert radio["band"] == "5GHz" assert radio["channel"] == 36 def test_2_4ghz_radio_disabled(self, driver): radio = driver.get_radio_status()["WLANConfiguration2"] assert radio["enabled"] is False assert radio["band"] == "2.4GHz" assert radio["channel"] == 1