napalm_proxmox/driver.py

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

napalm_proxmox/system_mixin.py

@@ -346,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.1.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_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()