Merge fix/snmp-apt-update: refresh apt cache before installing snmpd

napalm_linux/linux.py

@@ -1884,8 +1884,43 @@ class LinuxDriver(OSDriver):
             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}")
 
+    def _action_apt_update_upgrade(self) -> DeviceActionResultDict:
+        """Refresh the apt cache and upgrade all packages (apt-based systems only)."""
+        if self._pkg_manager != "apt":
+            return {
+                "success": True,
+                "output": f"Skipped — package manager is {self._pkg_manager!r}, not apt.",
+            }
+
+        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 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."
+                    ),
+                }
+
+        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 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] = []
@@ -1908,6 +1943,16 @@ class LinuxDriver(OSDriver):
         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",
@@ -1918,8 +1963,12 @@ class LinuxDriver(OSDriver):
         }
         cmd = install_cmd.get(pkg_mgr)
         if cmd:
-            out = self._sudo(cmd, read_timeout=120)
+            try:
-            lines.append(f"[install] {out.strip()[-200:]}")
+                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 = ""

tests/test_linux.py

@@ -661,3 +661,142 @@ def test_get_facts_fallback_vendor_when_dmi_empty(driver):
          patch.object(driver, "_send", side_effect=["host", "host.local", "Alpine Linux 3.19", "eth0"]):
         facts = driver.get_facts()
     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]
+        assert "apt-get 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()