422 lines
14 KiB
Python
422 lines
14 KiB
Python
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"]) == {}
|