diff --git a/napalm_proxmox/driver.py b/napalm_proxmox/driver.py index 7c3b279..11eb40e 100644 --- a/napalm_proxmox/driver.py +++ b/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.""" diff --git a/napalm_proxmox/system_mixin.py b/napalm_proxmox/system_mixin.py index 3c78524..26f6d77 100644 --- a/napalm_proxmox/system_mixin.py +++ b/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]: - """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 diff --git a/pyproject.toml b/pyproject.toml index fdd5f63..9bc0199 100644 --- a/pyproject.toml +++ b/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", diff --git a/tests/test_services.py b/tests/test_services.py new file mode 100644 index 0000000..f6aef06 --- /dev/null +++ b/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()