napalm-linux/tests/test_fix_templates.py

"""The fix templates (napalm-device-types ``fixes``): the caller supplies every value.

They replace ``run_device_action``. The driver knows how each step is done on a
Linux host; the community, the allowed sources, the proxy and the user are the
caller's (napalm-device-types docs/STRUCTURE.md).
"""

from __future__ import annotations

from unittest.mock import MagicMock

import pytest

from napalm_linux import LinuxDriver
from napalm_linux.linux import render_snmpd_conf


class TestSnmpdConf:
    def test_one_read_only_line_per_allowed_source(self):
        conf = render_snmpd_conf(
            {
                "community": "c0mm",
                "allowed_sources": ["10.0.0.5", "10.1.0.0/24"],
                "contact": "ops",
                "location": "rack 1",
            }
        )

        assert conf == (
            "agentAddress udp:161\n"
            "rocommunity c0mm 10.0.0.5/32\n"
            "rocommunity c0mm 10.1.0.0/24\n"
            "sysLocation    rack 1\n"
            "sysContact    ops\n"
        )

    def test_without_sources_every_address_may_ask(self):
        assert "rocommunity c0mm\n" in render_snmpd_conf(
            {"community": "c0mm", "allowed_sources": []}
        )

    def test_a_listen_address_binds_the_agent(self):
        conf = render_snmpd_conf({"community": "c", "allowed_sources": [], "listen": "10.0.0.1"})
        assert conf.startswith("agentAddress udp:10.0.0.1:161\n")

    def test_no_application_name_is_written(self):
        conf = render_snmpd_conf({"community": "c", "allowed_sources": []})
        assert "netork" not in conf.lower()

    @pytest.mark.parametrize("community", ["two words", 'quo"te', "has#hash", ""])
    def test_a_community_that_would_break_the_file_is_refused(self, community):
        with pytest.raises(ValueError):
            render_snmpd_conf({"community": community, "allowed_sources": []})

    def test_a_source_that_is_no_address_is_refused(self):
        with pytest.raises(ValueError):
            render_snmpd_conf({"community": "c", "allowed_sources": ["10.0.0.5; rm -rf /"]})


@pytest.fixture
def driver():
    d = LinuxDriver("h", "admin", "p")
    d._device = MagicMock()
    d._pkg_manager = "apt"
    d._root = True
    return d


class TestSnmpAgentSteps:
    SETTINGS = {"community": "c0mm", "allowed_sources": ["10.0.0.5"]}

    def _record(self, driver, status=0, firewall="ufw"):
        ran: list[str] = []

        def sudo_status(command, read_timeout=100):
            ran.append(command)
            return ("", status)

        driver._sudo_status = sudo_status
        driver._send = lambda command, read_timeout=100: (
            f"/usr/sbin/{firewall}" if command == f"command -v {firewall} 2>/dev/null" else ""
        )
        driver._install_file_as_root = MagicMock(
            return_value=MagicMock(exit_code=0, stdout="", stderr="")
        )
        return ran

    def test_installs_writes_opens_and_starts(self, driver):
        ran = self._record(driver)

        result = driver.configure_snmp_agent(self.SETTINGS)

        assert result["success"] is True
        assert [s["step"] for s in result["steps"]] == ["install", "configure", "firewall", "start"]
        assert "apt-get install -y snmpd snmp" in ran[0]
        path, content = driver._install_file_as_root.call_args.args
        assert path == "/etc/snmp/snmpd.conf" and "rocommunity c0mm 10.0.0.5/32" in content
        assert ran[1] == "ufw allow from 10.0.0.5/32 to any port 161 proto udp"
        assert "systemctl restart snmpd" in ran[2]

    def test_the_firewall_opens_only_for_the_given_sources_with_iptables(self, driver):
        ran = self._record(driver, firewall="iptables")

        driver.configure_snmp_agent(self.SETTINGS)

        assert "iptables -I INPUT -s 10.0.0.5/32 -p udp --dport 161 -j ACCEPT" in ran[1]

    def test_no_sources_change_no_firewall(self, driver):
        ran = self._record(driver)

        result = driver.configure_snmp_agent({"community": "c0mm", "allowed_sources": []})

        assert result["success"] is True
        assert not any("ufw" in c or "iptables" in c for c in ran)

    def test_a_failed_install_stops_the_fix(self, driver):
        ran = self._record(driver, status=100)

        result = driver.configure_snmp_agent(self.SETTINGS)

        assert result["success"] is False
        assert len(ran) == 1
        driver._install_file_as_root.assert_not_called()


