"""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()