Christian Manivong
13e0073659
Closes netork#116. `get_port_forwards` reports IPv4 mappings and IPv6 pinholes together, which is right — both are inbound rules someone configured. `get_nat_translations` then iterated all of them, and that is not: a pinhole is a firewall hole, and IPv6 does not do NAT at all. The entries it produced were nonsense in two ways. They paired an IPv6 host with the IPv4 WAN address, and `inside_local` came out as "2001:db8::10:22", where the last colon is a port separator and everything before it is an address that also contains colons. The two failing tests were pulling in opposite directions. `TestGetNatTranslations` was right and the implementation had drifted past it when pinhole support landed. `TestGetPortForwards::test_returns_all_rules` was the opposite: it still expected one rule from before pinholes existed, which put it in direct contradiction with TestIPv6Pinholes further down the same file.
763 lines
26 KiB
Python
763 lines
26 KiB
Python
"""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, _to_float
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 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}
|
|
|
|
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
|
|
# ---------------------------------------------------------------------------
|
|
|
|
_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,
|
|
("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,
|
|
}
|
|
|
|
|
|
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) == ("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:
|
|
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 (WANIPConnection1 / cable)."""
|
|
mock_fc = MagicMock()
|
|
mock_fc.services = {
|
|
"DeviceInfo1": MagicMock(),
|
|
"LANEthernetInterfaceConfig1": MagicMock(),
|
|
"WANCommonInterfaceConfig1": MagicMock(),
|
|
"WANIPConnection1": MagicMock(),
|
|
"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
|
|
|
|
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_ipv4_mappings_and_ipv6_pinholes(self, driver):
|
|
"""One IPv4 mapping plus one IPv6 pinhole.
|
|
|
|
This asserted 1 from before pinhole support existed, which put it in
|
|
direct contradiction with TestIPv6Pinholes further down the same file.
|
|
"""
|
|
forwards = driver.get_port_forwards()
|
|
assert len(forwards) == 2
|
|
assert {f["protocol"] for f in forwards} == {"TCP", "TCP6"}
|
|
|
|
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_ipv4_port_forwards(self, driver):
|
|
"""Only the IPv4 mapping. IPv6 pinholes are firewall holes, not
|
|
translations, and IPv6 does not do NAT — including them paired an IPv6
|
|
host with the IPv4 WAN address."""
|
|
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
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# _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_pinholes_are_not_reported_as_nat_translations(self, driver):
|
|
"""A pinhole opens the firewall for a globally routable address; there
|
|
is nothing to translate."""
|
|
assert all(":" not in t["inside_local"].rsplit(":", 1)[0]
|
|
for t in driver.get_nat_translations())
|
|
|
|
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)
|