feat: report where an update comes from and whether it is a security fix, refresh the index, read the host status

For netOrk MVP 5, on napalm-device-types 2.3.0:

- get_available_updates reads `apt list --upgradable` over the exec path
  (APT_UPGRADABLE_COMMAND), so each update carries its suite and security
  status; the APT API, which names only "Debian"/"Proxmox", is the fallback
  with security unknown. It raises when neither answers instead of returning
  [] -- it used to swallow every error.
- refresh_available_updates(): POST nodes/{n}/apt/update.
- HostStatusMixin over the exec path (reboot required, self-patching).

napalm_proxmox/driver.py

@@ -34,7 +34,14 @@ from typing import Any
 
 logger = logging.getLogger(__name__)
 
-from napalm_device_types import FingerprintRule, HypervisorDriver, KernelFactsMixin, PortSpec
+from napalm_device_types import (
+    FingerprintRule,
+    HostStatusMixin,
+    HypervisorDriver,
+    KernelFactsMixin,
+    PortSpec,
+    SystemdServicesMixin,
+)
 from napalm.base.exceptions import ConnectionException
 
 try:
@@ -77,6 +84,8 @@ class ProxmoxDriver(
     ProxmoxRoutingMixin,
     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__)
@@ -346,101 +348,63 @@ class ProxmoxSystemMixin:
         return warnings
 
     # ------------------------------------------------------------------ #
-    # Services (systemd)
+    # Services (systemd, through napalm-device-types' SystemdServicesMixin)
     # ------------------------------------------------------------------ #
 
-    def get_services(self) -> list[_JsonDict]:
-        """Return systemd services with running and enabled state.
+    def _run_service_command(self, command: str, *, privileged: bool, timeout: int) -> str:
+        """The transport for ``SystemdServicesMixin``: the exec path, as root.
 
-        Uses two ``systemctl`` invocations combined in a single SSH command:
-        - ``list-unit-files`` for the static enabled/disabled state
-        - ``list-units`` for the live running state
+        *privileged* needs nothing more on a node the driver reaches as root, and
+        the exec path keeps its own timeout.
         """
-        raw = self._exec_ssh_command(
-            "{ systemctl list-unit-files --type=service --no-pager --no-legend --full 2>/dev/null;"
-            " echo '---UNITS---';"
-            " systemctl list-units --type=service --all --no-pager --no-legend --full 2>/dev/null;"
-            " } || true"
-        )
-
-        # Parse enabled state from list-unit-files
-        enabled_map: dict[str, bool] = {}
-        section = "files"
-        for line in raw.splitlines():
-            if line.strip() == "---UNITS---":
-                section = "units"
-                continue
-            parts = line.strip().split(None, 1)
-            if len(parts) < 1:
-                continue
-            unit = parts[0].lstrip("\u25cf").strip()
-            if not unit.endswith(".service"):
-                continue
-            name = unit[: -len(".service")]
-            if section == "files":
-                state = parts[1].strip() if len(parts) > 1 else ""
-                enabled_map[name] = state in ("enabled", "enabled-runtime", "static")
-
-        # Parse running state from list-units
-        running_map: dict[str, bool] = {}
-        section = "files"
-        for line in raw.splitlines():
-            if line.strip() == "---UNITS---":
-                section = "units"
-                continue
-            if section != "units":
-                continue
-            parts = line.strip().lstrip("\u25cf").strip().split(None, 4)
-            if len(parts) < 4:
-                continue
-            unit = parts[0]
-            if not unit.endswith(".service"):
-                continue
-            name = unit[: -len(".service")]
-            sub_state = parts[3]
-            running_map[name] = sub_state == "running"
-
-        all_names = sorted(set(enabled_map) | set(running_map))
-        return [
-            {
-                "name": name,
-                "running": running_map.get(name, False),
-                "enabled": enabled_map.get(name, False),
-                "pid": 0,
-            }
-            for name in all_names
-        ]
-
-    def manage_service(self, name: str, action: str) -> _JsonDict:
-        """Start / stop / restart / enable / disable a systemd service."""
-        if not re.match(r'^[a-zA-Z0-9_\-\.@]+$', name):
-            raise ValueError(f"Invalid service name: {name!r}")
-        if action not in ('start', 'stop', 'restart', 'enable', 'disable'):
-            raise ValueError(f"Invalid action: {action!r}")
-        output = self._exec_ssh_command(f"systemctl {action} {name}.service 2>&1 || true")
-        return {"success": True, "output": output}
+        return str(self._exec_ssh_command(command))
 
     # ------------------------------------------------------------------ #
     # Available updates
     # ------------------------------------------------------------------ #
 
     def get_available_updates(self) -> list[_JsonDict]:
-        """Return list of upgradable packages from the Proxmox APT API."""
-        updates: list[_JsonDict] = []
+        """Return the node's upgradable packages, with origin and security status.
+
+        ``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():
-                pkg = upd.get("Package", "")
-                if not pkg:
-                    continue
-                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)
+            updates = parse_apt_upgradable(self._exec_ssh_command(APT_UPGRADABLE_COMMAND) or "")
+        except ValueError as exc:
+            logger.debug("apt list over the exec path failed, using the API: %s", exc)
+            updates = self._updates_from_api()
         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.1.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_services.py

