"""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 class TestFactsNotWarnings: """The driver reports facts; the application names warnings (#28, netOrk #1111).""" def test_no_warnings_of_its_own(self): assert "get_device_warnings" not in vars(LinuxDriver) def test_the_package_proxy_apt_fetches_through(self, driver): driver._send = lambda command, read_timeout=100: 'Acquire::http::Proxy "http://c:3142";' assert driver.get_package_proxy() == "http://c:3142" def test_no_proxy_set(self, driver): driver._send = lambda command, read_timeout=100: "" assert driver.get_package_proxy() is None def test_not_an_apt_host(self, driver): driver._pkg_manager = "dnf" assert driver.get_package_proxy() is None def test_the_applications_proxy_url_is_no_argument_any_more(self): d = LinuxDriver("h", "admin", "p", optional_args={"apt_proxy_url": "http://c:3142"}) assert not hasattr(d, "_apt_proxy_url")