import json from unittest.mock import MagicMock, patch import requests from napalm.base.exceptions import ConnectionException from napalm_zyxel import ZyxelVMGDriver MOCK_BASIC_INFO = { "ModelName": "VMG4005-B50A", "SoftwareVersion": "V5.13(ABKS.1)C0", "SerialNumber": "S220L12345678", "HostName": "ZyxelVMG", "WanIp": "85.2.34.56", "LanIp": "192.168.1.1", "MacAddress": "00:1A:2B:3C:4D:5E", "SystemUpTime": "1234567", "WanAccessType": "VDSL", "MaxUpstream": "30000000", "MaxDownstream": "100000000", } MOCK_RSA_KEY = ( "-----BEGIN PUBLIC KEY-----\n" "MIIBIjANBgkqhkiG9w0BAQEFAAOCAQ8AMIIBCgKCAQEAw8kFmWQhFfTfL8eP\n" "yGsqQw5g3n4R7pG9vLxKZJtRq0pQ+kCqWzXHmTZvLxKZJtRq0pQ+kCqWzX\n" "HmTZvLxKZJtRq0pQ+kCqWzXHmTZvLxKZJtRq0pQ+kCqWzXHmTZvLxKZJtR\n" "q0pQ+kCqWzXHmTZvLxKZJtRq0pQ+kCqWzXHmTZvLxKZJtRq0pQ+kCqWzXH\n" "mTZvLxKZJtRq0pQ+kCqWzXHmTZvLxKZJtRq0pQ+kCqWzXHmTZvLxKZJtRq\n" "0pQ+kCqWzXHmTZvLxKZJtRq0pQ+kCqWzXHmTZvLxKZJtRq0pQ+kCqWzXHm\n" "TZvLxKZJtRq0pQ+kCqWzXHmTZvLxKZJtRq0pQ+kCqWzXHmTZvLxKZJtRq0\n" "pQIDAQAB\n" "-----END PUBLIC KEY-----\n" ) MOCK_LOGIN_RESPONSE = { "content": "", "iv": "QkJCQkJCQkJCQkJCQkJCQg==", "key": "", } class MockResponse: def __init__(self, json_data, status_code=200, text=""): self._json = json_data self.status_code = status_code self._text = text or json.dumps(json_data) def json(self): return self._json @property def text(self): return self._text def raise_for_status(self): if self.status_code >= 400: raise requests.HTTPError(f"HTTP {self.status_code}") class TestZyxelVMGDriver: def setup_method(self): self.driver = ZyxelVMGDriver( "192.168.1.1", "admin", "password", ) def _setup_mocked_http(self, monkeypatch_crypto=True): session = MagicMock(spec=requests.Session) session.verify = False session.timeout = 60 session.trust_env = False self.driver._session = session self.driver._device_info = MOCK_BASIC_INFO def mock_get(url, **kwargs): if "getBasicInformation" in url: return MockResponse(MOCK_BASIC_INFO) if "getRSAPublickKey" in url: return MockResponse({"RSAPublicKey": MOCK_RSA_KEY}) if "UserLogout" in url: return MockResponse({}) return MockResponse({}) def mock_post(url, **kwargs): if "UserLogin" in url: return MockResponse(MOCK_LOGIN_RESPONSE) return MockResponse({}) session.get.side_effect = mock_get session.post.side_effect = mock_post if monkeypatch_crypto: crypto = MagicMock() crypto.encrypt_json.return_value = {"content": "ENC", "iv": "IV", "key": ""} crypto.decrypt_response.return_value = {"result": "ZCFG_SUCCESS", "sessionkey": "12345"} crypto.encrypt_aes_key_with_rsa.return_value = "ENC_KEY" crypto.aes_key = b"\x42" * 32 self.driver._crypto = crypto self.driver._sessionkey = "12345" def test_init(self): assert self.driver.hostname == "192.168.1.1" assert self.driver.username == "admin" assert self.driver.password == "password" assert self.driver.port == 80 assert self.driver.use_tls is False assert self.driver.VENDOR == "Zyxel" def test_open_success(self): with patch("napalm_zyxel.zyxel_driver.ZyxelCrypto") as mock_crypto_cls: mock_crypto = MagicMock() mock_crypto.encrypt_json.return_value = {"content": "ENC", "iv": "IV", "key": ""} mock_crypto.decrypt_response.return_value = { "result": "ZCFG_SUCCESS", "sessionkey": "12345" } mock_crypto.encrypt_aes_key_with_rsa.return_value = "ENC_KEY" mock_crypto_cls.return_value = mock_crypto session = MagicMock() session.get.return_value = MockResponse(MOCK_BASIC_INFO) session.post.return_value = MockResponse(MOCK_LOGIN_RESPONSE) with patch.object(self.driver, "_session", session, create=False): with patch.object( self.driver, "_get_rsa_pubkey", return_value=MOCK_RSA_KEY ): with patch.object( self.driver, "_get_basic_info", return_value=MOCK_BASIC_INFO ): self.driver._session = None self.driver._crypto = None self.driver._sessionkey = None with patch("requests.Session", return_value=session): self.driver.open() assert self.driver._sessionkey == "12345" assert self.driver._device_info == MOCK_BASIC_INFO def test_open_failure_unreachable(self): with patch("napalm_zyxel.zyxel_driver.ZyxelCrypto") as mock_crypto_cls: mock_crypto = MagicMock() mock_crypto_cls.return_value = mock_crypto self.driver._session = None with patch("requests.Session") as mock_session: sess = MagicMock() sess.get.side_effect = requests.ConnectionError("unreachable") mock_session.return_value = sess try: self.driver.open() assert False, "Expected ConnectionException" except ConnectionException: pass def test_close(self): self._setup_mocked_http() self.driver.close() assert self.driver._session is None assert self.driver._sessionkey is None def test_is_alive(self): with patch("socket.create_connection") as mock_conn: mock_conn.return_value.__enter__.return_value = MagicMock() assert self.driver.is_alive()["is_alive"] is True def test_is_alive_fail(self): with patch("socket.create_connection") as mock_conn: mock_conn.side_effect = OSError("refused") assert self.driver.is_alive()["is_alive"] is False def test_get_facts(self): self._setup_mocked_http() facts = self.driver.get_facts() assert facts["vendor"] == "Zyxel" assert facts["model"] == "VMG4005-B50A" assert facts["serial_number"] == "S220L12345678" assert facts["os_version"] == "V5.13(ABKS.1)C0" assert facts["hostname"] == "ZyxelVMG" def test_get_interfaces(self): self._setup_mocked_http() ifaces = self.driver.get_interfaces() assert "wan" in ifaces assert "lan" in ifaces assert "wlan_2_4ghz" in ifaces assert "wlan_5ghz" in ifaces def test_get_interfaces_ip(self): self._setup_mocked_http() ips = self.driver.get_interfaces_ip() assert "85.2.34.56" in ips["wan"]["ipv4"] assert "192.168.1.1" in ips["lan"]["ipv4"] def test_get_wan_status(self): self._setup_mocked_http() status = self.driver.get_wan_status() assert status["is_connected"] is True assert status["external_ip"] == "85.2.34.56" assert status["connection_type"] == "VDSL" def test_get_port_forwards_empty(self): self._setup_mocked_http() self.driver._session.get.side_effect = None def mock_get(url, **kwargs): if "getBasicInformation" in url: return MockResponse(MOCK_BASIC_INFO) return MockResponse({}, text="no forwards here") self.driver._session.get.side_effect = mock_get forwards = self.driver.get_port_forwards() assert forwards == [] def test_get_hosts_empty(self): self._setup_mocked_http() self.driver._session.get.side_effect = None def mock_get(url, **kwargs): if "getBasicInformation" in url: return MockResponse(MOCK_BASIC_INFO) return MockResponse({}, text="no dhcp here") self.driver._session.get.side_effect = mock_get hosts = self.driver.get_hosts() assert hosts == [] def test_get_nat_translations_empty(self): self._setup_mocked_http() self.driver._session.get.side_effect = None def mock_get(url, **kwargs): if "getBasicInformation" in url: return MockResponse(MOCK_BASIC_INFO) return MockResponse({}, text="") self.driver._session.get.side_effect = mock_get nat = self.driver.get_nat_translations() assert nat == [] def test_get_vpn_tunnels(self): self._setup_mocked_http() assert self.driver.get_vpn_tunnels() == {} def test_get_wireless_clients_empty(self): self._setup_mocked_http() self.driver._session.get.side_effect = None def mock_get(url, **kwargs): if "getBasicInformation" in url: return MockResponse(MOCK_BASIC_INFO) return MockResponse({}, text="") self.driver._session.get.side_effect = mock_get assert self.driver.get_wireless_clients() == [] def test_get_ssids_empty(self): self._setup_mocked_http() self.driver._session.get.side_effect = None def mock_get(url, **kwargs): if "getBasicInformation" in url: return MockResponse(MOCK_BASIC_INFO) return MockResponse({}, text="") self.driver._session.get.side_effect = mock_get assert self.driver.get_ssids() == {} def test_get_radio_status_empty(self): self._setup_mocked_http() self.driver._session.get.side_effect = None def mock_get(url, **kwargs): if "getBasicInformation" in url: return MockResponse(MOCK_BASIC_INFO) return MockResponse({}, text="") self.driver._session.get.side_effect = mock_get assert self.driver.get_radio_status() == {} def test_get_arp_table(self): self._setup_mocked_http() assert self.driver.get_arp_table() == [] def test_reboot(self): self._setup_mocked_http() result = self.driver.run_device_action("reboot") assert result["action"] == "reboot" def test_run_device_action_unknown(self): self._setup_mocked_http() result = self.driver.run_device_action("foobar") assert result["success"] is False assert "Unknown" in result["message"] def test_load_replace_candidate_raises(self): self._setup_mocked_http() try: self.driver.load_replace_candidate() assert False, "Expected NotImplementedError" except NotImplementedError: pass def test_load_merge_candidate_raises(self): self._setup_mocked_http() try: self.driver.load_merge_candidate() assert False, "Expected NotImplementedError" except NotImplementedError: pass def test_ping_not_implemented(self): self._setup_mocked_http() result = self.driver.ping("8.8.8.8") assert "error" in result def test_get_environment(self): self._setup_mocked_http() assert self.driver.get_environment() == {} def test_dal_query_not_connected(self): from napalm.base.exceptions import ConnectionClosedException self.driver._session = None try: self.driver._dal_query("PINGTEST", {}) assert False, "Expected ConnectionClosedException" except ConnectionClosedException: pass def test_parse_port_forwards_dal(self): self._setup_mocked_http() data = { "NATPortMap": [ { "Description": "Web Server", "Protocol": "TCP", "ExternalPort": 80, "InternalIP": "192.168.1.10", "InternalPort": 8080, "Enabled": True, } ] } result = self.driver._parse_port_forwards_dal(data) assert len(result) == 1 assert result[0]["name"] == "Web Server" assert result[0]["external_port"] == 80 assert result[0]["internal_ip"] == "192.168.1.10" def test_parse_hosts_dal(self): self._setup_mocked_http() data = { "LANHosts": [ { "MACAddress": "aa:bb:cc:dd:ee:ff", "IPAddress": "192.168.1.100", "HostName": "laptop", "InterfaceType": "lan", "Active": True, } ] } result = self.driver._parse_hosts_dal(data) assert len(result) == 1 assert result[0]["mac"] == "aa:bb:cc:dd:ee:ff" assert result[0]["ip"] == "192.168.1.100" assert result[0]["hostname"] == "laptop" def test_parse_wireless_clients_dal(self): self._setup_mocked_http() data = { "assoc": [ { "MACAddress": "11:22:33:44:55:66", "SSID": "MyWiFi", "Radio": "wlan0", "Signal": -45, "TxRate": 300.0, "RxRate": 150.0, "Uptime": 3600, "IPAddress": "192.168.1.50", } ] } result = self.driver._parse_wireless_clients_dal(data) assert len(result) == 1 assert result[0]["mac"] == "11:22:33:44:55:66" assert result[0]["signal"] == -45 assert result[0]["ip"] == "192.168.1.50" def test_parse_ssids_dal(self): self._setup_mocked_http() data = { "ssids": [ { "SSID": "MyWiFi", "Radio": "wlan0", "BSSID": "00:1a:2b:3c:4d:5e", "Encryption": "WPA2", "Enabled": True, "Hidden": False, "Clients": 3, } ] } result = self.driver._parse_ssids_dal(data) assert "MyWiFi" in result assert result["MyWiFi"]["encryption"] == "WPA2" assert result["MyWiFi"]["clients"] == 3 def test_get_config(self): self._setup_mocked_http() assert self.driver.get_config() == {} def test_cli(self): self._setup_mocked_http() assert self.driver.cli(["show version"]) == {}