Merge pull request 'feat: get_service_packages, which packages a node's systemd unit is (NetOrk/netork#1236)' (#29) from feat/service-packages-1236 into master

CHANGELOG.md

@@ -7,6 +7,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Added
+- `get_service_packages()` (napalm-device-types' `ServicePackagesMixin`, NetOrk/netork#1236): which installed packages each systemd unit of the node is made of — the one shipping its unit file and the one shipping its `ExecStart` binary. Read-only, over the exec path. Requires napalm-device-types 4.20.0; version 0.6.0.
+
 ### Security
 - The node's SSH session (the fallback for node commands and the provisioning steps) verifies its host key (NetOrk/netork#1167): the key the caller expects for port 22 in `optional_args["ssh_host_keys"]`, else the first one; every key seen goes to `optional_args["ssh_host_key_observer"]`. A changed key is refused before any login: a provisioning step fails, a best-effort read returns nothing. Both paths now share one session (`_node_ssh`). Replaces paramiko's `AutoAddPolicy`. Requires napalm-device-types 4.19.0; version 0.5.0.
 

napalm_proxmox/driver.py

@@ -43,6 +43,7 @@ from napalm_device_types import (
     LicenseMixin,
     ListeningSocketsMixin,
     PortSpec,
+    ServicePackagesMixin,
     SnmpAgentMixin,
     SystemIdentityMixin,
     SystemdServicesMixin,
@@ -90,6 +91,7 @@ class ProxmoxDriver(
     ProxmoxRoutingMixin,
     ProxmoxSystemMixin,
     KernelFactsMixin,
+    ServicePackagesMixin,
     ListeningSocketsMixin,
     SystemIdentityMixin,
     SystemdServicesMixin,

napalm_proxmox/system_mixin.py

@@ -269,6 +269,10 @@ class ProxmoxSystemMixin:
         """The transport for ``KernelFactsMixin.get_kernel_facts``: the exec path."""
         return self._exec_ssh_command(command)
 
+    def _run_service_packages_command(self, command: str) -> str:
+        """The transport for ``ServicePackagesMixin.get_service_packages``: the exec path."""
+        return str(self._exec_ssh_command(command))
+
     # ------------------------------------------------------------------ #
     # Listening sockets (ListeningSocketsMixin supplies get_listening_sockets)
     # ------------------------------------------------------------------ #

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "napalm-proxmox"
-version = "0.5.0"
+version = "0.6.0"
 description = "NAPALM driver for Proxmox VE (classic Linux networking, SDN, OVS)"
 readme = "README.md"
 license = { text = "Apache-2.0" }
@@ -25,7 +25,7 @@ classifiers = [
 requires-python = ">=3.9"
 dependencies = [
     "napalm>=5.0.0",
-    "napalm_device_types>=4.19.0",
+    "napalm_device_types>=4.20.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_service_packages.py

@@ -0,0 +1,43 @@
+"""`get_service_packages`: which installed packages each systemd unit of the node is.
+
+The node's own services -- pveproxy, corosync, ceph -- are where a decision
+about a service lands on a hypervisor (NetOrk/netork#1236). The command and its
+parse are napalm-device-types'; the driver only carries the command over its
+exec path.
+"""
+
+from __future__ import annotations
+
+from unittest.mock import patch
+
+import pytest
+
+from napalm_device_types import ServicePackagesMixin
+from napalm_device_types.service_packages import SERVICE_PACKAGES_COMMAND
+from napalm_proxmox.driver import ProxmoxDriver
+
+WIRE = (
+    "SVCPKG_BEGIN\n[units]\nId=pveproxy.service\n"
+    "FragmentPath=/lib/systemd/system/pveproxy.service\n"
+    "ExecStart={ path=/usr/bin/pveproxy ; argv[]=/usr/bin/pveproxy start }\n\n"
+    "[dpkg]\npve-manager: /lib/systemd/system/pveproxy.service\n"
+    "pve-manager: /usr/bin/pveproxy\nSVCPKG_END"
+)
+
+
+def test_the_driver_declares_the_contract():
+    assert issubclass(ProxmoxDriver, ServicePackagesMixin)
+
+
+def test_it_runs_the_shared_command(driver):
+    with patch.object(driver, "_exec_ssh_command", return_value=WIRE) as exec_:
+        found = driver.get_service_packages()
+
+    exec_.assert_called_once_with(SERVICE_PACKAGES_COMMAND)
+    assert found == {"pveproxy": ["pve-manager"]}
+
+
+def test_output_without_a_report_raises(driver):
+    with patch.object(driver, "_exec_ssh_command", return_value=""):
+        with pytest.raises(ValueError):
+            driver.get_service_packages()