napalm-proxmox/tests/test_misc.py

"""Tests for NTP, SNMP, users, config management, ping, traceroute, CLI."""

from __future__ import annotations

from unittest.mock import patch

import pytest


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