fix: report which services are enabled, and whether an action worked

get_services() read list-unit-files' second column, which since systemd 245
is followed by a preset column -- "enabled  enabled" never equalled
"enabled", so every service on every node was reported disabled (#6).
manage_service() ended in "|| true" and returned success whatever happened.

Both now come from napalm-device-types' SystemdServicesMixin (2.2.0): the
enabled state is UnitFileState from systemctl show, MainPID comes along in
the same round trip, and an action reports systemctl's exit status. The
driver keeps only _run_service_command(), its exec path as root. The exit
status marker keeps the output non-empty, so _exec_ssh_command's SSH fallback
on an empty API answer can no longer run an action twice.

The listing no longer includes templates, static unit files and aliases that
are not loaded (on a PVE 9 node: 280 entries -> 172).

Closes #6

napalm_proxmox/driver.py

@@ -34,7 +34,13 @@ from typing import Any
 
 logger = logging.getLogger(__name__)
 
-from napalm_device_types import FingerprintRule, HypervisorDriver, PortSpec
+from napalm_device_types import (
+    FingerprintRule,
+    HypervisorDriver,
+    KernelFactsMixin,
+    PortSpec,
+    SystemdServicesMixin,
+)
 from napalm.base.exceptions import ConnectionException
 
 try:
@@ -76,6 +82,8 @@ class ProxmoxDriver(
     ProxmoxVMProvisionMixin,
     ProxmoxRoutingMixin,
     ProxmoxSystemMixin,
+    KernelFactsMixin,
+    SystemdServicesMixin,
     HypervisorDriver,
 ):
     """NAPALM driver for Proxmox VE nodes."""

napalm_proxmox/system_mixin.py

@@ -221,6 +221,14 @@ class ProxmoxSystemMixin:
 
         return result
 
+    # ------------------------------------------------------------------ #
+    # Kernel facts (KernelFactsMixin supplies get_kernel_facts)
+    # ------------------------------------------------------------------ #
+
+    def _run_kernel_facts_command(self, command: str) -> str:
+        """The transport for ``KernelFactsMixin.get_kernel_facts``: the exec path."""
+        return self._exec_ssh_command(command)
+
     # ------------------------------------------------------------------ #
     # Packages (Debian APT)
     # ------------------------------------------------------------------ #
@@ -338,79 +346,16 @@ class ProxmoxSystemMixin:
         return warnings
 
     # ------------------------------------------------------------------ #
-    # Services (systemd)
+    # Services (systemd, through napalm-device-types' SystemdServicesMixin)
     # ------------------------------------------------------------------ #
 
-    def get_services(self) -> list[_JsonDict]:
+    def _run_service_command(self, command: str, *, privileged: bool, timeout: int) -> str:
-        """Return systemd services with running and enabled state.
+        """The transport for ``SystemdServicesMixin``: the exec path, as root.
 
-        Uses two ``systemctl`` invocations combined in a single SSH command:
+        *privileged* needs nothing more on a node the driver reaches as root, and
-        - ``list-unit-files`` for the static enabled/disabled state
+        the exec path keeps its own timeout.
-        - ``list-units`` for the live running state
         """
-        raw = self._exec_ssh_command(
+        return str(self._exec_ssh_command(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}
 
     # ------------------------------------------------------------------ #
     # Available updates

pyproject.toml

@@ -25,7 +25,7 @@ classifiers = [
 requires-python = ">=3.9"
 dependencies = [
     "napalm>=5.0.0",
-    "napalm_device_types>=2.0.0",
+    "napalm_device_types>=2.2.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_kernel_facts.py

@@ -0,0 +1,46 @@
+"""`get_kernel_facts`: what the node's kernel has built and loaded.
+
+A Proxmox node runs its own kernel under every guest, which makes it the host
+where a kernel CVE's preconditions matter most. The command and its parse are
+napalm-device-types'; the driver only carries the command over its exec path.
+"""
+
+from __future__ import annotations
+
+import base64
+import gzip
+from unittest.mock import patch
+
+import pytest
+
+from napalm_device_types import KernelFactsMixin
+from napalm_device_types.kernel import KERNEL_FACTS_COMMAND
+from napalm_proxmox.driver import ProxmoxDriver
+
+REPORT = (
+    "[release]\n6.8.12-4-pve\n[loaded]\nkvm_intel\n[builtin]\nkernel/net/ipv4/tcp_cubic.ko\n"
+    "[available]\nkernel/net/tipc/tipc.ko\n[config]\nCONFIG_TIPC=m\n"
+)
+WIRE = "KFACTS_BEGIN\n" + base64.encodebytes(gzip.compress(REPORT.encode())).decode() + "KFACTS_END"
+
+
+def test_the_driver_declares_the_contract():
+    assert issubclass(ProxmoxDriver, KernelFactsMixin)
+
+
+def test_it_runs_the_shared_command(driver):
+    with patch.object(driver, "_exec_ssh_command", return_value=WIRE) as exec_:
+        facts = driver.get_kernel_facts()
+
+    exec_.assert_called_once_with(KERNEL_FACTS_COMMAND)
+    assert facts["release"] == "6.8.12-4-pve"
+    assert facts["loaded"] == ["kvm_intel"]
+    assert facts["builtin"] == ["tcp_cubic"]
+    assert facts["available"] == ["tipc"]
+    assert facts["config"] == {"CONFIG_TIPC": "m"}
+
+
+def test_output_without_a_report_raises(driver):
+    with patch.object(driver, "_exec_ssh_command", return_value=""):
+        with pytest.raises(ValueError):
+            driver.get_kernel_facts()

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()