324 lines
12 KiB
Python
324 lines
12 KiB
Python
"""Tests for NTP, SNMP, users, config management, ping, traceroute, CLI."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import pytest
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
|
|
class TestGetNTPServers:
|
|
def test_returns_configured_servers(self, driver):
|
|
result = driver.get_ntp_servers()
|
|
assert "pool.ntp.org" in result
|
|
assert "time.cloudflare.com" in result
|
|
|
|
def test_empty_when_no_ntp(self, driver):
|
|
driver._node_api().ntp.get.return_value = {}
|
|
result = driver.get_ntp_servers()
|
|
assert result == {}
|
|
|
|
|
|
class TestGetNTPStats:
|
|
NTP_OUTPUT = (
|
|
"*192.168.1.1 10.0.0.1 4 u 12 64 377 1.234 0.567 0.123\n"
|
|
"+192.168.1.2 10.0.0.2 3 u 30 64 377 2.345 -0.123 0.234\n"
|
|
)
|
|
|
|
def test_synchronized_server(self, driver):
|
|
driver._node_api().execute.post.return_value = {"data": self.NTP_OUTPUT}
|
|
stats = driver.get_ntp_stats()
|
|
synced = [s for s in stats if s["synchronized"]]
|
|
assert len(synced) == 1
|
|
assert synced[0]["remote"] == "192.168.1.1"
|
|
|
|
def test_stratum(self, driver):
|
|
driver._node_api().execute.post.return_value = {"data": self.NTP_OUTPUT}
|
|
stats = driver.get_ntp_stats()
|
|
assert stats[0]["stratum"] == 4
|
|
|
|
def test_empty_when_no_ntp_command(self, driver):
|
|
driver._node_api().execute.post.return_value = {"data": ""}
|
|
stats = driver.get_ntp_stats()
|
|
assert stats == []
|
|
|
|
|
|
class TestGetSNMPInformation:
|
|
SNMP_CONF = (
|
|
"# SNMP config\n"
|
|
"rocommunity public\n"
|
|
"rwcommunity private\n"
|
|
"sysLocation Server Room A\n"
|
|
"sysContact admin@example.com\n"
|
|
)
|
|
|
|
def test_communities(self, driver):
|
|
driver._node_api().execute.post.return_value = {"data": self.SNMP_CONF}
|
|
result = driver.get_snmp_information()
|
|
assert "public" in result["community"]
|
|
assert result["community"]["public"]["mode"] == "ro"
|
|
assert "private" in result["community"]
|
|
assert result["community"]["private"]["mode"] == "rw"
|
|
|
|
def test_location(self, driver):
|
|
driver._node_api().execute.post.return_value = {"data": self.SNMP_CONF}
|
|
result = driver.get_snmp_information()
|
|
assert result["location"] == "Server Room A"
|
|
|
|
def test_contact(self, driver):
|
|
driver._node_api().execute.post.return_value = {"data": self.SNMP_CONF}
|
|
result = driver.get_snmp_information()
|
|
assert result["contact"] == "admin@example.com"
|
|
|
|
def test_chassis_id_is_node_name(self, driver):
|
|
driver._node_api().execute.post.return_value = {"data": ""}
|
|
result = driver.get_snmp_information()
|
|
assert result["chassis_id"] == "pve1"
|
|
|
|
|
|
class TestGetUsers:
|
|
def test_pve_users_returned(self, driver):
|
|
driver._node_api().execute.post.return_value = {"data": ""}
|
|
result = driver.get_users()
|
|
assert "root@pam" in result
|
|
assert "napalm@pam" in result
|
|
|
|
def test_user_has_required_keys(self, driver):
|
|
driver._node_api().execute.post.return_value = {"data": ""}
|
|
result = driver.get_users()
|
|
for user, data in result.items():
|
|
assert "level" in data
|
|
assert "password" in data
|
|
assert "sshkeys" in data
|
|
|
|
def test_root_local_user_from_passwd(self, driver):
|
|
passwd_output = "root:x:0:0:root:/root:/bin/bash\n"
|
|
driver._node_api().execute.post.return_value = {"data": passwd_output}
|
|
result = driver.get_users()
|
|
assert "root" in result
|
|
assert result["root"]["level"] == 15
|
|
|
|
|
|
class TestConfigManagement:
|
|
RUNNING = "auto eth0\niface eth0 inet static\n address 10.0.0.1/24\n"
|
|
CANDIDATE = "auto eth0\niface eth0 inet static\n address 10.0.0.2/24\n"
|
|
|
|
def test_load_merge_candidate_from_string(self, driver):
|
|
driver.load_merge_candidate(config=self.CANDIDATE)
|
|
assert driver._candidate_config == self.CANDIDATE
|
|
|
|
def test_load_replace_candidate_from_string(self, driver):
|
|
driver.load_replace_candidate(config=self.CANDIDATE)
|
|
assert driver._candidate_config == self.CANDIDATE
|
|
|
|
def test_load_merge_from_file(self, driver, tmp_path):
|
|
f = tmp_path / "candidate.txt"
|
|
f.write_text(self.CANDIDATE)
|
|
driver.load_merge_candidate(filename=str(f))
|
|
assert driver._candidate_config == self.CANDIDATE
|
|
|
|
def test_discard_clears_candidate(self, driver):
|
|
driver.load_replace_candidate(config=self.CANDIDATE)
|
|
driver.discard_config()
|
|
assert driver._candidate_config == ""
|
|
|
|
def test_compare_config_diff(self, driver):
|
|
driver._running_config = self.RUNNING
|
|
driver.load_replace_candidate(config=self.CANDIDATE)
|
|
diff = driver.compare_config()
|
|
assert "10.0.0.1" in diff
|
|
assert "10.0.0.2" in diff
|
|
|
|
def test_compare_config_no_diff_when_same(self, driver):
|
|
driver._running_config = self.RUNNING
|
|
driver.load_replace_candidate(config=self.RUNNING)
|
|
diff = driver.compare_config()
|
|
assert diff == ""
|
|
|
|
def test_get_config_running(self, driver):
|
|
driver._node_api().execute.post.return_value = {"data": self.RUNNING}
|
|
result = driver.get_config(retrieve="running")
|
|
assert self.RUNNING in result["running"]
|
|
assert result["candidate"] == ""
|
|
|
|
def test_get_config_sanitized_redacts_password(self, driver):
|
|
raw = "password mysecret\nauto eth0\n"
|
|
driver._node_api().execute.post.return_value = {"data": raw}
|
|
result = driver.get_config(retrieve="running", sanitized=True)
|
|
assert "mysecret" not in result["running"]
|
|
assert "****" in result["running"]
|
|
|
|
def test_load_merge_raises_without_input(self, driver):
|
|
with pytest.raises((ValueError, TypeError)):
|
|
driver.load_merge_candidate()
|
|
|
|
|
|
class TestPing:
|
|
PING_SUCCESS = (
|
|
"PING 8.8.8.8 (8.8.8.8) 100(128) bytes of data.\n"
|
|
"108 bytes from 8.8.8.8: icmp_seq=1 ttl=118 time=10.5 ms\n"
|
|
"108 bytes from 8.8.8.8: icmp_seq=2 ttl=118 time=11.2 ms\n"
|
|
"--- 8.8.8.8 ping statistics ---\n"
|
|
"2 packets transmitted, 2 received, 0% packet loss, time 1001ms\n"
|
|
"rtt min/avg/max/mdev = 10.500/10.850/11.200/0.350 ms\n"
|
|
)
|
|
PING_FAIL = (
|
|
"PING badhost: Name or service not known\n"
|
|
"0 packets transmitted, 0 received, 100% packet loss\n"
|
|
)
|
|
|
|
def test_success(self, driver):
|
|
driver._node_api().execute.post.return_value = {"data": self.PING_SUCCESS}
|
|
result = driver.ping("8.8.8.8")
|
|
assert "success" in result
|
|
assert result["success"]["probes_sent"] == 2
|
|
assert result["success"]["packet_loss"] == 0
|
|
assert result["success"]["rtt_min"] == pytest.approx(10.5, abs=0.01)
|
|
|
|
def test_failure(self, driver):
|
|
driver._node_api().execute.post.return_value = {"data": self.PING_FAIL}
|
|
result = driver.ping("badhost")
|
|
assert "error" in result
|
|
|
|
def test_exec_unavailable(self, driver):
|
|
driver._node_api().execute.post.return_value = {"data": ""}
|
|
result = driver.ping("8.8.8.8")
|
|
assert "error" in result
|
|
|
|
|
|
class TestTraceroute:
|
|
TR_OUTPUT = (
|
|
"traceroute to 8.8.8.8 (8.8.8.8), 30 hops max, 60 byte packets\n"
|
|
" 1 10.0.0.254 0.456 ms 0.423 ms 0.401 ms\n"
|
|
" 2 192.168.1.1 1.234 ms 1.210 ms 1.189 ms\n"
|
|
" 3 8.8.8.8 11.200 ms 11.150 ms 11.100 ms\n"
|
|
)
|
|
|
|
def test_success_hops(self, driver):
|
|
driver._node_api().execute.post.return_value = {"data": self.TR_OUTPUT}
|
|
result = driver.traceroute("8.8.8.8")
|
|
assert "success" in result
|
|
assert 1 in result["success"]
|
|
assert 3 in result["success"]
|
|
|
|
def test_hop_rtt(self, driver):
|
|
driver._node_api().execute.post.return_value = {"data": self.TR_OUTPUT}
|
|
result = driver.traceroute("8.8.8.8")
|
|
hop1 = result["success"][1]["probes"]
|
|
assert hop1[1]["ip_address"] == "10.0.0.254"
|
|
|
|
def test_empty_returns_error(self, driver):
|
|
driver._node_api().execute.post.return_value = {"data": ""}
|
|
result = driver.traceroute("8.8.8.8")
|
|
assert "error" in result
|
|
|
|
|
|
class TestCLI:
|
|
def test_cli_single_command(self, driver):
|
|
driver._node_api().execute.post.return_value = {"data": "hello world"}
|
|
result = driver.cli(["echo hello world"])
|
|
assert result["echo hello world"] == "hello world"
|
|
|
|
def test_cli_multiple_commands(self, driver):
|
|
responses = {"uptime": "up 2 days", "hostname": "pve1"}
|
|
call_count = 0
|
|
|
|
def _exec(command):
|
|
nonlocal call_count
|
|
cmd = command
|
|
call_count += 1
|
|
return {"data": responses.get(cmd, "")}
|
|
|
|
driver._node_api().execute.post.side_effect = _exec
|
|
result = driver.cli(["uptime", "hostname"])
|
|
assert "uptime" in result
|
|
assert "hostname" in result
|
|
|
|
|
|
class TestGetRouteTo:
|
|
ROUTE_OUTPUT = (
|
|
"10.0.0.0/24 dev eth0 proto kernel scope link src 10.0.0.1\n"
|
|
"default via 10.0.0.254 dev eth0 proto dhcp\n"
|
|
)
|
|
|
|
def test_returns_routes(self, driver):
|
|
driver._node_api().execute.post.return_value = {"data": self.ROUTE_OUTPUT}
|
|
result = driver.get_route_to(destination="10.0.0.0/24")
|
|
assert "10.0.0.0/24" in result
|
|
|
|
def test_route_fields(self, driver):
|
|
driver._node_api().execute.post.return_value = {"data": self.ROUTE_OUTPUT}
|
|
result = driver.get_route_to()
|
|
for prefix, routes in result.items():
|
|
for route in routes:
|
|
assert "protocol" in route
|
|
assert "next_hop" in route
|
|
assert "outgoing_interface" in route
|
|
|
|
def test_protocol_filter(self, driver):
|
|
driver._node_api().execute.post.return_value = {"data": self.ROUTE_OUTPUT}
|
|
result = driver.get_route_to(protocol="dhcp")
|
|
for prefix, routes in result.items():
|
|
for route in routes:
|
|
assert "dhcp" in route["protocol"].lower()
|
|
|
|
|
|
class TestLLDPNeighbors:
|
|
LLDP_SUMMARY = (
|
|
" Interface: eth0\n"
|
|
" SysName: sw01.example.com\n"
|
|
" PortID: ifname GigabitEthernet1/0/1\n"
|
|
" Interface: eth1\n"
|
|
" SysName: sw02.example.com\n"
|
|
" PortID: ifname GigabitEthernet1/0/2\n"
|
|
)
|
|
|
|
def test_neighbors_found(self, driver):
|
|
driver._node_api().execute.post.return_value = {"data": self.LLDP_SUMMARY}
|
|
result = driver.get_lldp_neighbors()
|
|
assert "eth0" in result
|
|
assert result["eth0"][0]["hostname"] == "sw01.example.com"
|
|
|
|
def test_empty_when_no_lldp(self, driver):
|
|
driver._node_api().execute.post.return_value = {"data": ""}
|
|
result = driver.get_lldp_neighbors()
|
|
assert result == {}
|
|
|
|
|
|
class TestIPv6Neighbors:
|
|
IPV6_NEIGH = (
|
|
"2001:db8::1 dev eth0 lladdr aa:bb:cc:dd:ee:ff REACHABLE\n"
|
|
"2001:db8::2 dev vmbr0 lladdr 11:22:33:44:55:66 STALE\n"
|
|
"fe80::1 dev eth0 FAILED\n"
|
|
)
|
|
|
|
def test_neighbors_parsed(self, driver):
|
|
driver._node_api().execute.post.return_value = {"data": self.IPV6_NEIGH}
|
|
result = driver.get_ipv6_neighbors_table()
|
|
assert len(result) == 2 # FAILED entry excluded
|
|
|
|
def test_mac_normalized(self, driver):
|
|
driver._node_api().execute.post.return_value = {"data": self.IPV6_NEIGH}
|
|
result = driver.get_ipv6_neighbors_table()
|
|
macs = {e["mac"] for e in result}
|
|
assert "aa:bb:cc:dd:ee:ff" in macs
|
|
|
|
def test_state_present(self, driver):
|
|
driver._node_api().execute.post.return_value = {"data": self.IPV6_NEIGH}
|
|
result = driver.get_ipv6_neighbors_table()
|
|
states = {e["state"] for e in result}
|
|
assert "REACHABLE" in states
|
|
|
|
|
|
class TestNotImplemented:
|
|
def test_get_bgp_config_raises(self, driver):
|
|
with pytest.raises(NotImplementedError):
|
|
driver.get_bgp_config()
|
|
|
|
def test_get_bgp_neighbors_raises(self, driver):
|
|
with pytest.raises(NotImplementedError):
|
|
driver.get_bgp_neighbors()
|
|
|
|
def test_get_optics_raises(self, driver):
|
|
with pytest.raises(NotImplementedError):
|
|
driver.get_optics()
|