napalm_proxmox/driver.py

@@ -36,6 +36,7 @@ logger = logging.getLogger(__name__)
 
 from napalm_device_types import (
     FingerprintRule,
+    HostStatusMixin,
     HypervisorDriver,
     KernelFactsMixin,
     PortSpec,
@@ -84,6 +85,7 @@ class ProxmoxDriver(
     ProxmoxSystemMixin,
     KernelFactsMixin,
     SystemdServicesMixin,
+    HostStatusMixin,
     HypervisorDriver,
 ):
     """NAPALM driver for Proxmox VE nodes."""

napalm_proxmox/system_mixin.py

@@ -20,6 +20,8 @@ import logging
 import re
 from typing import Any
 
+from napalm_device_types import APT_UPGRADABLE_COMMAND, parse_apt_upgradable
+
 from napalm_proxmox import utils
 
 logger = logging.getLogger(__name__)
@@ -362,22 +364,47 @@ class ProxmoxSystemMixin:
     # ------------------------------------------------------------------ #
 
     def get_available_updates(self) -> list[_JsonDict]:
-        """Return list of upgradable packages from the Proxmox APT API."""
+        """Return the node's upgradable packages, with origin and security status.
-        updates: list[_JsonDict] = []
+
+        ``apt list --upgradable`` over the exec path names each candidate's suite
+        (``trixie-security``); the APT API names only an Origin ("Debian",
+        "Proxmox") and is the fallback, with the security status unknown.
+
+        :raises Exception: when neither answers -- never an empty list for
+            "could not read".
+        """
         try:
-            for upd in self._api.nodes(self._node_name).apt.update.get():
+            updates = parse_apt_upgradable(self._exec_ssh_command(APT_UPGRADABLE_COMMAND) or "")
-                pkg = upd.get("Package", "")
+        except ValueError as exc:
-                if not pkg:
+            logger.debug("apt list over the exec path failed, using the API: %s", exc)
-                    continue
+            updates = self._updates_from_api()
-                updates.append({
-                    "name": pkg,
-                    "current_version": upd.get("OldVersion", ""),
-                    "new_version": upd.get("Version", ""),
-                })
-        except Exception as exc:
-            logger.debug("Failed to fetch available updates: %s", exc)
         return sorted(updates, key=lambda u: u["name"])
 
+    def _updates_from_api(self) -> list[_JsonDict]:
+        return [
+            {
+                "name": upd["Package"],
+                "current_version": upd.get("OldVersion", ""),
+                "new_version": upd.get("Version", ""),
+                "origin": upd.get("Origin"),
+                "security": None,
+            }
+            for upd in self._api.nodes(self._node_name).apt.update.get()  # type: ignore[union-attr]
+            if upd.get("Package")
+        ]
+
+    def refresh_available_updates(self) -> _JsonDict:
+        """Resynchronise the node's package index (``POST nodes/{n}/apt/update``)."""
+        try:
+            task = self._api.nodes(self._node_name).apt.update.post()  # type: ignore[union-attr]
+        except Exception as exc:
+            return {"success": False, "output": str(exc)}
+        return {"success": True, "output": f"Package index refresh started ({task})"}
+
+    def _run_host_status_command(self, command: str) -> str:
+        """The transport for ``HostStatusMixin.get_host_status``: the exec path."""
+        return str(self._exec_ssh_command(command))
+
     def apply_updates(self, packages: list[str]) -> _JsonDict:
         """Upgrade the given packages via ``apt-get install`` over SSH."""
         for pkg in packages:

pyproject.toml

@@ -25,7 +25,7 @@ classifiers = [
 requires-python = ">=3.9"
 dependencies = [
     "napalm>=5.0.0",
-    "napalm_device_types>=2.2.0",
+    "napalm_device_types>=2.3.0",
     "paramiko>=5.0.0",  # CVE-2026-44405; imported directly for SSH fallback (driver.py)
     "proxmoxer>=2.0.0",
     "netaddr>=0.9.0",

tests/test_updates.py

@@ -0,0 +1,120 @@
+"""Pending updates on a Proxmox node: from where, whether they are security fixes,
+and whether the node needs a reboot.
+
+The node's APT API names only an Origin ("Debian", "Proxmox"), the same for the
+main and the security archive. ``apt list --upgradable`` over the exec path
+names the suite (``trixie-security``), so that is read first; the API remains
+the fallback, with the security status left unknown. A reader that cannot read
+raises: an empty list would tell netOrk that nothing is pending.
+"""
+
+from __future__ import annotations
+
+from unittest.mock import MagicMock, patch
+
+import pytest
+from napalm_device_types import HostStatusMixin
+from napalm_device_types.host_status import HOST_STATUS_COMMAND
+from napalm_device_types.package_updates import APT_UPGRADABLE_COMMAND
+
+from napalm_proxmox.driver import ProxmoxDriver
+
+APT = (
+    "libssl3t64/stable-security 3.5.1-1+deb13u2 amd64 [upgradable from: 3.5.1-1+deb13u1]\n"
+    "ceph-common/stable 20.2.4-pve5 amd64 [upgradable from: 20.2.4-pve4]\n"
+)
+API_ENTRY = {
+    "Package": "librados2",
+    "OldVersion": "20.2.4-pve4",
+    "Version": "20.2.4-pve5",
+    "Origin": "Proxmox",
+}
+
+
+def _api_updates(driver, entries=None, error=None):
+    api = MagicMock()
+    getter = api.nodes.return_value.apt.update.get
+    if error:
+        getter.side_effect = error
+    else:
+        getter.return_value = entries or []
+    driver._api = api
+    return api
+
+
+class TestAvailableUpdates:
+    def test_apt_over_the_exec_path_names_the_suite(self, driver):
+        with patch.object(driver, "_exec_ssh_command", return_value=APT + "__APT_RC=0\n") as exec_:
+            updates = {u["name"]: u for u in driver.get_available_updates()}
+
+        exec_.assert_called_once_with(APT_UPGRADABLE_COMMAND)
+        assert updates["libssl3t64"]["security"] is True
+        assert updates["ceph-common"]["security"] is False
+        assert updates["ceph-common"]["origin"] == "stable"
+
+    def test_the_api_is_the_fallback_with_security_unknown(self, driver):
+        _api_updates(driver, [API_ENTRY])
+        with patch.object(driver, "_exec_ssh_command", return_value=""):
+            updates = driver.get_available_updates()
+
+        assert updates == [
+            {
+                "name": "librados2",
+                "current_version": "20.2.4-pve4",
+                "new_version": "20.2.4-pve5",
+                "origin": "Proxmox",
+                "security": None,
+            }
+        ]
+
+    def test_a_failed_apt_falls_back_too(self, driver):
+        _api_updates(driver, [API_ENTRY])
+        with patch.object(driver, "_exec_ssh_command", return_value="E: lock\n__APT_RC=100\n"):
+            assert [u["name"] for u in driver.get_available_updates()] == ["librados2"]
+
+    def test_nothing_readable_raises_instead_of_reporting_nothing(self, driver):
+        _api_updates(driver, error=RuntimeError("API timeout"))
+        with patch.object(driver, "_exec_ssh_command", return_value=""):
+            with pytest.raises(RuntimeError):
+                driver.get_available_updates()
+
+    def test_nothing_pending_is_an_empty_list(self, driver):
+        with patch.object(driver, "_exec_ssh_command", return_value="__APT_RC=0\n"):
+            assert driver.get_available_updates() == []
+
+
+class TestRefresh:
+    def test_the_node_refreshes_its_index_through_the_api(self, driver):
+        api = _api_updates(driver)
+        api.nodes.return_value.apt.update.post.return_value = "UPID:pve1:0001"
+
+        result = driver.refresh_available_updates()
+
+        assert result["success"] is True
+        api.nodes.return_value.apt.update.post.assert_called_once()
+
+    def test_a_refused_refresh_says_why(self, driver):
+        api = _api_updates(driver)
+        api.nodes.return_value.apt.update.post.side_effect = RuntimeError(
+            "403 Permission check failed"
+        )
+
+        result = driver.refresh_available_updates()
+
+        assert result == {"success": False, "output": "403 Permission check failed"}
+
+
+class TestHostStatus:
+    def test_the_node_is_read_over_the_exec_path(self, driver):
+        report = (
+            "HSTAT_BEGIN\n[kernel]\n7.0.14-19-pve\n[modules]\n7.0.14-19-pve\n7.0.2-6-pve\n"
+            "[timers]\napt-daily-upgrade.timer enabled\nHSTAT_END\n"
+        )
+        with patch.object(driver, "_exec_ssh_command", return_value=report) as exec_:
+            status = driver.get_host_status()
+
+        exec_.assert_called_once_with(HOST_STATUS_COMMAND)
+        assert status["reboot_required"] is False
+
+    def test_the_driver_declares_the_contract(self):
+        assert issubclass(ProxmoxDriver, HostStatusMixin)