napalm-proxmox/tests/test_fix_templates.py

"""The fixes as napalm-device-types templates, with the caller's values (netOrk #1105).

``fix_snmp`` wrote an application's snmpd.conf (community "public", "Managed by
netOrk") and opened ufw for the caller's /24 guessed from ``ss``;
``install_lldpd`` was an action of its own. Now ``configure_snmp_agent(settings)``
takes every value from the caller, and ``ensure_package`` installs through the
contract's ``install_package``. Every step reads its exit status.
"""

from __future__ import annotations

import re

import pytest

from napalm_proxmox import ProxmoxDriver

SETTINGS = {"community": "c0mm", "allowed_sources": ["10.7.224.20"], "location": "rack 1"}


class _Node:
    """The node's shell: answers by command fragment, with an exit status."""

    def __init__(self, **status):
        self.status = status  # fragment -> exit status
        self.ran: list[str] = []

    def __call__(self, command: str) -> str:
        self.ran.append(command)
        if "SSH_CONNECTION" in command:
            return "10.7.224.20 51234 10.0.0.9 22\n"
        if command.startswith("dpkg-query"):
            return "curl\tinstalled\n"
        rc = next((code for frag, code in self.status.items() if frag in command), 0)
        out = "ok" if rc == 0 else "E: failed"
        return f"{out}\n__PVE_RC={rc}" if "__PVE_RC" in command else out


@pytest.fixture
def node():
    return _Node()


@pytest.fixture
def driver(node):
    d = ProxmoxDriver("pve", "root@pam", "pw")
    d._exec_ssh_command = node
    return d


class TestSnmpAgent:
    def test_install_write_open_start_with_the_callers_values(self, driver, node):
        result = driver.configure_snmp_agent(SETTINGS)

        assert result["success"] is True
        assert [s["step"] for s in result["steps"]] == ["install", "configure", "firewall", "start"]
        assert any("apt-get install -y snmpd snmp" in c for c in node.ran)
        write = next(c for c in node.ran if "/etc/snmp/snmpd.conf" in c)
        assert "install -m 644" in write
        b64 = re.search(r"echo ([A-Za-z0-9+/=]+) \| base64 -d", write).group(1)
        import base64

        conf = base64.b64decode(b64).decode()
        assert "rocommunity c0mm 10.7.224.20/32" in conf
        assert "sysLocation    rack 1" in conf
        assert "public" not in conf and "netork" not in conf.lower()
        assert any("systemctl restart snmpd" in c for c in node.ran)

    def test_no_temp_file_with_an_application_name(self, driver, node):
        driver.configure_snmp_agent(SETTINGS)

        assert not any("netork" in c for c in node.ran)

    def test_ufw_opens_only_for_the_allowed_sources(self, driver, node):
        node.status["command -v ufw"] = 0

        driver.configure_snmp_agent(SETTINGS)

        assert any("ufw allow from 10.7.224.20/32 to any port 161 proto udp" in c for c in node.ran)
        assert not any(".0/24" in c for c in node.ran)

    def test_a_failed_install_stops_the_fix(self, driver, node):
        node.status["apt-get install -y snmpd"] = 100

        result = driver.configure_snmp_agent(SETTINGS)

        assert result["success"] is False
        assert not any("/etc/snmp/snmpd.conf" in c for c in node.ran)

    def test_the_management_peer(self, driver):
        assert driver.get_management_peer() == "10.7.224.20"


class TestPackages:
    def test_ensure_package_installs_what_is_missing(self, driver, node):
        result = driver.ensure_package("lldpd")

        assert result["success"] is True
        assert any("apt-get install -y lldpd" in c for c in node.ran)

    def test_ensure_package_leaves_alone_what_is_there(self, driver, node):
        assert driver.ensure_package("curl")["success"] is True
        assert not any("apt-get install" in c for c in node.ran)

    def test_install_package_raises_when_apt_fails(self, driver, node):
        node.status["apt-get install -y lldpd"] = 100

        with pytest.raises(RuntimeError, match="lldpd"):
            driver.install_package("lldpd")

    def test_a_name_that_is_no_package_name_is_refused(self, driver):
        with pytest.raises(ValueError):
            driver.install_package("lldpd; reboot")


def test_the_old_actions_are_gone():
    for name in ("run_device_action", "_action_fix_snmp", "_action_install_lldpd"):
        assert not hasattr(ProxmoxDriver, name), name


_APT_REPORT = (
    "APROXY_BEGIN\n[dump]\n{dump}[parts]\n{parts}[main]\n[sources]\n"
    "http://download.proxmox.com/debian/pve/dists/trixie/\nAPROXY_END\n"
)


class _AptNode(_Node):
    """A node that answers apt's proxy report and a fetch."""

    def __init__(self, proxy="", parts=""):
        super().__init__()
        self.proxy = proxy
        self.parts = parts

    def __call__(self, command: str) -> str:
        if "apt-config dump" in command:
            self.ran.append(command)
            dump = f'Acquire::http::Proxy "{self.proxy}";\n' if self.proxy else ""
            return _APT_REPORT.format(dump=dump, parts=self.parts)
        if "apt-helper" in command:
            self.ran.append(command)
            return "APROXY_BEGIN\n[fetch]\nGet:1 x\n[rc]\n0\n[match]\n2\nAPROXY_END\n"
        return super().__call__(command)


class TestPackageProxy:
    """Proxmox VE is Debian: apt's proxy is set and read as on any Debian host (netOrk #1397)."""

    def test_it_takes_the_fix(self):
        assert callable(getattr(ProxmoxDriver, "configure_package_proxy", None))

    def test_the_fact(self):
        d = ProxmoxDriver("pve", "root@pam", "pw")
        d._exec_ssh_command = _AptNode(proxy="http://c:3142")

        assert d.get_package_proxy() == "http://c:3142"

    def test_written_as_root_read_back_and_fetched_through(self):
        node = _AptNode(proxy="http://c:3142")
        d = ProxmoxDriver("pve", "root@pam", "pw")
        d._exec_ssh_command = node

        result = d.configure_package_proxy("http://c:3142")

        assert result["success"] is True, result["output"]
        write = next(c for c in node.ran if "/etc/apt/apt.conf.d/00proxy" in c)
        assert "install -m 644" in write
        b64 = re.search(r"echo ([A-Za-z0-9+/=]+) \| base64 -d", write).group(1)
        import base64

        assert base64.b64decode(b64).decode() == 'Acquire::http::Proxy "http://c:3142";\n'
        assert any("dists/trixie/InRelease" in c for c in node.ran)

    def test_another_proxy_is_commented_out(self):
        node = _AptNode(
            proxy="http://c:3142",
            parts='/etc/apt/apt.conf.d/90curtin-aptproxy:1:Acquire::http::Proxy "http://old:3128";\n',
        )
        d = ProxmoxDriver("pve", "root@pam", "pw")
        d._exec_ssh_command = node

        d.configure_package_proxy("http://c:3142")

        assert any(c.startswith("{ sed -i -e '1{") for c in node.ran)

    def test_the_proxy_answers(self):
        d = ProxmoxDriver("pve", "root@pam", "pw")
        d._exec_ssh_command = _AptNode()

        result = d.check_package_proxy("http://c:3142/acng-report.html", expect="Cache")

        assert result == {"reachable": True, "expected": True, "error": ""}