class TestPackageProxyAndEngineAccess:
    def test_the_proxy_is_the_caller_s(self, driver):
        driver._install_file_as_root = MagicMock(
            return_value=MagicMock(exit_code=0, stdout="", stderr="")
        )

        result = driver.configure_package_proxy("http://cache.lan:3142")

        assert result["success"] is True
        path, content = driver._install_file_as_root.call_args.args
        assert (path, content) == (
            "/etc/apt/apt.conf.d/00proxy",
            'Acquire::http::Proxy "http://cache.lan:3142";\n',
        )

    def test_a_proxy_url_that_would_break_the_file_is_refused(self, driver):
        driver._install_file_as_root = MagicMock()

        result = driver.configure_package_proxy('http://x"; Acquire::x "y')

        assert result["success"] is False
        driver._install_file_as_root.assert_not_called()

    def test_only_apt_takes_a_proxy(self, driver):
        driver._pkg_manager = "dnf"

        assert driver.configure_package_proxy("http://cache.lan:3142")["success"] is False

    def test_engine_access_for_the_user_the_caller_names(self, driver):
        ran = []
        driver._sudo_status = lambda command, read_timeout=100: ran.append(command) or ("", 0)

        result = driver.grant_container_engine_access("deploy")

        assert result["success"] is True
        assert ran == ["usermod -aG docker deploy"]

    def test_a_user_name_that_is_no_user_name_is_refused(self, driver):
        driver._sudo_status = MagicMock()

        assert driver.grant_container_engine_access("x; reboot")["success"] is False
        driver._sudo_status.assert_not_called()


def test_run_device_action_is_gone():
    assert (
        not hasattr(LinuxDriver, "run_device_action")
        or LinuxDriver.run_device_action is None
        or ("run_device_action" not in vars(LinuxDriver))
    )
    for name in (
        "_action_fix_snmp",
        "_action_fix_apt_proxy",
        "_action_fix_docker_permissions",
        "_action_apt_update_upgrade",
    ):
        assert not hasattr(LinuxDriver, name), name


def test_every_annotation_resolves():
    """``Dict``, ``List`` and ``Optional`` were used without being imported (#29);
    ``from __future__ import annotations`` hid it until something resolved them."""
    import inspect
    import typing

    for name, member in inspect.getmembers(LinuxDriver, inspect.isfunction):
        if member.__module__ == "napalm_linux.linux":
            typing.get_type_hints(member)


class TestManagementPeer:
    """The address the host sees this connection come from (``SSH_CONNECTION``)."""

    def _driver(self, answer):
        d = LinuxDriver("h", "admin", "p")
        d._device = MagicMock()
        d.run_command = MagicMock(return_value=MagicMock(stdout=answer, exit_code=0))
        return d

    def test_the_client_side_of_ssh_connection(self):
        assert self._driver("10.7.224.20 51234 10.0.0.5 22\n").get_management_peer() == "10.7.224.20"

    def test_an_ipv6_client(self):
        assert self._driver("fd00::20 51234 fd00::5 22\n").get_management_peer() == "fd00::20"

    @pytest.mark.parametrize("answer", ["", "\n", "garbage", "not-an-ip 1 2 3"])
    def test_nothing_usable_is_none(self, answer):
        assert self._driver(answer).get_management_peer() is None