@@ -0,0 +1,68 @@
+"""Services on a Proxmox node: systemd, through napalm-device-types' mixin.
+
+The listing used to read ``list-unit-files``' second column, which since
+systemd 245 is followed by a preset column -- so ``enabled`` was false for
+every service on every node (#6). An action ended in ``|| true`` and reported
+success whatever happened. Both now come from the shared mixin; the driver
+only carries the command over its exec path, as root.
+"""
+
+from __future__ import annotations
+
+from unittest.mock import patch
+
+import pytest
+from napalm_device_types import SystemdServicesMixin
+from napalm_device_types.systemd import SYSTEMD_SERVICES_COMMAND, service_action_command
+
+from napalm_proxmox.driver import ProxmoxDriver
+
+REPORT = (
+    "SVC_BEGIN\n[files]\npveproxy.service enabled enabled\n[units]\n"
+    "MainPID=1234\nId=pveproxy.service\nNames=pveproxy.service\nLoadState=loaded\n"
+    "ActiveState=active\nSubState=running\nUnitFileState=enabled\n"
+    "[generated]\nSVC_END\n"
+)
+
+
+def test_the_driver_uses_the_shared_mixin():
+    assert issubclass(ProxmoxDriver, SystemdServicesMixin)
+    assert ProxmoxDriver.get_services is SystemdServicesMixin.get_services
+    assert ProxmoxDriver.manage_service is SystemdServicesMixin.manage_service
+
+
+def test_listing_runs_the_shared_command(driver):
+    with patch.object(driver, "_exec_ssh_command", return_value=REPORT) as exec_:
+        services = driver.get_services()
+
+    exec_.assert_called_once_with(SYSTEMD_SERVICES_COMMAND)
+    assert services == [{"name": "pveproxy", "running": True, "enabled": True, "pid": 1234}]
+
+
+def test_an_unreadable_listing_raises_instead_of_reporting_no_services(driver):
+    with patch.object(driver, "_exec_ssh_command", return_value=""):
+        with pytest.raises(ValueError):
+            driver.get_services()
+
+
+def test_an_action_reports_its_real_outcome(driver):
+    failed = "Failed to restart nope.service: Unit nope.service not found.\n__SVC_RC=5"
+    with patch.object(driver, "_exec_ssh_command", return_value=failed) as exec_:
+        result = driver.manage_service("nope", "restart")
+
+    exec_.assert_called_once_with(service_action_command("nope", "restart"))
+    assert result["success"] is False
+    assert "not found" in result["output"]
+
+
+def test_a_successful_action(driver):
+    with patch.object(driver, "_exec_ssh_command", return_value="__SVC_RC=0"):
+        assert driver.manage_service("pveproxy", "restart") == {"success": True, "output": ""}
+
+
+def test_an_invalid_name_never_reaches_the_node(driver):
+    with patch.object(driver, "_exec_ssh_command") as exec_:
+        with pytest.raises(ValueError):
+            driver.manage_service("pveproxy; reboot", "stop")
+
+    exec_.assert_not_called()

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)