diff --git a/napalm_linux/linux.py b/napalm_linux/linux.py index 1d9a5ae..57ff7f7 100644 --- a/napalm_linux/linux.py +++ b/napalm_linux/linux.py @@ -18,6 +18,7 @@ A specific package manager can be forced with from __future__ import annotations +import ipaddress import logging import re import socket @@ -37,12 +38,15 @@ from napalm_device_types import ( ByteStream, CommandResult, ContainerEngineMixin, + EnsurePackageMixin, FingerprintRule, HostStatusMixin, KernelFactsMixin, ListeningSocketsMixin, OpenSshServerMixin, OSDriver, + PackageProxyMixin, + SnmpAgentMixin, SystemIdentityMixin, SystemdServicesMixin, SystemdUnavailable, @@ -50,17 +54,20 @@ from napalm_device_types import ( parse_apt_upgradable, open_stream_on_transport, parse_dnf_security, + run_fix_steps, run_on_transport, strip_terminal_codes, ) from napalm_device_types.models import ( ApplyUpdatesResultDict, CronJobDict, - DeviceActionResultDict, + FixResultDict, + FixStepOutcomeDict, PackageDict, ProcessDict, ServiceDict, SNMPConfigDict, + SnmpAgentSettingsDict, UpdateDict, UserDict, ) @@ -171,9 +178,8 @@ _STAGE = ( #: What to do when sudo wants a password netOrk does not have. _SUDO_PASSWORD_HINT = ( - "sudo requires a password on this device but none is configured in netOrk. " - "Please add the sudo password to a Credential Profile assigned to this device, " - "or configure passwordless sudo (NOPASSWD) for this user." + "sudo requires a password on this host and no sudo password was given: pass one, " + "or allow this user passwordless sudo (NOPASSWD)." ) # DMI field values that carry no useful information (OEM defaults, blanks) @@ -242,7 +248,49 @@ def _arm_vendor_from_model(model: str) -> str: return " ".join(brand) +#: Where net-snmp reads its configuration, and how each package manager installs it. +_SNMPD_CONF = "/etc/snmp/snmpd.conf" +_SNMP_INSTALL = { + "apt": "apt-get update -y >/dev/null 2>&1; DEBIAN_FRONTEND=noninteractive apt-get install -y snmpd snmp 2>&1", + "dnf": "dnf install -y net-snmp net-snmp-utils 2>&1", + "yum": "yum install -y net-snmp net-snmp-utils 2>&1", + "apk": "apk add --no-cache net-snmp net-snmp-tools 2>&1", + "pacman": "pacman -Sy --noconfirm net-snmp 2>&1", +} +_APT_PROXY_CONF = "/etc/apt/apt.conf.d/00proxy" +_PROXY_URL = re.compile(r"^https?://[A-Za-z0-9.\-\[\]:@%_~/]+$") +_USER_NAME = re.compile(r"^[a-z_][a-z0-9_.\-]{0,31}$") +#: What may stand in an snmpd.conf value: no whitespace, no quote, no comment. +_SNMP_WORD = re.compile(r"^[^\s\"#]+$") + + +def _source(value: str) -> str: + """An allowed source as an address or network, or ValueError.""" + return str(ipaddress.ip_network(value.strip(), strict=False)) + + +def render_snmpd_conf(settings: SnmpAgentSettingsDict) -> str: + """net-snmp's configuration for *settings*: read-only v2c, per allowed source.""" + community = settings["community"] + if not _SNMP_WORD.match(community): + raise ValueError("the community must be one word without quotes or #") + listen = settings.get("listen") + lines = [f"agentAddress udp:{listen + ':' if listen else ''}161"] + sources = [_source(s) for s in settings.get("allowed_sources", [])] + lines += [f"rocommunity {community} {s}" for s in sources] or [f"rocommunity {community}"] + for key, directive in (("location", "sysLocation"), ("contact", "sysContact")): + value = settings.get(key, "") + if value: + if "\n" in value: + raise ValueError(f"{key} must be one line") + lines.append(f"{directive} {value}") + return "\n".join(lines) + "\n" + + class LinuxDriver( + SnmpAgentMixin, + EnsurePackageMixin, + PackageProxyMixin, KernelFactsMixin, ListeningSocketsMixin, VirtualIpsMixin, @@ -1940,212 +1988,81 @@ class LinuxDriver( logger.debug("get_snmp_config() failed: %s", exc) return None - def run_device_action(self, action: str) -> DeviceActionResultDict: - """Execute a named action on the device.""" - if action == "fix_docker_permissions": - return self._action_fix_docker_permissions() - if action == "fix_snmp": - return self._action_fix_snmp() - if action == "fix_apt_proxy": - return self._action_fix_apt_proxy() - if action == "apt_update_upgrade": - return self._action_apt_update_upgrade() - raise NotImplementedError(f"Unknown action: {action!r}") + # ------------------------------------------------------------------ + # Fix templates (napalm-device-types ``fixes``): the caller supplies every value + # ------------------------------------------------------------------ - def _action_apt_update_upgrade(self) -> DeviceActionResultDict: - """Refresh the apt cache and fully upgrade all packages (apt-based systems only). + def _root_outcome(self, command: str, timeout: float, done: str) -> FixStepOutcomeDict: + """Run *command* as root and report it as a step: its exit status decides.""" + output, status = self._sudo_status(command, timeout) + if status != 0 and "password is required" in output: + output = f"{output}\n{_SUDO_PASSWORD_HINT}" + return {"ok": status == 0, "output": (output.strip()[-300:] if status != 0 else done)} - Uses full-upgrade (not plain upgrade) — plain "apt-get upgrade" refuses - to install/remove packages even when required to satisfy a newer - version's dependencies, silently leaving those updates pending. - """ - if self._pkg_manager != "apt": - return { - "success": True, - "output": f"Skipped — package manager is {self._pkg_manager!r}, not apt.", - } + def _install_snmp_agent(self) -> FixStepOutcomeDict: + pm = self._pkg_manager or self._detect_pkg_manager() + command = _SNMP_INSTALL.get(pm or "") + if not command: + return {"ok": False, "output": f"no supported package manager ({pm or 'none found'})"} + return self._root_outcome(command, 300, f"net-snmp installed ({pm})") - sudo_check = self._send("sudo -n true 2>&1 || echo __SUDO_NEEDS_PW__") - if "__SUDO_NEEDS_PW__" in sudo_check or "password is required" in sudo_check.lower(): - if not self._sudo_password: - return { - "success": False, - "output": _SUDO_PASSWORD_HINT, - } - - lines: list[str] = [] - try: - out = self._sudo("apt-get update -y 2>&1", read_timeout=90) - lines.append(f"[update] {out.strip()[-300:]}") - out = self._sudo( - "DEBIAN_FRONTEND=noninteractive apt-get full-upgrade -y 2>&1", read_timeout=240 - ) - lines.append(f"[upgrade] {out.strip()[-300:]}") - return {"success": True, "output": "\n".join(lines)} - except Exception as exc: - lines.append(f"[error] {exc}") - return {"success": False, "output": "\n".join(lines)} - - def _action_fix_snmp(self) -> DeviceActionResultDict: - """Install, configure and start snmpd with community 'public'.""" - lines: list[str] = [] - - # 0. Verify sudo access before attempting anything - sudo_check = self._send("sudo -n true 2>&1 || echo __SUDO_NEEDS_PW__") - if "__SUDO_NEEDS_PW__" in sudo_check or "password is required" in sudo_check.lower(): - if not self._sudo_password: - return { - "success": False, - "output": _SUDO_PASSWORD_HINT, - } - - # 1. Install snmpd if missing - pkg_mgr = self._detect_pkg_manager() - if not pkg_mgr: - return {"success": False, "output": "Package manager not detected — cannot install snmpd."} - - # Refresh the package index first — a freshly provisioned (or simply - # long-untouched) system's cache can be stale/empty, which makes the - # install below fail outright rather than just being slow. - if pkg_mgr == "apt": - try: - update_out = self._sudo("apt-get update -y 2>&1", read_timeout=90) - lines.append(f"[update] {update_out.strip()[-200:]}") - except Exception as exc: - lines.append(f"[warn] apt-get update failed: {exc}") - - # Install both snmpd (daemon) and snmp (client tools incl. snmpget for probing) - install_cmd: dict[str, str] = { - "apt": "DEBIAN_FRONTEND=noninteractive apt-get install -y snmpd snmp 2>&1", - "dnf": "dnf install -y net-snmp net-snmp-utils 2>&1", - "yum": "yum install -y net-snmp net-snmp-utils 2>&1", - "apk": "apk add --no-cache net-snmp net-snmp-tools 2>&1", - "pacman": "pacman -Sy --noconfirm net-snmp 2>&1", - } - cmd = install_cmd.get(pkg_mgr) - if cmd: - try: - out = self._sudo(cmd, read_timeout=120) - lines.append(f"[install] {out.strip()[-200:]}") - except Exception as exc: - lines.append(f"[error] install failed: {exc}") - return {"success": False, "output": "\n".join(lines)} - - # 2. Determine the IP netOrk is connecting from by checking the established SSH connection - netork_ip = "" - try: - # ss shows the remote peer of the current SSH connection - raw = self._send( - "ss -tnp 2>/dev/null | awk '/sshd/{print $5}' | head -1 | cut -d: -f1" - ).strip() - if raw and raw not in ("", "0.0.0.0", "::", "127.0.0.1"): - netork_ip = raw - except Exception: - pass - - # Write snmpd.conf: staged unprivileged, installed by root (#1039). - # agentAddress udp:161 overrides Debian's localhost-only default. - conf_str = ( - "agentAddress udp:161\n" - "rocommunity public\n" - "sysLocation Managed by netOrk\n" - "sysContact netork@localhost\n" - ) - try: - written = self._install_file_as_root("/etc/snmp/snmpd.conf", conf_str) - install_error = written.stdout.strip() if written.exit_code else "" - except Exception as exc: - install_error = str(exc) - verify = self._send("cat /etc/snmp/snmpd.conf 2>/dev/null").strip() - if "agentAddress" in verify and "rocommunity" in verify: - lines.append("[config] Wrote /etc/snmp/snmpd.conf — agentAddress udp:161, rocommunity public.") - else: - lines.append(f"[warn] snmpd.conf write may have failed: {(install_error or verify)[:100]}") - - # 3. Open firewall for SNMP (UDP 161) — restrict to netOrk's source IP - if netork_ip: - try: - ufw = self._send("command -v ufw 2>/dev/null").strip() - ipt = self._send("command -v iptables 2>/dev/null").strip() - if ufw: - # Expand to /24 so all containers in the same Docker network can probe - parts = netork_ip.rsplit(".", 1) - subnet = f"{parts[0]}.0/24" if len(parts) == 2 else netork_ip - fw_out = self._sudo( - f"ufw allow from {subnet} to any port 161 proto udp 2>&1", read_timeout=10 - ) - lines.append(f"[firewall/ufw] {fw_out.strip()[:200]}") - elif ipt: - fw_out = self._sudo( - f"iptables -C INPUT -s {netork_ip} -p udp --dport 161 -j ACCEPT 2>/dev/null" - f" || iptables -I INPUT -s {netork_ip} -p udp --dport 161 -j ACCEPT", - read_timeout=10, - ) - lines.append(f"[firewall/iptables] rule added for {netork_ip}:161/udp") - except Exception as exc: - lines.append(f"[firewall] skipped — {exc}") - - # 4. Restart snmpd. - # - Redirect all output to /dev/null so netmiko's prompt detection is - # never confused by service status messages. - # - Append "; echo __OK__" so there is always a known token to wait for. - import time as _time - # Stop any running snmpd (systemctl-managed or apt-started orphan) - self._sudo("systemctl stop snmpd >/dev/null 2>&1; echo s1", read_timeout=15) - self._sudo("pkill -9 snmpd >/dev/null 2>&1; echo s2", read_timeout=10) - _time.sleep(2) - # Enable and start fresh - self._sudo("systemctl enable snmpd >/dev/null 2>&1; echo s3", read_timeout=15) - self._sudo("systemctl start snmpd >/dev/null 2>&1; echo s4", read_timeout=20) - _time.sleep(2) - lines.append("[service] snmpd restarted.") - - # 5. Verify snmpd responds via local SNMP probe (sysDescr.0). - # Success requires actual SNMP data types in the output, not just - # the absence of error keywords. - _time.sleep(2) - probe_out = self._send( - "snmpget -v2c -cpublic -t2 -r0 -Ov 127.0.0.1 1.3.6.1.2.1.1.1.0 2>&1 || true" - ).strip() - # snmpget returns lines like "STRING: Linux ..." or "Timeticks: (n) ..." - _snmp_types = ("STRING:", "INTEGER:", "OID:", "Timeticks:", "Hex-STRING:", "IpAddress:") - success = any(t in probe_out for t in _snmp_types) - if success: - lines.append(f"[ok] SNMP probe successful — community 'public' is working.") - else: - lines.append(f"[warn] SNMP probe failed — output: {probe_out[:200]}") - - return {"success": success, "output": "\n".join(lines)} - - def _action_fix_docker_permissions(self) -> dict[str, Any]: - """Add the SSH user to the 'docker' group via sudo usermod.""" - user = self._send("whoami 2>/dev/null || id -un").strip().splitlines()[-1].strip() - out = self._sudo(f"usermod -aG docker {user}") - low = out.lower() - success = not any(kw in low for kw in ("error", "invalid", "no such", "command not found")) - if not out.strip(): - out = f"Added {user!r} to the docker group. Reconnect or run a new poll to verify." - return {"success": success, "output": out} - - def _action_fix_apt_proxy(self) -> DeviceActionResultDict: - """Write /etc/apt/apt.conf.d/00proxy with the configured proxy URL.""" - proxy_url = self._apt_proxy_url - if not proxy_url: - return {"success": False, "output": "No apt_proxy_url configured."} - if self._pkg_manager != "apt": - return {"success": False, "output": f"Package manager is {self._pkg_manager!r}, not apt — skipping."} - - content = f'Acquire::http::Proxy "{proxy_url}";\n' - # Staged unprivileged, installed by root (#1039). - written = self._install_file_as_root("/etc/apt/apt.conf.d/00proxy", content) - verify = self._send("cat /etc/apt/apt.conf.d/00proxy 2>/dev/null").strip() - success = proxy_url in verify - if success: - return {"success": True, "output": f"Wrote /etc/apt/apt.conf.d/00proxy — proxy: {proxy_url}"} - output = f"Write may have failed. File content: {verify[:200]}" + def _write_snmp_agent_config(self, settings: SnmpAgentSettingsDict) -> FixStepOutcomeDict: + written = self._install_file_as_root(_SNMPD_CONF, render_snmpd_conf(settings)) if written.exit_code: - output = f"{output}\n{written.stdout.strip()}" - return {"success": False, "output": output} + return {"ok": False, "output": (written.stdout or written.stderr).strip()[-300:]} + return {"ok": True, "output": f"wrote {_SNMPD_CONF}"} + + def _allow_snmp_sources(self, settings: SnmpAgentSettingsDict) -> FixStepOutcomeDict: + sources = [_source(s) for s in settings.get("allowed_sources", [])] + if not sources: + return {"ok": True, "output": "no sources given; firewall unchanged"} + if self._send("command -v ufw 2>/dev/null"): + rules = [f"ufw allow from {s} to any port 161 proto udp" for s in sources] + elif self._send("command -v iptables 2>/dev/null"): + rules = [ + f"{{ iptables -C INPUT -s {s} -p udp --dport 161 -j ACCEPT 2>/dev/null" + f" || iptables -I INPUT -s {s} -p udp --dport 161 -j ACCEPT; }}" + for s in sources + ] + else: + return {"ok": True, "output": "no host firewall (ufw, iptables); nothing to open"} + return self._root_outcome(" && ".join(rules), 30, f"udp/161 open for {', '.join(sources)}") + + def _start_snmp_agent(self) -> FixStepOutcomeDict: + return self._root_outcome( + "systemctl enable snmpd && systemctl restart snmpd && systemctl is-active snmpd", + 60, + "snmpd enabled and running", + ) + + def configure_package_proxy(self, url: str) -> FixResultDict: + """apt fetches through *url* (``/etc/apt/apt.conf.d/00proxy``); ``""`` removes it.""" + + def write() -> FixStepOutcomeDict: + if self._pkg_manager != "apt": + return {"ok": False, "output": f"package manager is {self._pkg_manager!r}, not apt"} + if not url: + return self._root_outcome(f"rm -f {_APT_PROXY_CONF}", 30, "proxy removed") + if not _PROXY_URL.match(url): + raise ValueError(f"not a proxy URL: {url!r}") + written = self._install_file_as_root(_APT_PROXY_CONF, f'Acquire::http::Proxy "{url}";\n') + if written.exit_code: + return {"ok": False, "output": (written.stdout or written.stderr).strip()[-300:]} + return {"ok": True, "output": f"apt fetches through {url}"} + + return run_fix_steps([("configure", write)]) + + def grant_container_engine_access(self, user: str) -> FixResultDict: + """Add *user* to the ``docker`` group, so the engine answers without sudo.""" + + def grant() -> FixStepOutcomeDict: + if not _USER_NAME.match(user): + raise ValueError(f"not a user name: {user!r}") + return self._root_outcome( + f"usermod -aG docker {user}", 30, f"{user} is in the docker group from the next login" + ) + + return run_fix_steps([("grant", grant)]) def get_vpn_tunnels(self) -> dict[str, Any]: """Return WireGuard status via ``wg show all dump`` (requires root/sudo). diff --git a/tests/test_fix_templates.py b/tests/test_fix_templates.py new file mode 100644 index 0000000..60aadc8 --- /dev/null +++ b/tests/test_fix_templates.py @@ -0,0 +1,182 @@ +"""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 diff --git a/tests/test_linux.py b/tests/test_linux.py index 4f0cbce..33c4876 100644 --- a/tests/test_linux.py +++ b/tests/test_linux.py @@ -807,151 +807,10 @@ def test_get_facts_fallback_vendor_when_dmi_empty(driver): assert facts["vendor"] == "Linux" # fallback to VENDOR class attribute -# --------------------------------------------------------------------------- -# _action_fix_snmp / _action_apt_update_upgrade -# --------------------------------------------------------------------------- - - -def _fix_snmp_send_side_effect(command: str, read_timeout: float = 100) -> str: - """Canned responses covering every _send() call _action_fix_snmp makes.""" - if command.startswith("sudo -n true"): - return "" # passwordless sudo works - if command.startswith("command -v apt"): - return "/usr/bin/apt" - if command.startswith("command -v"): - return "" # ufw/iptables not found, other pkg managers not found - if command.startswith("ss -tnp"): - return "" # no netork_ip detected — skips firewall step - if command.startswith("cat /etc/snmp/snmpd.conf"): - return "agentAddress udp:161\nrocommunity public\n" - if command.startswith("snmpget"): - return "STRING: Linux test" - return "" - - -class TestActionFixSnmp: - def test_runs_apt_get_update_before_install(self, driver): - driver._pkg_manager = "apt" - driver._sudo_password = None - sudo_calls: list[str] = [] - - def _sudo_side_effect(command: str, read_timeout: float = 100) -> str: - sudo_calls.append(command) - return "" - - with ( - patch.object(driver, "_send", side_effect=_fix_snmp_send_side_effect), - patch.object(driver, "_sudo", side_effect=_sudo_side_effect), - ): - driver._action_fix_snmp() - - update_idx = next(i for i, c in enumerate(sudo_calls) if "apt-get update" in c) - install_idx = next(i for i, c in enumerate(sudo_calls) if "apt-get install" in c) - assert update_idx < install_idx - - def test_install_exception_returns_failure_instead_of_raising(self, driver): - driver._pkg_manager = "apt" - driver._sudo_password = None - - def _sudo_side_effect(command: str, read_timeout: float = 100) -> str: - if "apt-get install" in command: - raise TimeoutError("connection timed out") - return "" - - with ( - patch.object(driver, "_send", side_effect=_fix_snmp_send_side_effect), - patch.object(driver, "_sudo", side_effect=_sudo_side_effect), - ): - result = driver._action_fix_snmp() - - assert result["success"] is False - assert "install failed" in result["output"] - - def test_apt_get_update_failure_is_non_fatal(self, driver): - """apt-get update failing (e.g. transient network issue) must not - abort the whole action — install is still attempted.""" - driver._pkg_manager = "apt" - driver._sudo_password = None - - def _sudo_side_effect(command: str, read_timeout: float = 100) -> str: - if "apt-get update" in command: - raise TimeoutError("network unreachable") - return "" - - with ( - patch.object(driver, "_send", side_effect=_fix_snmp_send_side_effect), - patch.object(driver, "_sudo", side_effect=_sudo_side_effect), - ): - result = driver._action_fix_snmp() - - assert "apt-get update failed" in result["output"] - - -class TestActionAptUpdateUpgrade: - def test_skips_when_not_apt(self, driver): - driver._pkg_manager = "dnf" - result = driver._action_apt_update_upgrade() - assert result["success"] is True - assert "Skipped" in result["output"] - - def test_fails_when_sudo_needs_password_and_none_configured(self, driver): - driver._pkg_manager = "apt" - driver._sudo_password = None - with patch.object(driver, "_send", return_value="__SUDO_NEEDS_PW__"): - result = driver._action_apt_update_upgrade() - assert result["success"] is False - assert "sudo requires a password" in result["output"] - - def test_runs_update_then_upgrade_successfully(self, driver): - driver._pkg_manager = "apt" - driver._sudo_password = "secret" # noqa: S105 - with ( - patch.object(driver, "_send", return_value=""), - patch.object( - driver, - "_sudo", - side_effect=["Reading package lists... Done", "0 upgraded, 0 newly installed"], - ) as mock_sudo, - ): - result = driver._action_apt_update_upgrade() - - assert result["success"] is True - assert "[update]" in result["output"] - assert "[upgrade]" in result["output"] - update_call, upgrade_call = mock_sudo.call_args_list - assert "apt-get update" in update_call.args[0] - # full-upgrade (not plain upgrade) — plain upgrade silently holds back - # packages whose newer version needs to install/remove dependencies. - assert "apt-get full-upgrade" in upgrade_call.args[0] - - def test_exception_during_upgrade_returns_failure(self, driver): - driver._pkg_manager = "apt" - driver._sudo_password = "secret" # noqa: S105 - with ( - patch.object(driver, "_send", return_value=""), - patch.object( - driver, "_sudo", side_effect=["update ok", TimeoutError("connection lost")] - ), - ): - result = driver._action_apt_update_upgrade() - - assert result["success"] is False - assert "[error]" in result["output"] - - -class TestRunDeviceActionDispatch: - def test_apt_update_upgrade_action_dispatches(self, driver): - with patch.object( - driver, "_action_apt_update_upgrade", return_value={"success": True, "output": ""} - ) as mock_action: - driver.run_device_action("apt_update_upgrade") - mock_action.assert_called_once() - - class TestNoDockerOfItsOwn: """Containers are read and handled by netOrk over the Engine API, through ``open_container_engine`` (NetOrk/netork#765). The old CLI methods are gone; - only the OS action that lets the login user use Docker stays.""" + only the fix that lets the login user use Docker stays.""" def test_the_cli_methods_are_gone(self): for name in ("get_docker_info", "get_docker_outdated", "reconstruct_docker_run", "_docker_bin"): @@ -959,7 +818,7 @@ class TestNoDockerOfItsOwn: def test_the_engine_access_and_the_permission_fix_stay(self): assert hasattr(LinuxDriver, "open_container_engine") - assert hasattr(LinuxDriver, "_action_fix_docker_permissions") + assert callable(LinuxDriver.grant_container_engine_access) class TestUninstallPackage: diff --git a/tests/test_staged_root_writes.py b/tests/test_staged_root_writes.py index 91b1f33..ae3eec1 100644 --- a/tests/test_staged_root_writes.py +++ b/tests/test_staged_root_writes.py @@ -138,10 +138,16 @@ def _every_line(driver: LinuxDriver, host: Host) -> list[str]: return typed + [run.line for run in host.runs] +_SETTINGS = {"community": "c0mm", "allowed_sources": []} + ACTIONS = [ - ("_action_fix_snmp", "/etc/snmp/snmpd.conf", b"agentAddress udp:161"), ( - "_action_fix_apt_proxy", + lambda driver: driver.configure_snmp_agent(_SETTINGS), + "/etc/snmp/snmpd.conf", + b"agentAddress udp:161", + ), + ( + lambda driver: driver.configure_package_proxy("http://proxy:3142"), "/etc/apt/apt.conf.d/00proxy", b'Acquire::http::Proxy "http://proxy:3142";', ), @@ -160,7 +166,7 @@ class TestStagedWrite: host = Host() driver = _driver(host) - getattr(driver, action)() + action(driver) for planted in _PLANTED: assert not any(planted in line for line in _every_line(driver, host)) @@ -171,7 +177,7 @@ class TestStagedWrite: host = Host() driver = _driver(host) - getattr(driver, action)() + action(driver) stage = next(run for run in host.unprivileged if "mktemp" in run.line) assert "umask 077" in stage.line @@ -182,7 +188,7 @@ class TestStagedWrite: host = Host() driver = _driver(host) - getattr(driver, action)() + action(driver) install = next(script for script in host.root_scripts if "install -m" in script) part = f"{target}.netork-tmp" @@ -194,7 +200,7 @@ class TestStagedWrite: host = Host() driver = _driver(host) - getattr(driver, action)() + action(driver) assert not any(re.search(r"\bch(own|mod)\b", script) for script in host.root_scripts) @@ -202,7 +208,7 @@ class TestStagedWrite: host = Host(stage_output="mktemp: No space left on device\n", stage_status=1) driver = _driver(host) - getattr(driver, action)() + action(driver) assert not any("install -m" in script for script in host.root_scripts) @@ -210,7 +216,7 @@ class TestStagedWrite: host = Host(stage_output="/etc/shadow") driver = _driver(host) - getattr(driver, action)() + action(driver) assert not any("/etc/shadow" in script for script in host.root_scripts) @@ -223,7 +229,7 @@ def test_a_failed_install_is_reported(): driver = _driver(host) driver._device.send_command.side_effect = lambda command, **_k: "" - result = driver._action_fix_apt_proxy() + result = driver.configure_package_proxy("http://proxy:3142") assert result["success"] is False assert "install: cannot create" in result["output"]