Christian Manivong
20fcf2ebc3
Closes netork#115. The suite had been red long enough that it stopped being read. Four of the fourteen failures were the tests being right. `interfaces_mixin.py` used `re.match` without importing `re`, so `get_mac_address_table` raised NameError against any node with a Linux bridge. The tests never reached that line: they mocked the API call underneath `_exec_ssh_command`, which takes two positional arguments where the doubles accepted one, and which base64-wraps the command — so a fixture keyed on "bridge fdb" appearing in the text matched nothing and the helper returned "". They mock `_exec_ssh_command` itself now, which is the driver's own seam. `is_alive` called `_resolve_node()`, which returns early without touching the API whenever a node was configured through optional_args. A dead connection reported itself alive. It probes `GET /version` now. The documented `realm` optional_arg was read into `self._realm` in `__init__` and then never used. Proxmox authenticates against "<user>@<realm>" and rejects a bare username, so the option had no effect and callers had to know to type the realm themselves. `get_vlans` filtered out entries with no member ports on one return path while the OVS path returned them, so a configured SDN VNet was visible or invisible depending on which branch ran. A VNet exists on the node whether or not anything is attached to it, and netOrk's VLAN discovery reads this. `get_ipv6_neighbors_table` was simply missing and fell through to NAPALM's stub; it is implemented against `ip -6 neigh show`, dropping FAILED entries. The rest were stale tests. The DNS fixture put an FQDN where a search domain belongs, which made `get_facts` build "pve1.pve1.example.com" and look like a driver bug. The LLDP fixture was a simplified shape that real `lldpcli show neighbors summary` does not produce — the parser matches on the ", via: LLDP" that follows the interface name. And `test_bridge_vlan_show_parsing` covered a fallback that was replaced by VM-config scanning, asserting an "interfaces" key this method has never returned; it is now a test of the fallback that exists.
335 lines
13 KiB
Python
335 lines
13 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:
|
|
# Real `lldpcli show neighbors summary` output. The interface line carries
|
|
# ", via: LLDP, ..." after the name, which is what the parser matches on —
|
|
# the previous fixture stopped at the name and matched nothing.
|
|
LLDP_SUMMARY = (
|
|
"LLDP neighbors:\n"
|
|
"-------------------------------------------------------------------------------\n"
|
|
"Interface: eth0, via: LLDP, RID: 1, Time: 0 day, 00:11:22\n"
|
|
" Chassis:\n"
|
|
" SysName: sw01.example.com\n"
|
|
" Port:\n"
|
|
" PortID: ifname GigabitEthernet1/0/1\n"
|
|
"-------------------------------------------------------------------------------\n"
|
|
"Interface: eth1, via: LLDP, RID: 2, Time: 0 day, 00:11:22\n"
|
|
" Chassis:\n"
|
|
" SysName: sw02.example.com\n"
|
|
" Port:\n"
|
|
" PortID: ifname GigabitEthernet1/0/2\n"
|
|
"-------------------------------------------------------------------------------\n"
|
|
)
|
|
|
|
def test_neighbors_found(self, driver):
|
|
with patch.object(driver, "_exec_ssh_command", return_value=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()
|