Merge pull request 'chore: adopt the NAPALM driver standard' (#33) from chore/driver-standard into main

- CI moves to .gitea/workflows/ci.yml and calls the shared driver workflow
  of NAPALM/driver-standard; .github/ is gone. napalm-device-types comes
  from git until 4.22.0 (AptProxyMixin) is on PyPI; the floor stays at
  4.22.0.
- requires-python >=3.10 (napalm-device-types needs it), py.typed, SPDX
  license string, a dev extra with floors (pytest-mock and responses are
  dropped: no test uses them), links to git.netork.io (README included).
  pytest no longer adds --cov by itself; CI passes it.
- mypy strict for the package; exempt for now, with 204 errors between
  them: napalm_proxmox.config_mixin, .driver, .interfaces_mixin,
  .lldp_mixin, .routing_mixin, .sdn_mixin, .system_mixin, .utils,
  .vm_contract_mixin, .vm_mixin and .vm_provision_mixin. They are recorded
  in the quality baseline with the long and complex functions and the two
  unannotated public methods.
- Coverage floor 68 % (measured 68.94 %).

.gitea/workflows/ci.yml

@@ -0,0 +1,17 @@
+name: CI
+
+on:
+  push:
+    branches: [main, master]
+    tags: ["v*"]
+  pull_request:
+
+jobs:
+  driver:
+    uses: NAPALM/driver-standard/.gitea/workflows/driver.yml@main
+    with:
+      standard-ref: main
+      # Until napalm-device-types 4.22.0 (AptProxyMixin, #53) is on PyPI.
+      pre-install: "git+https://git.netork.io/NAPALM/napalm-device-types.git"
+    secrets:
+      PYPI_TOKEN: ${{ secrets.PYPI_TOKEN }}

.github/workflows/ci.yml

@@ -1,48 +0,0 @@
-name: CI
-
-on:
-  push:
-    branches: ["**"]
-  pull_request:
-    branches: ["**"]
-
-jobs:
-  test:
-    runs-on: ubuntu-latest
-    strategy:
-      fail-fast: false
-      matrix:
-        python-version: ["3.10", "3.11", "3.12"]
-    steps:
-      - name: Checkout
-        uses: actions/checkout@v4
-
-      - name: Setup Python
-        uses: actions/setup-python@v5
-        with:
-          python-version: ${{ matrix.python-version }}
-          cache: pip
-
-      - name: Install package with dev extras
-        run: |
-          python -m pip install --upgrade pip
-          # napalm-device-types lives in git.netork.io/NAPALM, not on PyPI: without this
-          # pip looks there, finds an unrelated 0.1.0 and the job dies before any test.
-          python -m pip install "napalm-device-types @ git+https://git.netork.io/NAPALM/napalm-device-types.git"
-          python -m pip install -e ".[dev]"
-
-      - name: Run unit tests
-        run: |
-          python -m pytest -q --tb=short
-
-      - name: Build wheel and sdist
-        run: |
-          python -m pip install build
-          python -m build
-
-      - name: Upload dist artifacts
-        # v4 refuses to run on Gitea ("not currently supported on GHES").
-        uses: actions/upload-artifact@v3
-        with:
-          name: dist-${{ matrix.python-version }}
-          path: dist/*

CHANGELOG.md

@@ -7,7 +7,49 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+## [0.8.0] - 2026-10-23
+
+The first release on PyPI. It collects everything since 0.1.0.
+
+### Changed
+- The driver follows the NAPALM driver standard
+  (https://git.netork.io/NAPALM/driver-standard): CI on Gitea runs ruff, mypy, the
+  tests on Python 3.10–3.13 and at the lowest declared versions, bandit, pip-audit and
+  a build check, and a release tag publishes to PyPI. The package now ships `py.typed`.
+  **Python 3.9 is no longer supported**; napalm-device-types never ran on it.
+- Exceptions the driver deliberately ignores (closing the node's SSH session, the
+  `is_alive()` probe) are now logged at debug level.
+- The project links point at git.netork.io.
+
 ### Added
+- apt's proxy (napalm-device-types' `AptProxyMixin`, NetOrk/netork#1397): `get_package_proxy()`, `get_package_proxy_report()`, `check_package_proxy(url, expect)` and the fix `configure_package_proxy(url)`, as on any Debian host. The node is logged in as root, so the file is written and other proxy lines are commented out over the exec path. Requires napalm-device-types 4.22.0; version 0.7.0.
+- `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.
+
+### Changed
+- **Breaking:** `get_device_warnings` is gone; the node reports facts and the caller
+  names the warnings (netOrk #1111). lldpd and the pending updates were already facts
+  (`get_packages`, `get_available_updates`). The subscription is now
+  `get_licenses()` (napalm-device-types 4.14 `LicenseMixin`): one `pve-subscription`
+  entry with the API's status as a `LicenseStatus` (`notfound` → `none`, `new` and
+  `suspended` → `unknown`), its message as `detail`, product name and due date when
+  subscribed. The old check compared against `NotFound` while the API answers
+  `notfound`, so it never fired. Requires napalm-device-types 4.14.0; version 0.4.0.
+- **Breaking:** `run_device_action` (`fix_snmp`, `install_lldpd`) is gone. The driver
+  implements napalm-device-types 4.13's fix templates: `configure_snmp_agent(settings)`
+  with the caller's community and allowed sources (one `rocommunity` line each, ufw only
+  for those sources, no fixed temp file), `get_management_peer()`, and
+  `install_package(name)` as the contract says, so `ensure_package` comes from the base
+  package. Every step reads its exit status. Requires napalm-device-types 4.13.0;
+  version 0.3.0.
+
+### Added
+- `get_system_identity()` (napalm-device-types 4.9 `SystemIdentityMixin`): the node's SMBIOS
+  system UUID, product serial and board serial over the exec path, vendor placeholders
+  empty. netOrk matches a server to its BMC on these. Requires napalm-device-types 4.9.0;
+  version 0.2.0.
 - FreeBSD and OpenBSD guests: `create_vm_from_cloud_init(guest_os=...)` with
   `GUEST_AGENTS` from napalm-device-types 3.0. A BSD guest gets OS type
   `other` and a network-config v2 snippet of its own (`cicustom network=`),

README.md

@@ -12,8 +12,9 @@ It supports all three networking domains of Proxmox VE:
 
 ## Requirements
 
-- Python ≥ 3.9
+- Python ≥ 3.10
 - NAPALM ≥ 5.0.0
+- napalm-device-types ≥ 4.22.0
 - proxmoxer ≥ 2.0.0
 - netaddr ≥ 0.9.0
 - requests ≥ 2.31.0
@@ -27,7 +28,7 @@ pip install napalm-proxmox
 Or directly from source:
 
 ```bash
-git clone https://github.com/example/napalm-proxmox.git
+git clone https://git.netork.io/NAPALM/napalm-proxmox.git
 cd napalm-proxmox
 pip install -e .
 ```
@@ -45,9 +46,9 @@ device = driver(
     username="root",
     password="secret",
     optional_args={
-        "realm": "pam",       # default: "pam"
-        "port": 8006,         # default: 8006
-        "verify_ssl": True,   # default: True
+        "realm": "pam",  # default: "pam"
+        "port": 8006,  # default: 8006
+        "verify_ssl": True,  # default: True
     },
 )
 

napalm_proxmox/config_mixin.py

@@ -19,7 +19,6 @@ from __future__ import annotations
 import re
 from typing import Any
 
-
 _JsonDict = dict[str, Any]
 
 
@@ -44,13 +43,9 @@ class ProxmoxConfigMixin:
         configs: _JsonDict = {"running": "", "candidate": "", "startup": ""}
 
         if retrieve in ("running", "all", "startup"):
-            raw = self._exec_ssh_command(
-                "cat /etc/network/interfaces 2>/dev/null || true"
-            )
+            raw = self._exec_ssh_command("cat /etc/network/interfaces 2>/dev/null || true")
             # Append SDN config if available
-            sdn_raw = self._exec_ssh_command(
-                "cat /etc/pve/sdn/vnets.cfg 2>/dev/null || true"
-            )
+            sdn_raw = self._exec_ssh_command("cat /etc/pve/sdn/vnets.cfg 2>/dev/null || true")
             running = raw
             if sdn_raw:
                 running += "\n# === SDN VNets ===\n" + sdn_raw

napalm_proxmox/driver.py

@@ -32,36 +32,41 @@ from __future__ import annotations
 import logging
 from typing import Any
 
-logger = logging.getLogger(__name__)
-
+from napalm.base.exceptions import ConnectionException
 from napalm_device_types import (
+    AptProxyMixin,
+    EnsurePackageMixin,
     FingerprintRule,
     HostStatusMixin,
     HypervisorDriver,
     KernelFactsMixin,
+    LicenseMixin,
     ListeningSocketsMixin,
     PortSpec,
+    ServicePackagesMixin,
+    SnmpAgentMixin,
     SystemdServicesMixin,
+    SystemIdentityMixin,
 )
-from napalm.base.exceptions import ConnectionException
+from napalm_device_types.ssh_host_keys import HostKeyCheck, SshHostKeyMismatch
 
 try:
     from proxmoxer import ProxmoxAPI
 except ImportError as exc:
-    raise ImportError(
-        "proxmoxer is required: pip install proxmoxer"
-    ) from exc
+    raise ImportError("proxmoxer is required: pip install proxmoxer") from exc
 
-from napalm_proxmox.interfaces_mixin import ProxmoxInterfaceMixin
-from napalm_proxmox.sdn_mixin import ProxmoxSDNMixin
-from napalm_proxmox.lldp_mixin import ProxmoxLLDPMixin
 from napalm_proxmox.config_mixin import ProxmoxConfigMixin
-from napalm_proxmox.vm_mixin import ProxmoxVMMixin
-from napalm_proxmox.vm_contract_mixin import ProxmoxVMContractMixin
-from napalm_proxmox.vm_snapshot_mixin import ProxmoxVMSnapshotMixin
-from napalm_proxmox.vm_provision_mixin import ProxmoxVMProvisionMixin
+from napalm_proxmox.interfaces_mixin import ProxmoxInterfaceMixin
+from napalm_proxmox.lldp_mixin import ProxmoxLLDPMixin
 from napalm_proxmox.routing_mixin import ProxmoxRoutingMixin
+from napalm_proxmox.sdn_mixin import ProxmoxSDNMixin
 from napalm_proxmox.system_mixin import ProxmoxSystemMixin
+from napalm_proxmox.vm_contract_mixin import ProxmoxVMContractMixin
+from napalm_proxmox.vm_mixin import ProxmoxVMMixin
+from napalm_proxmox.vm_provision_mixin import ProxmoxVMProvisionMixin
+from napalm_proxmox.vm_snapshot_mixin import ProxmoxVMSnapshotMixin
+
+logger = logging.getLogger(__name__)
 
 # --------------------------------------------------------------------------- #
 # Type aliases
@@ -85,9 +90,15 @@ class ProxmoxDriver(
     ProxmoxRoutingMixin,
     ProxmoxSystemMixin,
     KernelFactsMixin,
+    ServicePackagesMixin,
     ListeningSocketsMixin,
+    SystemIdentityMixin,
     SystemdServicesMixin,
     HostStatusMixin,
+    LicenseMixin,
+    SnmpAgentMixin,
+    EnsurePackageMixin,
+    AptProxyMixin,
     HypervisorDriver,
 ):
     """NAPALM driver for Proxmox VE nodes."""
@@ -131,7 +142,8 @@ class ProxmoxDriver(
 
         self._port: int = self.optional_args.get("port", 8006)
         self._verify_ssl: bool = self.optional_args.get(
-            "verify_ssl", self.optional_args.get("ssl_verify", self.optional_args.get("verify", True))
+            "verify_ssl",
+            self.optional_args.get("ssl_verify", self.optional_args.get("verify", True)),
         )
         self._realm: str = self.optional_args.get("realm", "pam")
         self._token_name: str | None = self.optional_args.get("token_name")
@@ -144,7 +156,8 @@ class ProxmoxDriver(
 
         self._api: ProxmoxAPI | None = None
         self._node_name: str = ""
-        self._ssh_client: "paramiko.SSHClient | None" = None
+        # paramiko.SSHClient; paramiko is imported lazily (see _node_ssh).
+        self._ssh_client: Any = None
 
         # Candidate config (merge/replace)
         self._candidate_config: str = ""
@@ -198,9 +211,7 @@ class ProxmoxDriver(
             self._resolve_node()
 
         except Exception as exc:
-            raise ConnectionException(
-                f"Cannot connect to {self.hostname}: {exc}"
-            ) from exc
+            raise ConnectionException(f"Cannot connect to {self.hostname}: {exc}") from exc
 
     def _resolve_node(self) -> str:
         """Resolve the PVE node this connection talks to.
@@ -218,9 +229,7 @@ class ProxmoxDriver(
             self._node_name = self._node
             return self._node_name
 
-        members = [
-            s for s in (self._api.cluster.status.get() or []) if s.get("type") == "node"
-        ]
+        members = [s for s in (self._api.cluster.status.get() or []) if s.get("type") == "node"]
         short_host = self.hostname.split(".")[0]
         for pick in (
             lambda s: s.get("local"),
@@ -249,7 +258,7 @@ class ProxmoxDriver(
             try:
                 self._ssh_client.close()
             except Exception:
-                pass
+                logger.debug("%s: closing the SSH session: ignored", self.hostname, exc_info=True)
             self._ssh_client = None
 
     def is_alive(self) -> _JsonDict:
@@ -266,7 +275,7 @@ class ProxmoxDriver(
                 self._api.version.get()
                 alive = True
             except Exception:
-                pass
+                logger.debug("%s: liveness probe failed", self.hostname, exc_info=True)
         return {"is_alive": alive}
 
     # ------------------------------------------------------------------ #
@@ -292,7 +301,6 @@ class ProxmoxDriver(
         falls back to a direct paramiko SSH connection.
         """
         import base64 as _b64
-        import time as _time
 
         # Try API exec endpoint
         try:
@@ -310,39 +318,51 @@ class ProxmoxDriver(
 
         # Fallback: direct paramiko SSH
         try:
-            import paramiko
-
-            if self._ssh_client is None:
-                ssh_user = self._ssh_username or self.username
-                ssh_pass = self._ssh_password or self.password
-                ssh_pkey = None
-                if self._ssh_key and not ssh_pass:
-                    from io import StringIO as _StringIO
-                    ssh_pkey = paramiko.RSAKey.from_private_key(_StringIO(self._ssh_key))
-                self._ssh_client = paramiko.SSHClient()
-                self._ssh_client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
-                connect_kwargs: _JsonDict = {
-                    "hostname": self.hostname,
-                    "port": 22,
-                    "username": ssh_user,
-                    "timeout": self.timeout,
-                }
-                if ssh_pkey:
-                    connect_kwargs["pkey"] = ssh_pkey
-                else:
-                    connect_kwargs["password"] = ssh_pass
-                self._ssh_client.connect(**connect_kwargs)
-            _, stdout, stderr = self._ssh_client.exec_command(command, timeout=self.timeout)
+            _, stdout, stderr = self._node_ssh().exec_command(command, timeout=self.timeout)
             err = stderr.read().decode().strip()
             out = stdout.read().decode().strip()
             return out or err
         except ImportError:
             logger.warning("paramiko not installed — cannot exec SSH commands")
+        except SshHostKeyMismatch as exc:
+            # Refused before any login, and already reported to the caller's store.
+            logger.warning("%s", exc)
         except Exception as exc:
             logger.debug("SSH exec command failed: %s", exc)
 
         return ""
 
+    def _node_ssh(self) -> Any:  # paramiko.SSHClient; paramiko is imported lazily
+        """The SSH session to the node on port 22, opened once.
+
+        Its host key is the one the caller expects for port 22, or on first use
+        whichever the node presents; either way it is reported (netork#1167).
+        """
+        import paramiko
+
+        if self._ssh_client is None:
+            ssh_pass = self._ssh_password or self.password
+            connect_kwargs: _JsonDict = {
+                "username": self._ssh_username or self.username,
+                "timeout": self.timeout,
+            }
+            if self._ssh_key and not ssh_pass:
+                from io import StringIO as _StringIO
+
+                connect_kwargs["pkey"] = paramiko.RSAKey.from_private_key(_StringIO(self._ssh_key))
+            else:
+                connect_kwargs["password"] = ssh_pass
+            client = paramiko.SSHClient()
+            try:
+                HostKeyCheck(self.optional_args, self.hostname, 22).connect(
+                    client, **connect_kwargs
+                )
+            except Exception:
+                client.close()
+                raise
+            self._ssh_client = client
+        return self._ssh_client
+
     # ------------------------------------------------------------------ #
     # Node info helpers
     # ------------------------------------------------------------------ #
@@ -413,12 +433,9 @@ class ProxmoxDriver(
 
         uptime = float(status.get("uptime", 0))
         dns_search = dns.get("search", "")
-        hostname = self._node_name
         fqdn = f"{self._node_name}.{dns_search}" if dns_search else self.hostname
 
-        iface_list = sorted(
-            iface["iface"] for iface in network if iface.get("iface")
-        )
+        iface_list = sorted(iface["iface"] for iface in network if iface.get("iface"))
 
         pve_version = version.get("version", "")
         release = version.get("release", "")
@@ -442,7 +459,7 @@ class ProxmoxDriver(
                 "n=$(cat /sys/class/dmi/id/product_name 2>/dev/null); "
                 "r=$(cat /sys/class/dmi/id/product_version 2>/dev/null); "
                 "s=$(cat /sys/class/dmi/id/product_serial 2>/dev/null); "
-                "printf '%s\\n%s\\n%s\\n%s\\n' \"$v\" \"$n\" \"$r\" \"$s\""
+                'printf \'%s\\n%s\\n%s\\n%s\\n\' "$v" "$n" "$r" "$s"'
             )
             lines = self._exec_ssh_command(dmi_cmd).splitlines()
             if len(lines) >= 4:
@@ -492,5 +509,3 @@ class ProxmoxDriver(
     def cli(self, commands: list[str], encoding: str = "text") -> dict[str, str]:
         """Execute a list of shell commands and return their outputs."""
         return {cmd: self._exec_ssh_command(cmd) for cmd in commands}
-
-

napalm_proxmox/interfaces_mixin.py

@@ -174,8 +174,13 @@ class ProxmoxInterfaceMixin:
             parts = line.split()
             if len(parts) < 6:
                 continue
-            ip_addr, _, flags, mac, _, iface = (
-                parts[0], parts[1], parts[2], parts[3], parts[4], parts[5]
+            ip_addr, _, _flags, mac, _, iface = (
+                parts[0],
+                parts[1],
+                parts[2],
+                parts[3],
+                parts[4],
+                parts[5],
             )
             if mac in ("00:00:00:00:00:00", ""):
                 continue
@@ -202,9 +207,7 @@ class ProxmoxInterfaceMixin:
             if iface.get("type") in ("bridge",) and iface.get("iface")
         ]
         for bridge in linux_bridges:
-            raw = self._exec_ssh_command(
-                f"bridge fdb show br {bridge} 2>/dev/null || true"
-            )
+            raw = self._exec_ssh_command(f"bridge fdb show br {bridge} 2>/dev/null || true")
             for line in raw.splitlines():
                 parts = line.split()
                 if len(parts) < 3:
@@ -240,9 +243,7 @@ class ProxmoxInterfaceMixin:
             if iface.get("type") in ("OVSBridge",) and iface.get("iface")
         ]
         for bridge in ovs_bridges:
-            raw = self._exec_ssh_command(
-                f"ovs-appctl fdb/show {bridge} 2>/dev/null || true"
-            )
+            raw = self._exec_ssh_command(f"ovs-appctl fdb/show {bridge} 2>/dev/null || true")
             for line in raw.splitlines():
                 parts = line.split()
                 if len(parts) < 4:

napalm_proxmox/lldp_mixin.py

@@ -19,7 +19,6 @@ from __future__ import annotations
 import re
 from typing import Any
 
-
 _JsonDict = dict[str, Any]
 
 
@@ -29,9 +28,7 @@ class ProxmoxLLDPMixin:
     def get_lldp_neighbors(self) -> dict[str, list[_JsonDict]]:
         """Return LLDP neighbours (requires lldpd on the Proxmox node)."""
         result: dict[str, list[_JsonDict]] = {}
-        raw = self._exec_ssh_command(
-            "lldpcli show neighbors summary 2>/dev/null || true"
-        )
+        raw = self._exec_ssh_command("lldpcli show neighbors summary 2>/dev/null || true")
         current_iface = ""
         for line in raw.splitlines():
             m_iface = re.match(r"^\s*Interface:\s+(\S+?),?\s", line)
@@ -54,9 +51,7 @@ class ProxmoxLLDPMixin:
     def get_lldp_neighbors_detail(self, interface: str = "") -> dict[str, list[_JsonDict]]:
         """Return detailed LLDP neighbour information."""
         result: dict[str, list[_JsonDict]] = {}
-        raw = self._exec_ssh_command(
-            "lldpcli show neighbors details 2>/dev/null || true"
-        )
+        raw = self._exec_ssh_command("lldpcli show neighbors details 2>/dev/null || true")
         current_iface = ""
         current_entry: _JsonDict = {}
 

napalm_proxmox/routing_mixin.py

@@ -21,7 +21,6 @@ from typing import Any
 
 import napalm.base.constants as C
 
-
 _JsonDict = dict[str, Any]
 
 
@@ -92,12 +91,8 @@ class ProxmoxRoutingMixin:
 
         # Individual probe lines
         probes = []
-        for m_probe in re.finditer(
-            r"icmp_seq=\d+.*?time=([\d.]+) ms.*?from ([\d.a-fA-F:]+)", raw
-        ):
-            probes.append(
-                {"ip_address": m_probe.group(2), "rtt": float(m_probe.group(1))}
-            )
+        for m_probe in re.finditer(r"icmp_seq=\d+.*?time=([\d.]+) ms.*?from ([\d.a-fA-F:]+)", raw):
+            probes.append({"ip_address": m_probe.group(2), "rtt": float(m_probe.group(1))})
 
         return {
             "success": {

napalm_proxmox/sdn_mixin.py

@@ -153,7 +153,7 @@ class ProxmoxSDNMixin:
                     tags.setdefault(str(tag), set()).add(bridge)
 
         try:
-            for vm in (self._node_api().qemu.get() or []):
+            for vm in self._node_api().qemu.get() or []:
                 vmid = int(vm.get("vmid", 0))
                 try:
                     config = self._node_api().qemu(vmid).config.get() or {}
@@ -164,7 +164,7 @@ class ProxmoxSDNMixin:
             logger.warning("_get_vm_vlan_tags: failed to list QEMU VMs: %s", exc)
 
         try:
-            for ct in (self._node_api().lxc.get() or []):
+            for ct in self._node_api().lxc.get() or []:
                 vmid = int(ct.get("vmid", 0))
                 try:
                     config = self._node_api().lxc(vmid).config.get() or {}
@@ -238,9 +238,7 @@ class ProxmoxSDNMixin:
         2. LLDP basic fallback: first physical port with any LLDP neighbour.
         """
         network = {
-            iface["iface"]: iface
-            for iface in self._get_node_network()
-            if iface.get("iface")
+            iface["iface"]: iface for iface in self._get_node_network() if iface.get("iface")
         }
 
         try:
@@ -309,10 +307,7 @@ class ProxmoxSDNMixin:
                 return True
             min_id = min(int(n.get("nodeid", 9999)) for n in online_nodes)
             for n in online_nodes:
-                if (
-                    n.get("name") == self._node_name
-                    and int(n.get("nodeid", 9999)) == min_id
-                ):
+                if n.get("name") == self._node_name and int(n.get("nodeid", 9999)) == min_id:
                     return True
             return False
         except Exception as exc:
@@ -361,9 +356,7 @@ class ProxmoxSDNMixin:
             vlan_id: VLAN identifier (1-4094).
             config:  Dict that may contain ``"name"`` for the VLAN alias.
         """
-        name: str = (
-            (config.get("name") or f"VLAN{vlan_id}") if config else f"VLAN{vlan_id}"
-        )
+        name: str = (config.get("name") or f"VLAN{vlan_id}") if config else f"VLAN{vlan_id}"
         vnet_id = f"vlan{vlan_id:04d}"
 
         uplink = self._find_switch_uplink()

napalm_proxmox/system_mixin.py

@@ -12,22 +12,62 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-"""System-level NAPALM getters for Proxmox VE nodes (environment, NTP, SNMP, users, packages, services, updates, disk SMART)."""
+"""System-level NAPALM getters for Proxmox VE nodes.
+
+Environment, NTP, SNMP, users, packages, services, updates and disk SMART.
+"""
 
 from __future__ import annotations
 
+import base64
+import ipaddress
 import logging
 import re
+import shlex
 from typing import Any
 
 from napalm_device_types import APT_UPGRADABLE_COMMAND, parse_apt_upgradable
 
-from napalm_proxmox import utils
-
 logger = logging.getLogger(__name__)
 
 _JsonDict = dict[str, Any]
 
+#: The subscription API's status (lower-cased) as a napalm-device-types
+#: ``LicenseStatus``; "new" and "suspended" have none of their own.
+_SUBSCRIPTION_STATUS = {
+    "notfound": "none",
+    "active": "active",
+    "invalid": "invalid",
+    "expired": "expired",
+}
+
+
+_PACKAGE_NAME = re.compile(r"^[a-zA-Z0-9][a-zA-Z0-9_.+:~-]*$")
+#: What may stand in an snmpd.conf value: no whitespace, no quote, no comment.
+_SNMP_WORD = re.compile(r'^[^\s"#]+$')
+
+
+def _net(value: str) -> str:
+    return str(ipaddress.ip_network(value.strip(), strict=False))
+
+
+def render_snmpd_conf(settings: dict) -> str:
+    """net-snmp's configuration for *settings*: read-only v2c, per allowed source."""
+    community = settings["community"]
+    if not _SNMP_WORD.match(community):
+        raise ValueError("the community must be one word without quotes or #")
+    listen = settings.get("listen")
+    lines = [f"agentAddress udp:{listen + ':' if listen else ''}161"]
+    sources = [_net(s) for s in settings.get("allowed_sources", [])]
+    lines += [f"rocommunity {community} {s}" for s in sources] or [f"rocommunity {community}"]
+    for key, directive in (("location", "sysLocation"), ("contact", "sysContact")):
+        value = settings.get(key, "")
+        if value:
+            if "\n" in value:
+                raise ValueError(f"{key} must be one line")
+            lines.append(f"{directive}    {value}")
+    return "\n".join(lines) + "\n"
+
 
 class ProxmoxSystemMixin:
     """Mixin providing system-level NAPALM methods."""
@@ -138,9 +178,7 @@ class ProxmoxSystemMixin:
         Proxmox does not expose SNMP configuration via the REST API.
         We read /etc/snmp/snmpd.conf via exec if available.
         """
-        raw = self._exec_ssh_command(
-            "cat /etc/snmp/snmpd.conf 2>/dev/null || true"
-        )
+        raw = self._exec_ssh_command("cat /etc/snmp/snmpd.conf 2>/dev/null || true")
         communities: dict[str, _JsonDict] = {}
         location = ""
         contact = ""
@@ -189,7 +227,6 @@ class ProxmoxSystemMixin:
             if not uid:
                 continue
             # Proxmox roles: Administrator -> 15, otherwise 1
-            groups = user.get("groups", "") or ""
             level = 1
             try:
                 roles = self._api.access.users(uid).get() or {}  # type: ignore[union-attr]
@@ -199,7 +236,8 @@ class ProxmoxSystemMixin:
                 logger.debug("Failed to fetch roles for user %s: %s", uid, exc)
             result[uid] = {
                 "level": level,
-                "password": "",
+                # NAPALM's get_users shape; the password is never read.
+                "password": "",  # nosec B105
                 "sshkeys": [],
             }
 
@@ -217,7 +255,8 @@ class ProxmoxSystemMixin:
             if uname not in result and uid_int < 1000 or uid_int == 0:
                 result[uname] = {
                     "level": 15 if uid_int == 0 else 0,
-                    "password": "",
+                    # NAPALM's get_users shape; the password is never read.
+                    "password": "",  # nosec B105
                     "sshkeys": [],
                 }
 
@@ -231,10 +270,19 @@ 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)
     # ------------------------------------------------------------------ #
 
+    def _run_system_identity_command(self, command: str, *, privileged: bool) -> str:
+        """The transport for ``SystemIdentityMixin.get_system_identity``: the exec path,
+        which runs as root either way."""
+        return str(self._exec_ssh_command(command))
+
     def _run_listening_sockets_command(self, command: str, *, privileged: bool) -> str:
         """The transport for ``ListeningSocketsMixin.get_listening_sockets``:
         the exec path, which runs as root either way."""
@@ -296,65 +344,45 @@ class ProxmoxSystemMixin:
             if not name or status != "installed":
                 continue
             size_bytes = int(size_kb) * 1024 if size_kb.isdigit() else 0
-            result.append({
-                "name": name,
-                "version": version,
-                "installed": True,
-                "description": "",
-                "size": size_bytes,
-                "source": "pve",
-                "source_package": source_package,
-                "source_version": source_version,
-                "upgrade_version": upgradable.get(name, ""),
-            })
+            result.append(
+                {
+                    "name": name,
+                    "version": version,
+                    "installed": True,
+                    "description": "",
+                    "size": size_bytes,
+                    "source": "pve",
+                    "source_package": source_package,
+                    "source_version": source_version,
+                    "upgrade_version": upgradable.get(name, ""),
+                }
+            )
         return result
 
     # ------------------------------------------------------------------ #
-    # Device warnings
+    # Licenses (napalm-device-types' LicenseMixin)
     # ------------------------------------------------------------------ #
 
-    def get_device_warnings(self) -> list[_JsonDict]:
-        """Return warnings for the Proxmox node.
+    def get_licenses(self) -> list[_JsonDict]:
+        """The node's subscription, as the API reports it.
 
-        Currently detects:
-        - lldpd not installed (LLDP neighbor discovery unavailable)
-        - Available package updates (via Proxmox APT API)
-        - Missing / invalid subscription
+        A fact for the caller: whether running without one is a problem is not
+        the driver's to say. An unreadable subscription reports nothing.
         """
-        warnings: list[_JsonDict] = []
-
-        # 0. LLDP daemon
-        try:
-            lldpd_path = self._exec_ssh_command("command -v lldpd 2>/dev/null").strip()
-            if not lldpd_path:
-                warnings.append({"code": "lldpd_not_installed"})
-        except Exception as exc:
-            logger.debug("Failed to check for lldpd: %s", exc)
-
-        # 1. Available package updates
-        try:
-            updates = self.get_available_updates()
-            if updates:
-                warnings.append({
-                    "code": "updates_available",
-                    "meta": {
-                        "count": len(updates),
-                        "packages": [u["name"] for u in updates],
-                    },
-                })
-        except Exception as exc:
-            logger.debug("Failed to check available updates: %s", exc)
-
-        # 2. Subscription status
-        try:
-            sub = self._get_node_subscription()
-            status = sub.get("status", "")
-            if status in ("NotFound", "Invalid", "Expired"):
-                warnings.append({"code": "no_subscription", "meta": {"status": status}})
-        except Exception as exc:
-            logger.debug("Failed to check subscription status: %s", exc)
-
-        return warnings
+        sub = self._get_node_subscription()
+        raw = str(sub.get("status", ""))
+        if not raw:
+            return []
+        lic: _JsonDict = {
+            "name": "pve-subscription",
+            "status": _SUBSCRIPTION_STATUS.get(raw.lower(), "unknown"),
+            "detail": sub.get("message") or raw,
+        }
+        if sub.get("productname"):
+            lic["edition"] = sub["productname"]
+        if sub.get("nextduedate"):
+            lic["expires"] = sub["nextduedate"]
+        return [lic]
 
     # ------------------------------------------------------------------ #
     # Services (systemd, through napalm-device-types' SystemdServicesMixin)
@@ -417,11 +445,12 @@ class ProxmoxSystemMixin:
     def apply_updates(self, packages: list[str]) -> _JsonDict:
         """Upgrade the given packages via ``apt-get install`` over SSH."""
         for pkg in packages:
-            if not re.match(r'^[a-zA-Z0-9_\-\+\.]+$', pkg):
+            if not re.match(r"^[a-zA-Z0-9_\-\+\.]+$", pkg):
                 raise ValueError(f"Invalid package name: {pkg!r}")
         pkg_args = " ".join(packages)
         output = self._exec_ssh_command(
-            f"DEBIAN_FRONTEND=noninteractive apt-get install --only-upgrade -y {pkg_args} 2>&1 || true"
+            "DEBIAN_FRONTEND=noninteractive apt-get install --only-upgrade -y "
+            f"{pkg_args} 2>&1 || true"
         )
         return {"success": True, "output": output}
 
@@ -451,7 +480,8 @@ class ProxmoxSystemMixin:
         port = 161
         try:
             conf = self._exec_ssh_command(
-                "grep -E '^[[:space:]]*(ro|rw)?community' /etc/snmp/snmpd.conf 2>/dev/null | head -5"
+                "grep -E '^[[:space:]]*(ro|rw)?community' /etc/snmp/snmpd.conf "
+                "2>/dev/null | head -5"
             )
             for line in conf.splitlines():
                 parts = line.split()
@@ -469,131 +499,88 @@ class ProxmoxSystemMixin:
 
         return SNMPConfigDict(running=True, community=community, port=port, version="2c")
 
-    def run_device_action(self, action: str) -> dict:
-        """Execute a named administrative action on the Proxmox node."""
-        if action == "fix_snmp":
-            return self._action_fix_snmp()
-        if action == "install_lldpd":
-            return self._action_install_lldpd()
-        raise NotImplementedError(f"Unknown action: {action!r}")
+    # ------------------------------------------------------------------
+    # Fixes (napalm-device-types SnmpAgentMixin, EnsurePackageMixin): the
+    # caller's values. The node's shell runs as root.
+    # ------------------------------------------------------------------
 
-    def _action_install_lldpd(self) -> dict:
-        """Install, enable and start lldpd on the Proxmox node.
-
-        Proxmox runs Debian/Linux underneath, so this is a plain apt install
-        followed by enabling the systemd service. lldpd's defaults (listen on
-        all interfaces) are sufficient to discover the directly-connected
-        switch via the management bridge (e.g. vmbr0).
-        """
-        lines: list[str] = []
-
-        install_out = self._exec_ssh_command(
-            "DEBIAN_FRONTEND=noninteractive apt-get install -y lldpd 2>&1 | tail -5"
-        )
-        lines.append(f"[install] {install_out.strip()[-200:]}")
-
-        enable_out = self._exec_ssh_command(
-            "systemctl enable --now lldpd 2>&1 || service lldpd start 2>&1 || true"
-        )
-        lines.append(f"[service] {enable_out.strip()[-200:]}")
-
-        verify = self._exec_ssh_command("command -v lldpd 2>/dev/null").strip()
-        success = bool(verify)
-        if success:
-            lines.append("[ok] lldpd installed. Neighbors will appear after a short warm-up period.")
-        else:
-            lines.append("[warn] lldpd does not appear to be installed after the attempt.")
-
-        return {"success": success, "output": "\n".join(lines)}
-
-    def _action_fix_snmp(self) -> dict:
-        """Install, configure and start snmpd on the Proxmox node.
-
-        Proxmox runs Debian/Linux underneath. _exec_ssh_command runs as root
-        (either via Proxmox API execute endpoint or SSH with root credentials),
-        so no sudo is needed.
-        """
-        import base64 as _b64
-        lines: list[str] = []
-
-        # 1. Install snmpd and snmp client tools
-        install_out = self._exec_ssh_command(
-            "DEBIAN_FRONTEND=noninteractive apt-get install -y snmpd snmp 2>&1 | tail -5"
-        )
-        lines.append(f"[install] {install_out.strip()[-200:]}")
-
-        # 2. Detect the IP this connection comes from (for firewall rule)
-        netork_ip = ""
+    def _rc(self, command: str) -> tuple[str, int]:
+        """Run *command* and return its output and exit status (-1 when unknown)."""
+        out = self._exec_ssh_command(f"{{ {command}; }} 2>&1; echo __PVE_RC=$?")
+        text, _, tail = out.rpartition("__PVE_RC=")
         try:
-            raw = self._exec_ssh_command(
-                "ss -tnp 2>/dev/null | awk '/sshd/{print $5}' | head -1 | cut -d: -f1"
-            ).strip()
-            if raw and raw not in ("", "0.0.0.0", "::", "127.0.0.1"):
-                netork_ip = raw
-        except Exception as exc:
-            logger.debug("Failed to detect network origin IP: %s", exc)
+            return text.strip(), int(tail.strip())
+        except ValueError:
+            return out.strip(), -1
 
-        # 3. Write snmpd.conf via /tmp (no permission issues)
-        conf_str = (
-            "agentAddress udp:161\n"
-            "rocommunity public\n"
-            "sysLocation    Managed by netOrk\n"
-            "sysContact     netork@localhost\n"
+    def _step(self, command: str, done: str) -> dict:
+        output, status = self._rc(command)
+        return {"ok": status == 0, "output": done if status == 0 else output[-300:]}
+
+    # apt's proxy (napalm-device-types AptProxyMixin): the transports. The node
+    # is logged in as root, so every command runs as root.
+
+    def _run_apt_command(self, command: str) -> str:
+        return str(self._exec_ssh_command(command))
+
+    def _run_apt_root_command(self, command: str, done: str) -> dict:
+        return self._step(command, done)
+
+    def _write_apt_file(self, path: str, content: str) -> dict:
+        encoded = base64.b64encode(content.encode()).decode()
+        # No ``exit`` here: ``_rc`` reads the status after the command's braces.
+        return self._step(
+            f'umask 077 && t=$(mktemp) && echo {encoded} | base64 -d > "$t" && '
+            f'install -m 644 -o root -g root "$t" {shlex.quote(path)}; '
+            's=$?; rm -f "$t"; [ "$s" -eq 0 ]',
+            f"wrote {path}",
         )
-        conf_b64 = _b64.b64encode(conf_str.encode()).decode()
-        self._exec_ssh_command(f"echo {conf_b64} | base64 -d > /tmp/netork_snmpd.conf")
-        self._exec_ssh_command(
-            "mv /tmp/netork_snmpd.conf /etc/snmp/snmpd.conf && "
-            "chown root:root /etc/snmp/snmpd.conf && chmod 644 /etc/snmp/snmpd.conf"
+
+    def install_package(self, name: str, version: str = "") -> None:
+        """apt-get install *name* (``PackageManagementMixin``); raises when apt fails."""
+        if not _PACKAGE_NAME.match(name) or (version and not _PACKAGE_NAME.match(version)):
+            raise ValueError(f"not a package name: {name!r}")
+        target = f"{name}={version}" if version else name
+        output, status = self._rc(f"DEBIAN_FRONTEND=noninteractive apt-get install -y {target}")
+        if status != 0:
+            raise RuntimeError(f"apt-get install {target} failed: {output[-300:]}")
+
+    def get_management_peer(self) -> str | None:
+        """The address the node sees the SSH connection come from (``SSH_CONNECTION``)."""
+        out = self._exec_ssh_command('printf "%s\\n" "$SSH_CONNECTION"').split()
+        try:
+            return str(ipaddress.ip_address(out[0])) if out else None
+        except ValueError:
+            return None
+
+    def _install_snmp_agent(self) -> dict:
+        return self._step(
+            "DEBIAN_FRONTEND=noninteractive apt-get install -y snmpd snmp", "net-snmp installed"
         )
-        verify = self._exec_ssh_command("cat /etc/snmp/snmpd.conf 2>/dev/null").strip()
-        if "agentAddress" in verify and "rocommunity" in verify:
-            lines.append("[config] Wrote /etc/snmp/snmpd.conf - agentAddress udp:161, rocommunity public.")
-        else:
-            lines.append(f"[warn] snmpd.conf write may have failed: {verify[:100]}")
 
-        # 4. Open firewall if ufw is present
-        if netork_ip:
-            try:
-                ufw = self._exec_ssh_command("command -v ufw 2>/dev/null").strip()
-                if ufw:
-                    parts = netork_ip.rsplit(".", 1)
-                    subnet = f"{parts[0]}.0/24" if len(parts) == 2 else netork_ip
-                    fw_out = self._exec_ssh_command(
-                        f"ufw allow from {subnet} to any port 161 proto udp 2>&1"
-                    )
-                    lines.append(f"[firewall/ufw] {fw_out.strip()[:200]}")
-            except Exception as exc:
-                lines.append(f"[firewall] skipped - {exc}")
-
-        # 5. Stop and restart snmpd cleanly (no DBus needed for stop+start)
-        self._exec_ssh_command(
-            "service snmpd stop 2>/dev/null; pkill -9 snmpd 2>/dev/null; true"
+    def _write_snmp_agent_config(self, settings: dict) -> dict:
+        encoded = base64.b64encode(render_snmpd_conf(settings).encode()).decode()
+        return self._step(
+            f'umask 077 && t=$(mktemp) && echo {encoded} | base64 -d > "$t" && '
+            f'install -m 644 -o root -g root "$t" /etc/snmp/snmpd.conf; '
+            's=$?; rm -f "$t"; exit $s',
+            "wrote /etc/snmp/snmpd.conf",
         )
-        import time as _time
-        _time.sleep(1)
-        start_out = self._exec_ssh_command(
-            "service snmpd start 2>&1 || systemctl start snmpd 2>&1 || true"
+
+    def _allow_snmp_sources(self, settings: dict) -> dict:
+        sources = [_net(s) for s in settings.get("allowed_sources", [])]
+        if not sources:
+            return {"ok": True, "output": "no sources given; firewall unchanged"}
+        if self._rc("command -v ufw")[1] != 0:
+            return {"ok": True, "output": "no ufw; pve-firewall rules are the operator's"}
+        rules = " && ".join(f"ufw allow from {s} to any port 161 proto udp" for s in sources)
+        return self._step(rules, f"udp/161 open for {', '.join(sources)}")
+
+    def _start_snmp_agent(self) -> dict:
+        return self._step(
+            "systemctl enable snmpd && systemctl restart snmpd && systemctl is-active snmpd",
+            "snmpd enabled and running",
         )
-        lines.append(f"[service] {start_out.strip()[-200:]}")
-
-        # 6. Verify via local probe
-        _time.sleep(2)
-        probe_out = self._exec_ssh_command(
-            "snmpget -v2c -cpublic -t2 -r0 -Ov 127.0.0.1 1.3.6.1.2.1.1.1.0 2>&1 || true"
-        ).strip()
-        _snmp_types = ("STRING:", "INTEGER:", "OID:", "Timeticks:", "Hex-STRING:", "IpAddress:")
-        success = any(t in probe_out for t in _snmp_types)
-        if success:
-            lines.append("[ok] SNMP probe successful - community 'public' is working.")
-        else:
-            lines.append(f"[warn] SNMP probe failed - output: {probe_out[:200]}")
-
-        return {"success": success, "output": "\n".join(lines)}
-
-    # ------------------------------------------------------------------ #
-    # set_hostname
-    # ------------------------------------------------------------------ #
 
     def set_hostname(self, new_hostname: str) -> None:
         """Set the system hostname on the Proxmox node.
@@ -631,9 +618,11 @@ class ProxmoxSystemMixin:
             fqdn = new_hostname
 
         # Derive old short hostname for substitution in /etc/hosts
-        old_short = self._exec_ssh_command(
-            "cat /etc/hostname 2>/dev/null || hostname -s 2>/dev/null"
-        ).strip().split(".")[0]
+        old_short = (
+            self._exec_ssh_command("cat /etc/hostname 2>/dev/null || hostname -s 2>/dev/null")
+            .strip()
+            .split(".")[0]
+        )
 
         # ── 1. /etc/hostname ─────────────────────────────────────────────
         hostname_b64 = _b64.b64encode(f"{short_hostname}\n".encode()).decode()
@@ -648,6 +637,7 @@ class ProxmoxSystemMixin:
             # Both "hostname" and "hostname.domain" forms are covered because
             # the new value is the full fqdn when a domain was given.
             import re as _re
+
             new_replace = fqdn if domain else short_hostname
             hosts_new = _re.sub(
                 r"\b" + _re.escape(old_short) + r"\b",
@@ -664,16 +654,12 @@ class ProxmoxSystemMixin:
         # ── 3. /etc/mailname ─────────────────────────────────────────────
         mailname_b64 = _b64.b64encode(f"{fqdn}\n".encode()).decode()
         self._exec_ssh_command(
-            f"[ -f /etc/mailname ]"
-            f" && echo {mailname_b64} | base64 -d > /etc/mailname"
-            f" || true"
+            f"[ -f /etc/mailname ] && echo {mailname_b64} | base64 -d > /etc/mailname || true"
         )
 
         # ── 4. Postfix myhostname ─────────────────────────────────────────
         self._exec_ssh_command(
-            f"command -v postconf >/dev/null 2>&1"
-            f" && postconf -e 'myhostname = {fqdn}'"
-            f" || true"
+            f"command -v postconf >/dev/null 2>&1 && postconf -e 'myhostname = {fqdn}' || true"
         )
 
         # ── 5. Apply hostname at runtime ─────────────────────────────────
@@ -716,7 +702,9 @@ class ProxmoxSystemMixin:
         logger.info(
             "Proxmox set_hostname: %s → %s (fqdn: %s). "
             "Reboot required for node name to update in web UI / cluster.",
-            old_short, short_hostname, fqdn,
+            old_short,
+            short_hostname,
+            fqdn,
         )
 
     # ------------------------------------------------------------------ #
@@ -753,12 +741,12 @@ class ProxmoxSystemMixin:
             if not dev:
                 continue
             entry: dict = {
-                "model":   disk.get("model", ""),
-                "serial":  disk.get("serial", ""),
-                "type":    disk.get("type", ""),
-                "size":    disk.get("size", 0),
-                "health":  disk.get("health", "unknown").lower(),
-                "wearout": disk.get("wearout"),          # NVMe wear indicator 0-100
+                "model": disk.get("model", ""),
+                "serial": disk.get("serial", ""),
+                "type": disk.get("type", ""),
+                "size": disk.get("size", 0),
+                "health": disk.get("health", "unknown").lower(),
+                "wearout": disk.get("wearout"),  # NVMe wear indicator 0-100
                 "temperature": None,
                 "percentage_used": None,
                 "available_spare": None,
@@ -789,7 +777,7 @@ class ProxmoxSystemMixin:
                 # HDD-specific SMART attributes
                 for attr in smart.get("attributes", []):
                     name = attr.get("name", "").lower()
-                    raw  = attr.get("raw", "")
+                    raw = attr.get("raw", "")
                     try:
                         raw_int = int(str(raw).split()[0])
                     except (ValueError, TypeError):
@@ -827,7 +815,8 @@ class ProxmoxSystemMixin:
             "timezone": None,
             "ntp_servers": [],
             "ssh_port": 22,
-            "ssh_password_auth": False,
+            # A flag, not a password.
+            "ssh_password_auth": False,  # nosec B105
         }
 
         # Cluster name and node list from /cluster/status

napalm_proxmox/utils.py

@@ -14,7 +14,6 @@ from typing import Any
 
 from netaddr import EUI, AddrFormatError
 
-
 # ---------------------------------------------------------------------------
 # MAC address helpers
 # ---------------------------------------------------------------------------
@@ -47,6 +46,7 @@ def normalize_mac(mac: str) -> str:
 # IP / prefix helpers
 # ---------------------------------------------------------------------------
 
+
 def prefix_length(cidr: str) -> int:
     """Return prefix length from a CIDR string such as '192.168.1.0/24' → 24."""
     try:
@@ -141,6 +141,7 @@ def speed_mbps(iface_data: dict[str, Any]) -> float:
 # ARP / neighbour table helpers
 # ---------------------------------------------------------------------------
 
+
 def parse_arp_output(raw: str) -> list[dict[str, Any]]:
     """Parse Linux ``arp -n`` text output into NAPALM ARP dicts.
 
@@ -202,6 +203,7 @@ def sdn_zone_to_network_instance(zone: dict[str, Any]) -> dict[str, Any]:
 # OVS helpers
 # ---------------------------------------------------------------------------
 
+
 def parse_ovs_vsctl_show(raw: str) -> dict[str, Any]:
     """Very lightweight parser for ``ovs-vsctl show`` output.
 

napalm_proxmox/vm_mixin.py

@@ -20,8 +20,6 @@ import logging
 import re
 from typing import Any
 
-from napalm_proxmox import utils
-
 logger = logging.getLogger(__name__)
 
 _JsonDict = dict[str, Any]
@@ -30,9 +28,7 @@ _JsonDict = dict[str, Any]
 class ProxmoxVMMixin:
     """Mixin providing VM and container NAPALM methods."""
 
-    def get_vm_interfaces(
-        self, vmid: int, vm_type: str
-    ) -> tuple[dict[str, _JsonDict], bool, bool]:
+    def get_vm_interfaces(self, vmid: int, vm_type: str) -> tuple[dict[str, _JsonDict], bool, bool]:
         """Return network interfaces for a single VM or LXC container.
 
         Returns a 3-tuple ``(interfaces, agent_running, agent_enabled)``:
@@ -105,7 +101,7 @@ class ProxmoxVMMixin:
                 logger.debug("get_vm_interfaces: LXC %s config failed: %s", vmid, exc)
 
             try:
-                for iface in (self._node_api().lxc(vmid).interfaces.get() or []):
+                for iface in self._node_api().lxc(vmid).interfaces.get() or []:
                     name = iface.get("name", "")
                     if not name or name == "lo":
                         continue
@@ -233,7 +229,7 @@ class ProxmoxVMMixin:
 
         # QEMU VMs
         try:
-            for vm in (self._node_api().qemu.get() or []):
+            for vm in self._node_api().qemu.get() or []:
                 vmid = int(vm.get("vmid", 0))
                 name = vm.get("name", f"vm-{vmid}")
                 status = vm.get("status", "unknown")
@@ -256,30 +252,32 @@ class ProxmoxVMMixin:
                 )
 
                 disks, onboot = self._get_vm_disk_and_boot(vmid, "qemu")
-                result.append({
-                    "vmid": str(vmid),
-                    "name": name,
-                    "type": "vm",
-                    "status": status,
-                    "vcpus": vcpus,
-                    "memory": memory_mb,
-                    "cpu_usage": round(cpu_usage, 4),
-                    "memory_usage": memory_usage_mb,
-                    "uptime": uptime,
-                    "node": self._node_name,
-                    "interfaces": interfaces,
-                    "ipv4": ipv4,
-                    "agent_enabled": agent_enabled,
-                    "agent_running": agent_running,
-                    "disks": disks,
-                    "onboot": onboot,
-                })
+                result.append(
+                    {
+                        "vmid": str(vmid),
+                        "name": name,
+                        "type": "vm",
+                        "status": status,
+                        "vcpus": vcpus,
+                        "memory": memory_mb,
+                        "cpu_usage": round(cpu_usage, 4),
+                        "memory_usage": memory_usage_mb,
+                        "uptime": uptime,
+                        "node": self._node_name,
+                        "interfaces": interfaces,
+                        "ipv4": ipv4,
+                        "agent_enabled": agent_enabled,
+                        "agent_running": agent_running,
+                        "disks": disks,
+                        "onboot": onboot,
+                    }
+                )
         except Exception as exc:
             logger.warning("get_vms: failed to list QEMU VMs: %s", exc)
 
         # LXC containers
         try:
-            for ct in (self._node_api().lxc.get() or []):
+            for ct in self._node_api().lxc.get() or []:
                 vmid = int(ct.get("vmid", 0))
                 name = ct.get("name", f"ct-{vmid}")
                 status = ct.get("status", "unknown")
@@ -300,24 +298,26 @@ class ProxmoxVMMixin:
                 )
 
                 disks, onboot = self._get_vm_disk_and_boot(vmid, "lxc")
-                result.append({
-                    "vmid": str(vmid),
-                    "name": name,
-                    "type": "container",
-                    "status": status,
-                    "vcpus": vcpus,
-                    "memory": memory_mb,
-                    "cpu_usage": round(cpu_usage, 4),
-                    "memory_usage": memory_usage_mb,
-                    "uptime": uptime,
-                    "node": self._node_name,
-                    "interfaces": interfaces,
-                    "ipv4": ipv4,
-                    "agent_enabled": agent_enabled,
-                    "agent_running": agent_running,
-                    "disks": disks,
-                    "onboot": onboot,
-                })
+                result.append(
+                    {
+                        "vmid": str(vmid),
+                        "name": name,
+                        "type": "container",
+                        "status": status,
+                        "vcpus": vcpus,
+                        "memory": memory_mb,
+                        "cpu_usage": round(cpu_usage, 4),
+                        "memory_usage": memory_usage_mb,
+                        "uptime": uptime,
+                        "node": self._node_name,
+                        "interfaces": interfaces,
+                        "ipv4": ipv4,
+                        "agent_enabled": agent_enabled,
+                        "agent_running": agent_running,
+                        "disks": disks,
+                        "onboot": onboot,
+                    }
+                )
         except Exception as exc:
             logger.warning("get_vms: failed to list LXC containers: %s", exc)
 
@@ -328,9 +328,7 @@ class ProxmoxVMMixin:
     # Disk-key prefix for LXC: rootfs, mp (mount points)
     _DISK_KEYS_CT = re.compile(r"^(rootfs|mp\d+)$")
 
-    def _get_vm_disk_and_boot(
-        self, vmid: int, vm_type: str
-    ) -> tuple[list[dict], bool]:
+    def _get_vm_disk_and_boot(self, vmid: int, vm_type: str) -> tuple[list[dict], bool]:
         """Return (disks, onboot) for a VM or container.
 
         Fetches config from /nodes/{node}/{qemu|lxc}/{vmid}/config.
@@ -376,8 +374,8 @@ class ProxmoxVMMixin:
             logger.debug("Failed to fetch disk/boot config for vmid %s: %s", vmid, exc)
             return [], False
 
-    _POWER_ACTIONS_VM = {'start', 'stop', 'shutdown', 'reboot', 'reset'}
-    _POWER_ACTIONS_CT = {'start', 'stop', 'shutdown', 'reboot'}
+    _POWER_ACTIONS_VM = {"start", "stop", "shutdown", "reboot", "reset"}
+    _POWER_ACTIONS_CT = {"start", "stop", "shutdown", "reboot"}
 
     def power_vm(self, vmid: int, vm_type: str, action: str) -> dict:
         """Send a power action to a VM or container on this node.
@@ -385,11 +383,16 @@ class ProxmoxVMMixin:
         Supported actions for VMs:    start, stop, shutdown, reboot, reset
         Supported actions for containers: start, stop, shutdown, reboot
         """
-        allowed = self._POWER_ACTIONS_VM if vm_type == 'vm' else self._POWER_ACTIONS_CT
+        allowed = self._POWER_ACTIONS_VM if vm_type == "vm" else self._POWER_ACTIONS_CT
         if action not in allowed:
-            return {"success": False, "error": f"Action '{action}' not supported for {vm_type} (allowed: {sorted(allowed)})"}
+            return {
+                "success": False,
+                "error": (
+                    f"Action '{action}' not supported for {vm_type} (allowed: {sorted(allowed)})"
+                ),
+            }
         try:
-            vm_api = self._node_api().qemu(vmid) if vm_type == 'vm' else self._node_api().lxc(vmid)
+            vm_api = self._node_api().qemu(vmid) if vm_type == "vm" else self._node_api().lxc(vmid)
             task_id = getattr(vm_api.status, action).post()
             return {"success": True, "task_id": task_id or ""}
         except Exception as exc:

napalm_proxmox/vm_provision_mixin.py

@@ -8,7 +8,7 @@ import ipaddress
 import logging
 import re
 import time
-from typing import TYPE_CHECKING, Any, Dict, List
+from typing import TYPE_CHECKING, Any
 from urllib.parse import quote
 
 import yaml
@@ -143,31 +143,7 @@ class ProxmoxVMProvisionMixin:
         disk import — where a non-zero exit or a caller-specific timeout must
         surface as a hard failure rather than an empty string.
         """
-        import paramiko
-
-        if self._ssh_client is None:
-            ssh_user = self._ssh_username or self.username
-            ssh_pass = self._ssh_password or self.password
-            ssh_pkey = None
-            if self._ssh_key and not ssh_pass:
-                from io import StringIO as _StringIO
-
-                ssh_pkey = paramiko.RSAKey.from_private_key(_StringIO(self._ssh_key))
-            self._ssh_client = paramiko.SSHClient()
-            self._ssh_client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
-            connect_kwargs: Dict[str, Any] = {
-                "hostname": self.hostname,
-                "port": 22,
-                "username": ssh_user,
-                "timeout": self.timeout,
-            }
-            if ssh_pkey:
-                connect_kwargs["pkey"] = ssh_pkey
-            else:
-                connect_kwargs["password"] = ssh_pass
-            self._ssh_client.connect(**connect_kwargs)
-
-        _, stdout, stderr = self._ssh_client.exec_command(command, timeout=timeout)
+        _, stdout, stderr = self._node_ssh().exec_command(command, timeout=timeout)
         exit_status = stdout.channel.recv_exit_status()
         out = stdout.read().decode().strip()
         err = stderr.read().decode().strip()
@@ -281,7 +257,7 @@ class ProxmoxVMProvisionMixin:
             timeout=30,
         )
 
-    def _nic_macs(self, vmid: int, nics: List[Dict[str, Any]]) -> list[tuple[str, bool]]:
+    def _nic_macs(self, vmid: int, nics: list[dict[str, Any]]) -> list[tuple[str, bool]]:
         """``(mac, dhcp)`` per NIC, with the MAC Proxmox assigned where none was given."""
         config = self._node_api().qemu(vmid).config.get()
         macs = []
@@ -453,9 +429,9 @@ class ProxmoxVMProvisionMixin:
             )
         return path
 
-    def get_image_storages(self) -> List[StorageTargetDict]:
+    def get_image_storages(self) -> list[StorageTargetDict]:
         """List node-available storage pools suitable for a new VM's root disk."""
-        targets: List[StorageTargetDict] = []
+        targets: list[StorageTargetDict] = []
         for storage in self._node_api().storage.get():
             content = storage.get("content", "")
             if "images" not in content or storage.get("enabled", 1) == 0:
@@ -562,10 +538,10 @@ class ProxmoxVMProvisionMixin:
         image_url: str,
         cpu: int,
         memory: int,
-        nics: List[Dict[str, Any]],
-        cloud_init_config: Dict[str, Any],
+        nics: list[dict[str, Any]],
+        cloud_init_config: dict[str, Any],
         image_checksum: str | None = None,
-        ssh_public_keys: List[str] | None = None,
+        ssh_public_keys: list[str] | None = None,
         disk_resize_gb: int | None = None,
         storage: str | None = None,
         cpu_type: str | None = None,
@@ -657,7 +633,7 @@ class ProxmoxVMProvisionMixin:
             # Step 4: Configure network interfaces (CPU/memory already set at shell creation)
             _logger.info(f"Configuring {len(nics)} NIC(s) for VM {vmid}")
 
-            config_args: Dict[str, Any] = {}
+            config_args: dict[str, Any] = {}
 
             # Build NIC config strings generically
             for i, nic in enumerate(nics):
@@ -963,16 +939,16 @@ class ProxmoxVMProvisionMixin:
             raise
         except Exception as e:
             _logger.exception(f"Error getting status for VM {vmid}: {e}")
-            raise RuntimeError(f"Failed to get VM {vmid} status: {e}")
+            raise RuntimeError(f"Failed to get VM {vmid} status: {e}") from e
 
-    def get_network_targets(self) -> List[NetworkTargetDict]:
+    def get_network_targets(self) -> list[NetworkTargetDict]:
         """
         List selectable network targets (bridges + SDN vnets) for a new VM's NIC.
 
         Excludes physical NICs, bonds, and other non-bridge interface types —
         those are never valid ``NICConfigDict.bridge`` values on Proxmox.
         """
-        targets: List[NetworkTargetDict] = []
+        targets: list[NetworkTargetDict] = []
 
         for iface in self._get_node_network():
             iface_type = iface.get("type")

pyproject.toml

@@ -1,31 +1,29 @@
 [build-system]
-requires = ["setuptools>=61", "wheel"]
+requires = ["setuptools>=77"]
 build-backend = "setuptools.build_meta"
 
 [project]
 name = "napalm-proxmox"
-version = "0.1.0"
+version = "0.8.0"
 description = "NAPALM driver for Proxmox VE (classic Linux networking, SDN, OVS)"
 readme = "README.md"
-license = { text = "Apache-2.0" }
+license = "Apache-2.0"
 authors = [{ name = "Christian Manivong" }]
 keywords = ["napalm", "proxmox", "network", "sdn", "ovs", "automation"]
 classifiers = [
     "Development Status :: 3 - Alpha",
     "Intended Audience :: Developers",
     "Intended Audience :: System Administrators",
-    "License :: OSI Approved :: Apache Software License",
     "Programming Language :: Python :: 3",
-    "Programming Language :: Python :: 3.9",
     "Programming Language :: Python :: 3.10",
     "Programming Language :: Python :: 3.11",
     "Programming Language :: Python :: 3.12",
     "Topic :: System :: Networking",
 ]
-requires-python = ">=3.9"
+requires-python = ">=3.10"
 dependencies = [
     "napalm>=5.0.0",
-    "napalm_device_types>=3.0.0",
+    "napalm_device_types>=4.22.0",
     "paramiko>=5.0.0",  # CVE-2026-44405; imported directly for SSH fallback (driver.py)
     "proxmoxer>=2.0.0",
     "netaddr>=0.9.0",
@@ -33,34 +31,66 @@ dependencies = [
 ]
 
 [project.optional-dependencies]
-dev = [
-    "pytest>=7.4",
-    "pytest-cov>=4.1",
-    "pytest-mock>=3.12",
-    "responses>=0.24",
-    "ruff>=0.1",
-    "mypy>=1.7",
-]
+dev = ["pytest>=8", "pytest-cov>=5", "mypy>=1.11", "ruff>=0.6"]
 
 [project.entry-points."napalm.drivers"]
 proxmox = "napalm_proxmox:ProxmoxDriver"
 
+[project.urls]
+Repository = "https://git.netork.io/NAPALM/napalm-proxmox"
+Issues = "https://git.netork.io/NAPALM/napalm-proxmox/issues"
+Changelog = "https://git.netork.io/NAPALM/napalm-proxmox/src/branch/main/CHANGELOG.md"
+
 [tool.setuptools.packages.find]
 where = ["."]
 include = ["napalm_proxmox*"]
 
+[tool.setuptools.package-data]
+napalm_proxmox = ["py.typed"]
+
 [tool.ruff]
 line-length = 100
-target-version = "py39"
+target-version = "py310"
 
 [tool.ruff.lint]
-select = ["E", "F", "W", "I", "UP"]
+select = ["E", "F", "I", "UP", "B"]
+
+[tool.ruff.lint.per-file-ignores]
+# Recorded device output keeps its lines.
+"tests/**" = ["E501"]
+
+[tool.mypy]
+python_version = "3.10"
+strict = true
+# napalm itself is untyped, so every driver subclasses Any.
+disallow_subclassing_any = false
+
+[[tool.mypy.overrides]]
+module = ["napalm.*", "proxmoxer.*", "paramiko.*", "netaddr.*", "yaml.*"]
+ignore_missing_imports = true
+
+[[tool.mypy.overrides]]
+# Not yet under strict mypy (quality baseline); may only shrink.
+module = [
+    "napalm_proxmox.config_mixin",
+    "napalm_proxmox.driver",
+    "napalm_proxmox.interfaces_mixin",
+    "napalm_proxmox.lldp_mixin",
+    "napalm_proxmox.routing_mixin",
+    "napalm_proxmox.sdn_mixin",
+    "napalm_proxmox.system_mixin",
+    "napalm_proxmox.utils",
+    "napalm_proxmox.vm_contract_mixin",
+    "napalm_proxmox.vm_mixin",
+    "napalm_proxmox.vm_provision_mixin",
+]
+ignore_errors = true
 
 [tool.pytest.ini_options]
 testpaths = ["tests"]
-addopts = "--cov=napalm_proxmox --cov-report=term-missing"
 
-[tool.mypy]
-python_version = "3.9"
-warn_return_any = true
-warn_unused_configs = true
+[tool.coverage.run]
+source = ["napalm_proxmox"]
+
+[tool.coverage.report]
+fail_under = 68

tests/conftest.py

@@ -2,11 +2,11 @@
 
 from __future__ import annotations
 
-import pytest
 from unittest.mock import MagicMock, patch
 
-from napalm_proxmox.driver import ProxmoxDriver
+import pytest
 
+from napalm_proxmox.driver import ProxmoxDriver
 
 # ---------------------------------------------------------------------------
 # Sample API payloads
@@ -124,6 +124,7 @@ PVE_USERS = [
 # Driver fixture
 # ---------------------------------------------------------------------------
 
+
 def _build_mock_api(
     nodes=None,
     node_status=None,

tests/quality_baseline.json

@@ -0,0 +1,65 @@
+{
+ "coverage_floor": 68.0,
+ "function_complexity": {
+  "napalm_proxmox/driver.py::ProxmoxDriver._resolve_node": 13,
+  "napalm_proxmox/driver.py::ProxmoxDriver.get_facts": 16,
+  "napalm_proxmox/interfaces_mixin.py::ProxmoxInterfaceMixin.get_interfaces_counters": 14,
+  "napalm_proxmox/interfaces_mixin.py::ProxmoxInterfaceMixin.get_mac_address_table": 22,
+  "napalm_proxmox/routing_mixin.py::ProxmoxRoutingMixin.get_route_to": 13,
+  "napalm_proxmox/routing_mixin.py::ProxmoxRoutingMixin.ping": 12,
+  "napalm_proxmox/sdn_mixin.py::ProxmoxSDNMixin._find_switch_uplink": 13,
+  "napalm_proxmox/sdn_mixin.py::ProxmoxSDNMixin._get_ovs_bridge_for_port": 13,
+  "napalm_proxmox/sdn_mixin.py::ProxmoxSDNMixin._get_vm_vlan_tags": 11,
+  "napalm_proxmox/sdn_mixin.py::ProxmoxSDNMixin._is_cluster_master": 13,
+  "napalm_proxmox/sdn_mixin.py::ProxmoxSDNMixin.get_network_instances": 13,
+  "napalm_proxmox/sdn_mixin.py::ProxmoxSDNMixin.get_vlans": 16,
+  "napalm_proxmox/sdn_mixin.py::ProxmoxSDNMixin.set_vlan": 13,
+  "napalm_proxmox/system_mixin.py::ProxmoxSystemMixin.get_disk_smart": 22,
+  "napalm_proxmox/system_mixin.py::ProxmoxSystemMixin.get_packages": 16,
+  "napalm_proxmox/system_mixin.py::ProxmoxSystemMixin.get_users": 15,
+  "napalm_proxmox/utils.py::addresses_from_node_network": 15,
+  "napalm_proxmox/vm_contract_mixin.py::parse_vm_config": 16,
+  "napalm_proxmox/vm_mixin.py::ProxmoxVMMixin._get_vm_disk_and_boot": 16,
+  "napalm_proxmox/vm_mixin.py::ProxmoxVMMixin.get_vm_interfaces": 37,
+  "napalm_proxmox/vm_mixin.py::ProxmoxVMMixin.get_vms": 21,
+  "napalm_proxmox/vm_provision_mixin.py::ProxmoxVMProvisionMixin.create_vm_from_cloud_init": 26,
+  "napalm_proxmox/vm_provision_mixin.py::ProxmoxVMProvisionMixin.get_vm_status": 16
+ },
+ "function_lines": {
+  "napalm_proxmox/driver.py::ProxmoxDriver.get_facts": 82,
+  "napalm_proxmox/interfaces_mixin.py::ProxmoxInterfaceMixin.get_mac_address_table": 71,
+  "napalm_proxmox/lldp_mixin.py::ProxmoxLLDPMixin.get_lldp_neighbors_detail": 52,
+  "napalm_proxmox/routing_mixin.py::ProxmoxRoutingMixin.ping": 78,
+  "napalm_proxmox/sdn_mixin.py::ProxmoxSDNMixin._get_vm_vlan_tags": 54,
+  "napalm_proxmox/sdn_mixin.py::ProxmoxSDNMixin.get_vlans": 66,
+  "napalm_proxmox/sdn_mixin.py::ProxmoxSDNMixin.set_vlan": 70,
+  "napalm_proxmox/system_mixin.py::ProxmoxSystemMixin.get_disk_smart": 74,
+  "napalm_proxmox/system_mixin.py::ProxmoxSystemMixin.get_packages": 66,
+  "napalm_proxmox/system_mixin.py::ProxmoxSystemMixin.get_users": 51,
+  "napalm_proxmox/system_mixin.py::ProxmoxSystemMixin.set_hostname": 124,
+  "napalm_proxmox/vm_mixin.py::ProxmoxVMMixin.get_vm_interfaces": 179,
+  "napalm_proxmox/vm_mixin.py::ProxmoxVMMixin.get_vms": 114,
+  "napalm_proxmox/vm_provision_mixin.py::ProxmoxVMProvisionMixin._download_cloud_image": 61,
+  "napalm_proxmox/vm_provision_mixin.py::ProxmoxVMProvisionMixin.create_vm_from_cloud_init": 232,
+  "napalm_proxmox/vm_provision_mixin.py::ProxmoxVMProvisionMixin.destroy_vm": 65,
+  "napalm_proxmox/vm_provision_mixin.py::ProxmoxVMProvisionMixin.get_vm_status": 92
+ },
+ "mypy_relaxed": [
+  "napalm_proxmox.config_mixin",
+  "napalm_proxmox.driver",
+  "napalm_proxmox.interfaces_mixin",
+  "napalm_proxmox.lldp_mixin",
+  "napalm_proxmox.routing_mixin",
+  "napalm_proxmox.sdn_mixin",
+  "napalm_proxmox.system_mixin",
+  "napalm_proxmox.utils",
+  "napalm_proxmox.vm_contract_mixin",
+  "napalm_proxmox.vm_mixin",
+  "napalm_proxmox.vm_provision_mixin"
+ ],
+ "python_file_lines": {},
+ "untyped_public": [
+  "napalm_proxmox/sdn_mixin.py::ProxmoxSDNMixin.set_vlan",
+  "napalm_proxmox/system_mixin.py::ProxmoxSystemMixin.get_snmp_config"
+ ]
+}

tests/test_connection.py

@@ -2,12 +2,13 @@
 
 from __future__ import annotations
 
-import pytest
-from unittest.mock import MagicMock, patch
+from unittest.mock import patch
 
+import pytest
 from napalm.base.exceptions import ConnectionException
+
 from napalm_proxmox.driver import ProxmoxDriver
-from tests.conftest import _build_mock_api, NODES_LIST
+from tests.conftest import _build_mock_api
 
 
 class TestOpen:
@@ -101,9 +102,7 @@ def _open_with(hostname, **api_kwargs):
 
 class TestClusterNodeResolution:
     def test_local_node_wins_over_first_listed(self):
-        drv = _open_with(
-            "172.22.8.120", nodes=CLUSTER_NODES_LIST, cluster_status=CLUSTER_STATUS
-        )
+        drv = _open_with("172.22.8.120", nodes=CLUSTER_NODES_LIST, cluster_status=CLUSTER_STATUS)
         assert drv._node_name == "pve-dual"
 
     def test_matches_by_ip_without_local_flag(self):
@@ -118,9 +117,7 @@ class TestClusterNodeResolution:
 
     def test_matches_short_name_of_fqdn(self):
         status = [{**n, "local": 0} for n in CLUSTER_STATUS if n["type"] == "node"]
-        drv = _open_with(
-            "pve-01.mgmt.example.com", nodes=CLUSTER_NODES_LIST, cluster_status=status
-        )
+        drv = _open_with("pve-01.mgmt.example.com", nodes=CLUSTER_NODES_LIST, cluster_status=status)
         assert drv._node_name == "pve-01"
 
     def test_single_node_without_cluster_status(self):

tests/test_environment.py

@@ -2,11 +2,12 @@
 
 from __future__ import annotations
 
-import pytest
-from tests.conftest import _build_mock_api, NODE_STATUS
 from unittest.mock import patch
 
+import pytest
+
 from napalm_proxmox.driver import ProxmoxDriver
+from tests.conftest import _build_mock_api
 
 
 class TestGetEnvironment:
@@ -25,9 +26,6 @@ class TestGetEnvironment:
         assert isinstance(result["fans"], dict)
 
     def test_temperature_with_sensors(self):
-        from tests.conftest import _build_mock_api, NODE_STATUS
-        from unittest.mock import patch
-        from napalm_proxmox.driver import ProxmoxDriver
 
         sensors_data = [
             {"name": "CPU Package", "value": "55.0"},
@@ -49,9 +47,6 @@ class TestGetEnvironment:
         assert result["temperature"]["GPU Core"]["is_critical"] is False
 
     def test_temperature_critical(self):
-        from tests.conftest import _build_mock_api
-        from unittest.mock import patch
-        from napalm_proxmox.driver import ProxmoxDriver
 
         sensors_data = [{"name": "NVMe", "value": "98.0"}]
         mock_api = _build_mock_api(sensors=sensors_data)

tests/test_fix_templates.py

@@ -0,0 +1,192 @@
+"""The fixes as napalm-device-types templates, with the caller's values (netOrk #1105).
+
+``fix_snmp`` wrote an application's snmpd.conf (community "public", "Managed by
+netOrk") and opened ufw for the caller's /24 guessed from ``ss``;
+``install_lldpd`` was an action of its own. Now ``configure_snmp_agent(settings)``
+takes every value from the caller, and ``ensure_package`` installs through the
+contract's ``install_package``. Every step reads its exit status.
+"""
+
+from __future__ import annotations
+
+import re
+
+import pytest
+
+from napalm_proxmox import ProxmoxDriver
+
+SETTINGS = {"community": "c0mm", "allowed_sources": ["10.7.224.20"], "location": "rack 1"}
+
+
+class _Node:
+    """The node's shell: answers by command fragment, with an exit status."""
+
+    def __init__(self, **status):
+        self.status = status  # fragment -> exit status
+        self.ran: list[str] = []
+
+    def __call__(self, command: str) -> str:
+        self.ran.append(command)
+        if "SSH_CONNECTION" in command:
+            return "10.7.224.20 51234 10.0.0.9 22\n"
+        if command.startswith("dpkg-query"):
+            return "curl\tinstalled\n"
+        rc = next((code for frag, code in self.status.items() if frag in command), 0)
+        out = "ok" if rc == 0 else "E: failed"
+        return f"{out}\n__PVE_RC={rc}" if "__PVE_RC" in command else out
+
+
+@pytest.fixture
+def node():
+    return _Node()
+
+
+@pytest.fixture
+def driver(node):
+    d = ProxmoxDriver("pve", "root@pam", "pw")
+    d._exec_ssh_command = node
+    return d
+
+
+class TestSnmpAgent:
+    def test_install_write_open_start_with_the_callers_values(self, driver, node):
+        result = driver.configure_snmp_agent(SETTINGS)
+
+        assert result["success"] is True
+        assert [s["step"] for s in result["steps"]] == ["install", "configure", "firewall", "start"]
+        assert any("apt-get install -y snmpd snmp" in c for c in node.ran)
+        write = next(c for c in node.ran if "/etc/snmp/snmpd.conf" in c)
+        assert "install -m 644" in write
+        b64 = re.search(r"echo ([A-Za-z0-9+/=]+) \| base64 -d", write).group(1)
+        import base64
+
+        conf = base64.b64decode(b64).decode()
+        assert "rocommunity c0mm 10.7.224.20/32" in conf
+        assert "sysLocation    rack 1" in conf
+        assert "public" not in conf and "netork" not in conf.lower()
+        assert any("systemctl restart snmpd" in c for c in node.ran)
+
+    def test_no_temp_file_with_an_application_name(self, driver, node):
+        driver.configure_snmp_agent(SETTINGS)
+
+        assert not any("netork" in c for c in node.ran)
+
+    def test_ufw_opens_only_for_the_allowed_sources(self, driver, node):
+        node.status["command -v ufw"] = 0
+
+        driver.configure_snmp_agent(SETTINGS)
+
+        assert any("ufw allow from 10.7.224.20/32 to any port 161 proto udp" in c for c in node.ran)
+        assert not any(".0/24" in c for c in node.ran)
+
+    def test_a_failed_install_stops_the_fix(self, driver, node):
+        node.status["apt-get install -y snmpd"] = 100
+
+        result = driver.configure_snmp_agent(SETTINGS)
+
+        assert result["success"] is False
+        assert not any("/etc/snmp/snmpd.conf" in c for c in node.ran)
+
+    def test_the_management_peer(self, driver):
+        assert driver.get_management_peer() == "10.7.224.20"
+
+
+class TestPackages:
+    def test_ensure_package_installs_what_is_missing(self, driver, node):
+        result = driver.ensure_package("lldpd")
+
+        assert result["success"] is True
+        assert any("apt-get install -y lldpd" in c for c in node.ran)
+
+    def test_ensure_package_leaves_alone_what_is_there(self, driver, node):
+        assert driver.ensure_package("curl")["success"] is True
+        assert not any("apt-get install" in c for c in node.ran)
+
+    def test_install_package_raises_when_apt_fails(self, driver, node):
+        node.status["apt-get install -y lldpd"] = 100
+
+        with pytest.raises(RuntimeError, match="lldpd"):
+            driver.install_package("lldpd")
+
+    def test_a_name_that_is_no_package_name_is_refused(self, driver):
+        with pytest.raises(ValueError):
+            driver.install_package("lldpd; reboot")
+
+
+def test_the_old_actions_are_gone():
+    for name in ("run_device_action", "_action_fix_snmp", "_action_install_lldpd"):
+        assert not hasattr(ProxmoxDriver, name), name
+
+
+_APT_REPORT = (
+    "APROXY_BEGIN\n[dump]\n{dump}[parts]\n{parts}[main]\n[sources]\n"
+    "http://download.proxmox.com/debian/pve/dists/trixie/\nAPROXY_END\n"
+)
+
+
+class _AptNode(_Node):
+    """A node that answers apt's proxy report and a fetch."""
+
+    def __init__(self, proxy="", parts=""):
+        super().__init__()
+        self.proxy = proxy
+        self.parts = parts
+
+    def __call__(self, command: str) -> str:
+        if "apt-config dump" in command:
+            self.ran.append(command)
+            dump = f'Acquire::http::Proxy "{self.proxy}";\n' if self.proxy else ""
+            return _APT_REPORT.format(dump=dump, parts=self.parts)
+        if "apt-helper" in command:
+            self.ran.append(command)
+            return "APROXY_BEGIN\n[fetch]\nGet:1 x\n[rc]\n0\n[match]\n2\nAPROXY_END\n"
+        return super().__call__(command)
+
+
+class TestPackageProxy:
+    """Proxmox VE is Debian: apt's proxy is set and read as on any Debian host (netOrk #1397)."""
+
+    def test_it_takes_the_fix(self):
+        assert callable(getattr(ProxmoxDriver, "configure_package_proxy", None))
+
+    def test_the_fact(self):
+        d = ProxmoxDriver("pve", "root@pam", "pw")
+        d._exec_ssh_command = _AptNode(proxy="http://c:3142")
+
+        assert d.get_package_proxy() == "http://c:3142"
+
+    def test_written_as_root_read_back_and_fetched_through(self):
+        node = _AptNode(proxy="http://c:3142")
+        d = ProxmoxDriver("pve", "root@pam", "pw")
+        d._exec_ssh_command = node
+
+        result = d.configure_package_proxy("http://c:3142")
+
+        assert result["success"] is True, result["output"]
+        write = next(c for c in node.ran if "/etc/apt/apt.conf.d/00proxy" in c)
+        assert "install -m 644" in write
+        b64 = re.search(r"echo ([A-Za-z0-9+/=]+) \| base64 -d", write).group(1)
+        import base64
+
+        assert base64.b64decode(b64).decode() == 'Acquire::http::Proxy "http://c:3142";\n'
+        assert any("dists/trixie/InRelease" in c for c in node.ran)
+
+    def test_another_proxy_is_commented_out(self):
+        node = _AptNode(
+            proxy="http://c:3142",
+            parts='/etc/apt/apt.conf.d/90curtin-aptproxy:1:Acquire::http::Proxy "http://old:3128";\n',
+        )
+        d = ProxmoxDriver("pve", "root@pam", "pw")
+        d._exec_ssh_command = node
+
+        d.configure_package_proxy("http://c:3142")
+
+        assert any(c.startswith("{ sed -i -e '1{") for c in node.ran)
+
+    def test_the_proxy_answers(self):
+        d = ProxmoxDriver("pve", "root@pam", "pw")
+        d._exec_ssh_command = _AptNode()
+
+        result = d.check_package_proxy("http://c:3142/acng-report.html", expect="Cache")
+
+        assert result == {"reachable": True, "expected": True, "error": ""}

tests/test_get_facts.py

@@ -3,10 +3,8 @@
 from __future__ import annotations
 
 import pytest
-from tests.conftest import _build_mock_api, NODE_NETWORK
-from unittest.mock import patch
 
-from napalm_proxmox.driver import ProxmoxDriver
+from tests.conftest import NODE_NETWORK
 
 
 @pytest.fixture

tests/test_interfaces.py

@@ -2,11 +2,7 @@
 
 from __future__ import annotations
 
-import pytest
-from tests.conftest import _build_mock_api, NODE_NETWORK
-from unittest.mock import patch
-
-from napalm_proxmox.driver import ProxmoxDriver
+from tests.conftest import NODE_NETWORK
 
 
 class TestGetInterfaces:
@@ -89,10 +85,18 @@ class TestGetInterfacesCounters:
         result = driver.get_interfaces_counters()
         eth0 = result["eth0"]
         for field in (
-            "tx_errors", "rx_errors", "tx_discards", "rx_discards",
-            "tx_octets", "rx_octets", "tx_unicast_packets", "rx_unicast_packets",
-            "tx_multicast_packets", "rx_multicast_packets",
-            "tx_broadcast_packets", "rx_broadcast_packets",
+            "tx_errors",
+            "rx_errors",
+            "tx_discards",
+            "rx_discards",
+            "tx_octets",
+            "rx_octets",
+            "tx_unicast_packets",
+            "rx_unicast_packets",
+            "tx_multicast_packets",
+            "rx_multicast_packets",
+            "tx_broadcast_packets",
+            "rx_broadcast_packets",
         ):
             assert field in eth0, f"Missing field: {field}"
 

tests/test_kernel_facts.py

@@ -12,9 +12,9 @@ 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 = (

tests/test_licenses.py

@@ -0,0 +1,65 @@
+"""The node's subscription as a fact (netOrk #1111).
+
+The driver named ``no_subscription`` itself, comparing against "NotFound" while
+the API answers "notfound" -- so it never fired. It reports the subscription
+now; whether a missing one is a warning is the caller's decision.
+"""
+
+from __future__ import annotations
+
+import pytest
+
+from napalm_proxmox import ProxmoxDriver
+
+
+def _driver(subscription: dict) -> ProxmoxDriver:
+    d = ProxmoxDriver("pve", "root@pam", "pw")
+    d._get_node_subscription = lambda: subscription
+    return d
+
+
+@pytest.mark.parametrize(
+    "raw, status",
+    [
+        ("notfound", "none"),
+        ("NotFound", "none"),
+        ("active", "active"),
+        ("invalid", "invalid"),
+        ("expired", "expired"),
+        ("new", "unknown"),
+        ("suspended", "unknown"),
+    ],
+)
+def test_the_api_status_as_a_license_status(raw, status):
+    [lic] = _driver({"status": raw, "message": "There is no subscription key"}).get_licenses()
+
+    assert lic["name"] == "pve-subscription"
+    assert lic["status"] == status
+    assert lic["detail"] == "There is no subscription key"
+
+
+def test_level_and_due_date_when_subscribed():
+    sub = {
+        "status": "active",
+        "productname": "Proxmox VE Community Subscription 1 CPU/year",
+        "nextduedate": "2027-03-01",
+    }
+    [lic] = _driver(sub).get_licenses()
+
+    assert lic["edition"] == "Proxmox VE Community Subscription 1 CPU/year"
+    assert lic["expires"] == "2027-03-01"
+
+
+def test_an_unreadable_subscription_reports_nothing():
+    """``_get_node_subscription`` gives {} when the API call fails: no fact, not "none"."""
+    assert _driver({}).get_licenses() == []
+
+
+def test_no_warnings_of_its_own():
+    """lldpd, the updates and the subscription are facts the caller reads
+    (get_packages, get_available_updates, get_licenses)."""
+    assert not any(
+        "get_device_warnings" in vars(c)
+        for c in ProxmoxDriver.__mro__[:-1]
+        if c.__module__.startswith("napalm_proxmox")
+    )

tests/test_listening_sockets.py

@@ -11,6 +11,7 @@ from __future__ import annotations
 from unittest.mock import patch
 
 from napalm_device_types import ListeningSocketsMixin
+
 from napalm_proxmox.driver import ProxmoxDriver
 
 WIRE = (

tests/test_misc.py

@@ -2,8 +2,9 @@
 
 from __future__ import annotations
 
+from unittest.mock import patch
+
 import pytest
-from unittest.mock import MagicMock, patch
 
 
 class TestGetNTPServers:
@@ -85,7 +86,7 @@ class TestGetUsers:
     def test_user_has_required_keys(self, driver):
         driver._node_api().execute.post.return_value = {"data": ""}
         result = driver.get_users()
-        for user, data in result.items():
+        for _user, data in result.items():
             assert "level" in data
             assert "password" in data
             assert "sshkeys" in data
@@ -248,7 +249,7 @@ class TestGetRouteTo:
     def test_route_fields(self, driver):
         driver._node_api().execute.post.return_value = {"data": self.ROUTE_OUTPUT}
         result = driver.get_route_to()
-        for prefix, routes in result.items():
+        for _prefix, routes in result.items():
             for route in routes:
                 assert "protocol" in route
                 assert "next_hop" in route
@@ -257,7 +258,7 @@ class TestGetRouteTo:
     def test_protocol_filter(self, driver):
         driver._node_api().execute.post.return_value = {"data": self.ROUTE_OUTPUT}
         result = driver.get_route_to(protocol="dhcp")
-        for prefix, routes in result.items():
+        for _prefix, routes in result.items():
             for route in routes:
                 assert "dhcp" in route["protocol"].lower()
 

tests/test_ovs_and_arp.py

@@ -2,12 +2,10 @@
 
 from __future__ import annotations
 
-import pytest
 from unittest.mock import patch
 
 from napalm_proxmox import utils
 
-
 PROC_NET_ARP = """\
 IP address       HW type  Flags  HW address          Mask   Device
 10.0.0.254       0x1      0x2    aa:bb:cc:dd:ee:ff   *      eth0
@@ -137,10 +135,7 @@ class TestGetMACAddressTable:
         "cc:dd:ee:ff:00:11 dev eth0 vlan 20 master vmbr0\n"
     )
 
-    OVS_APPCTL = (
-        " LOCAL     0  aa:bb:cc:dd:ee:03    0\n"
-        "     1    10  11:22:33:44:55:66  100\n"
-    )
+    OVS_APPCTL = " LOCAL     0  aa:bb:cc:dd:ee:03    0\n     1    10  11:22:33:44:55:66  100\n"
 
     def test_linux_bridge_entries(self, driver):
         def _exec(cmd):

tests/test_sdn.py

@@ -2,8 +2,7 @@
 
 from __future__ import annotations
 
-import pytest
-from tests.conftest import _build_mock_api, SDN_ZONES, SDN_VNETS
+from tests.conftest import _build_mock_api
 
 
 class TestGetVlans:
@@ -52,8 +51,8 @@ class TestGetVlans:
 
     def test_empty_sdn_returns_dict(self):
         from unittest.mock import patch
+
         from napalm_proxmox.driver import ProxmoxDriver
-        from tests.conftest import _build_mock_api
 
         mock_api = _build_mock_api(sdn_vnets=[], sdn_zones=[])
         drv = ProxmoxDriver("pve1", "root", "secret", optional_args={"verify_ssl": False})

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

tests/test_ssh_host_key.py

@@ -0,0 +1,133 @@
+"""The node's SSH session verifies its host key, and reports the one it sees (NetOrk/netork#1167).
+
+Both ways onto the node's shell — the best-effort ``_exec_ssh_command`` and the
+provisioning steps' ``_run_node_command`` — share one session, opened on port 22.
+"""
+
+from __future__ import annotations
+
+from unittest.mock import MagicMock
+
+import paramiko
+import pytest
+from napalm_device_types.ssh_host_keys import (
+    KNOWN_KEYS_ARG,
+    OBSERVER_ARG,
+    SshHostKey,
+    SshHostKeyMismatch,
+)
+
+from napalm_proxmox.driver import ProxmoxDriver
+
+SERVER_KEY = paramiko.ECDSAKey.generate()
+OTHER_KEY = paramiko.ECDSAKey.generate()
+
+
+class _Stream:
+    def __init__(self, data: bytes) -> None:
+        self._data = data
+        self.channel = MagicMock()
+        self.channel.recv_exit_status.return_value = 0
+
+    def read(self) -> bytes:
+        return self._data
+
+
+class _Client:
+    """paramiko.SSHClient as far as the driver uses it; refuses like paramiko would."""
+
+    def __init__(self) -> None:
+        self.lines: list[str] = []
+        self.policy = None
+        self.connected_with: dict | None = None
+        self.commands: list[str] = []
+
+    def set_missing_host_key_policy(self, policy) -> None:
+        self.policy = policy
+
+    def get_host_keys(self):
+        client = self
+
+        class _Keys:
+            def load(self, path: str) -> None:
+                with open(path) as f:
+                    client.lines += f.read().splitlines()
+
+        return _Keys()
+
+    def connect(self, hostname, **kwargs) -> None:
+        if self.lines and SshHostKey.of(SERVER_KEY).key not in self.lines[0]:
+            raise paramiko.BadHostKeyException(hostname, SERVER_KEY, OTHER_KEY)
+        if not self.lines:
+            self.policy.missing_host_key(self, hostname, SERVER_KEY)
+        self.connected_with = {"hostname": hostname, **kwargs}
+
+    def get_transport(self):
+        transport = MagicMock()
+        transport.get_remote_server_key.return_value = SERVER_KEY
+        return transport
+
+    def exec_command(self, command, timeout=None):
+        self.commands.append(command)
+        return None, _Stream(b"ok"), _Stream(b"")
+
+    def close(self) -> None:
+        pass
+
+
+@pytest.fixture
+def client(monkeypatch) -> _Client:
+    fake = _Client()
+    monkeypatch.setattr(paramiko, "SSHClient", lambda: fake)
+    return fake
+
+
+def _driver(seen: list, known: dict | None = None) -> ProxmoxDriver:
+    args: dict = {"port": 8006, OBSERVER_ARG: seen.append}
+    if known is not None:
+        args[KNOWN_KEYS_ARG] = known
+    driver = ProxmoxDriver("10.0.0.9", "root", "s3cret", optional_args=args)
+    driver._node_api = MagicMock(side_effect=RuntimeError("no exec endpoint"))  # type: ignore[method-assign]
+    return driver
+
+
+def test_first_use_runs_the_command_and_reports_the_key(client):
+    seen: list = []
+    assert _driver(seen)._run_node_command("true", timeout=5) == "ok"
+    assert client.connected_with["port"] == 22
+    assert seen[0]["accepted"] is True
+    assert (seen[0]["host"], seen[0]["port"]) == ("10.0.0.9", 22)
+
+
+def test_the_key_for_the_api_port_is_not_the_ssh_one(client):
+    seen: list = []
+    other = SshHostKey.of(OTHER_KEY).as_dict()
+    _driver(seen, {"8006": other})._run_node_command("true", timeout=5)
+    assert seen[0]["expected"] is None
+
+
+def test_a_changed_key_stops_a_provisioning_step(client):
+    seen: list = []
+    driver = _driver(seen, {"22": SshHostKey.of(OTHER_KEY).as_dict()})
+    with pytest.raises(SshHostKeyMismatch):
+        driver._run_node_command("qm importdisk …", timeout=5)
+    assert client.commands == []
+    assert client.connected_with is None
+    assert seen[0]["accepted"] is False
+
+
+def test_a_changed_key_leaves_the_best_effort_read_empty(client):
+    seen: list = []
+    driver = _driver(seen, {"22": SshHostKey.of(OTHER_KEY).as_dict()})
+    assert driver._exec_ssh_command("cat /etc/os-release") == ""
+    assert client.commands == []
+    assert seen[0]["accepted"] is False
+
+
+def test_the_known_key_reuses_one_session(client):
+    seen: list = []
+    driver = _driver(seen, {"22": SshHostKey.of(SERVER_KEY).as_dict()})
+    driver._exec_ssh_command("uptime")
+    driver._run_node_command("true", timeout=5)
+    assert client.commands == ["uptime", "true"]
+    assert len(seen) == 1 and seen[0]["accepted"] is True

tests/test_system_identity.py

@@ -0,0 +1,32 @@
+"""`get_system_identity`: the node's SMBIOS identity, what ties it to its BMC.
+
+The command, its parse and the placeholder filter are napalm-device-types'; the driver
+only carries the command over its exec path, which runs as root.
+"""
+
+from __future__ import annotations
+
+from unittest.mock import patch
+
+from napalm_device_types import SYSTEM_IDENTITY_COMMAND, SystemIdentityMixin
+
+from napalm_proxmox.driver import ProxmoxDriver
+
+# pve-dual, a Supermicro X10DRL-i: the product serial is the board vendor's default.
+WIRE = "IDBEGIN\n31303043-534D-C40C-7A72-6F1300000000\n0123456789\nZM155S026022\nIDEND\n"
+
+
+def test_the_driver_declares_the_contract():
+    assert issubclass(ProxmoxDriver, SystemIdentityMixin)
+
+
+def test_it_runs_the_shared_command(driver):
+    with patch.object(driver, "_exec_ssh_command", return_value=WIRE) as exec_:
+        identity = driver.get_system_identity()
+
+    exec_.assert_called_once_with(SYSTEM_IDENTITY_COMMAND)
+    assert identity == {
+        "system_uuid": "31303043-534d-c40c-7a72-6f1300000000",
+        "product_serial": "",
+        "board_serial": "ZM155S026022",
+    }

tests/test_utils.py

@@ -2,7 +2,6 @@
 
 from __future__ import annotations
 
-import pytest
 from napalm_proxmox import utils
 
 

tests/test_vm_provision_mixin.py

@@ -3,12 +3,11 @@
 from __future__ import annotations
 
 import base64
+from unittest.mock import MagicMock, patch
 
 import pytest
-from unittest.mock import MagicMock, patch
-from napalm_proxmox.driver import ProxmoxDriver
-from napalm_proxmox.vm_provision_mixin import ProxmoxVMProvisionMixin
 
+from napalm_proxmox.vm_provision_mixin import ProxmoxVMProvisionMixin
 
 # ---------------------------------------------------------------------------
 # Tests
@@ -616,8 +615,9 @@ def _mock_ssh_exec(exit_status: int, stdout_text: str = "", stderr_text: str = "
 
 def test_run_node_command_success_returns_stdout():
     mixin = ProxmoxVMProvisionMixin()
-    mixin._ssh_client = MagicMock()
-    mixin._ssh_client.exec_command.return_value = _mock_ssh_exec(0, stdout_text="hello\n")
+    ssh = MagicMock()
+    ssh.exec_command.return_value = _mock_ssh_exec(0, stdout_text="hello\n")
+    mixin._node_ssh = MagicMock(return_value=ssh)  # the driver's verified session
 
     result = mixin._run_node_command("echo hello", timeout=10)
 
@@ -626,10 +626,9 @@ def test_run_node_command_success_returns_stdout():
 
 def test_run_node_command_nonzero_exit_raises():
     mixin = ProxmoxVMProvisionMixin()
-    mixin._ssh_client = MagicMock()
-    mixin._ssh_client.exec_command.return_value = _mock_ssh_exec(
-        1, stderr_text="No such file or directory"
-    )
+    ssh = MagicMock()
+    ssh.exec_command.return_value = _mock_ssh_exec(1, stderr_text="No such file or directory")
+    mixin._node_ssh = MagicMock(return_value=ssh)
 
     with pytest.raises(RuntimeError, match="No such file or directory"):
         mixin._run_node_command("cat /nonexistent", timeout=10)