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

chore(release): 0.8.0
chore: adopt the NAPALM driver standard
2026-10-23 12:17:27 +00:00 · 2026-10-23 12:02:11 +02:00 · 2026-10-23 08:40:41 +02:00 · 2026-10-23 08:39:01 +02:00 · 2026-10-23 08:37:27 +02:00 · 2026-10-22 22:08:25 +00:00
39 changed files with 2545 additions and 564 deletions
@@ -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 }}
@@ -7,7 +7,74 @@ 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=`),
  because FreeBSD's nuageinit skips runcmd on the v1 Proxmox generates.
  OpenBSD's agent runs over ISA serial behind `serial0`, as the image keeps
  its console on com0. An unknown `guest_os` is refused before anything is
  created.
 - Packed cloud images (FreeBSD's `.qcow2.xz`) are unpacked on the node after
  the packed file's checksum is verified; only the unpacked file is cached.
  Proxmox's `download-url` unpacks ISOs only, so they always go over SSH.
 ### Fixed
 - `get_vm_status` no longer reports a loopback address as the VM's IP when
  the agent lists loopback under another name than `lo` (`lo0` on the BSDs,
  listed first by OpenBSD).
 - `destroy_vm` reads the VM's cloud-init snippets before deleting the VM, and
  removes all of them; it read them afterwards, when the config was gone (#15).
 - Cloud images for `create_vm_from_cloud_init` are downloaded by Proxmox itself
  when the node has an active storage with content type `import` (Proxmox
  8.2+): `download-url` with checksum verification, then `import-from` as the
  root disk. The image no longer passes through the node's root filesystem.
  Files are named `netork-<url hash>-<name>.<qcow2|raw|vmdk>` and reused;
  Ubuntu's `.img` is stored as `.qcow2`. Without such a storage, or for an
  image type Proxmox cannot import, the SSH download into
  `/var/lib/vz/template/netork-images` is used as before.
 - `HypervisorDriver` contract methods `start_vm`, `stop_vm`, `reboot_vm`,
  `suspend_vm` and `get_vm_config`. They accept a VM's name or vmid, raise
  `ValueError`/`RuntimeError` instead of returning a result dict, and wait
@@ -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 .
 ```
@@ -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(
+            raw = self._exec_ssh_command("cat /etc/network/interfaces 2>/dev/null || true")
                "cat /etc/network/interfaces 2>/dev/null || true"
            )
            # Append SDN config if available
-            sdn_raw = self._exec_ssh_command(
+            sdn_raw = self._exec_ssh_command("cat /etc/pve/sdn/vnets.cfg 2>/dev/null || true")
                "cat /etc/pve/sdn/vnets.cfg 2>/dev/null || true"
            )
            running = raw
            if sdn_raw:
                running += "\n# === SDN VNets ===\n" + sdn_raw
@@ -32,28 +32,41 @@ from __future__ import annotations
 import logging
 from typing import Any
 logger = logging.getLogger(__name__)
 from napalm_device_types import FingerprintRule, HypervisorDriver, PortSpec
 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_device_types.ssh_host_keys import HostKeyCheck, SshHostKeyMismatch
 try:
    from proxmoxer import ProxmoxAPI
 except ImportError as exc:
-    raise ImportError(
+    raise ImportError("proxmoxer is required: pip install proxmoxer") from exc
        "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.interfaces_mixin import ProxmoxInterfaceMixin
-from napalm_proxmox.vm_contract_mixin import ProxmoxVMContractMixin
+from napalm_proxmox.lldp_mixin import ProxmoxLLDPMixin
 from napalm_proxmox.vm_snapshot_mixin import ProxmoxVMSnapshotMixin
 from napalm_proxmox.vm_provision_mixin import ProxmoxVMProvisionMixin
 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
@@ -76,6 +89,16 @@ class ProxmoxDriver(
    ProxmoxVMProvisionMixin,
    ProxmoxRoutingMixin,
    ProxmoxSystemMixin,
    KernelFactsMixin,
    ServicePackagesMixin,
    ListeningSocketsMixin,
    SystemIdentityMixin,
    SystemdServicesMixin,
    HostStatusMixin,
    LicenseMixin,
    SnmpAgentMixin,
    EnsurePackageMixin,
    AptProxyMixin,
    HypervisorDriver,
 ):
    """NAPALM driver for Proxmox VE nodes."""
@@ -86,6 +109,10 @@ class ProxmoxDriver(
    USES_SSH = False
    # A PVE node reboots through a full init sequence plus storage checks.
    REBOOT_SETTLE_SECONDS = 90
    #: "Upgrade everything" must be a full upgrade on Proxmox VE: a plain
    #: ``apt-get upgrade`` holds back what needs new or removed packages and can
    #: leave the node half updated. netOrk reads this when it upgrades the host.
    FULL_UPGRADE = True
    PORT_SPECS = [
        PortSpec("https", 8006, weight=8.0),
    ]
@@ -115,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")
@@ -128,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 = ""
@@ -182,36 +211,46 @@ class ProxmoxDriver(
            self._resolve_node()
        except Exception as exc:
-            raise ConnectionException(
+            raise ConnectionException(f"Cannot connect to {self.hostname}: {exc}") from exc
                f"Cannot connect to {self.hostname}: {exc}"
            ) from exc
    def _resolve_node(self) -> str:
-        """Resolve the node name from the hostname or optional_args."""
+        """Resolve the PVE node this connection talks to.
        In a cluster, ``GET /nodes`` lists every member, so its first entry is
        just some node, not necessarily the one at ``self.hostname``. That used
        to be taken blindly, and a device then reported another node's name and
        VMs. ``GET /cluster/status`` marks the node the session landed on with
        ``local: 1``; failing that, the node is matched by IP or name.
        API errors propagate so that ``open()`` fails with the real cause (e.g.
        a TLS verification error) rather than succeeding on a guessed name.
        """
        if self._node:
            self._node_name = self._node
            return self._node_name
-        # Try the node's hostname via API cluster/resources
+        members = [s for s in (self._api.cluster.status.get() or []) if s.get("type") == "node"]
-        try:
+        short_host = self.hostname.split(".")[0]
-            nodes = self._api.nodes.get()
+        for pick in (
-            # Match by hostname or IP
+            lambda s: s.get("local"),
-            for n in nodes:
+            lambda s: s.get("ip") == self.hostname,
-                n_node = n.get("node", "")
+            lambda s: s.get("name") in (self.hostname, short_host),
-                if n_node:
+        ):
-                    # First match: the node exists in the cluster
+            match = next((s for s in members if pick(s)), None)
-                    self._node_name = n_node
+            if match:
                self._node_name = match["name"]
                return self._node_name
        except Exception:
            pass
-        # Fallback: use the configured hostname as node name
+        names = [n.get("node") for n in (self._api.nodes.get() or []) if n.get("node")]
-        # (may not match the PVE node name — SSH-based methods will fail,
+        if len(names) == 1:
-        # but API methods that target a specific node name require correct
+            self._node_name = names[0]
        # resolution)
        self._node_name = self.hostname
            return self._node_name
        raise ConnectionException(
            f"Cannot tell which PVE node {self.hostname} is among {names}; "
            f"set the 'node' driver argument"
        )
    def close(self) -> None:
        """Close the connection."""
        self._api = None
@@ -219,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:
@@ -236,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}
    # ------------------------------------------------------------------ #
@@ -262,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:
@@ -280,39 +318,51 @@ class ProxmoxDriver(
        # Fallback: direct paramiko SSH
        try:
-            import paramiko
+            _, stdout, stderr = self._node_ssh().exec_command(command, timeout=self.timeout)
            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)
            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
    # ------------------------------------------------------------------ #
@@ -383,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_list = sorted(iface["iface"] for iface in network if iface.get("iface"))
            iface["iface"] for iface in network if iface.get("iface")
        )
        pve_version = version.get("version", "")
        release = version.get("release", "")
@@ -412,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:
@@ -462,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}
@@ -174,8 +174,13 @@ class ProxmoxInterfaceMixin:
            parts = line.split()
            if len(parts) < 6:
                continue
-            ip_addr, _, flags, mac, _, iface = (
+            ip_addr, _, _flags, mac, _, iface = (
-                parts[0], parts[1], parts[2], parts[3], parts[4], parts[5]
+                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(
+            raw = self._exec_ssh_command(f"bridge fdb show br {bridge} 2>/dev/null || true")
                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(
+            raw = self._exec_ssh_command(f"ovs-appctl fdb/show {bridge} 2>/dev/null || true")
                f"ovs-appctl fdb/show {bridge} 2>/dev/null || true"
            )
            for line in raw.splitlines():
                parts = line.split()
                if len(parts) < 4:
@@ -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(
+        raw = self._exec_ssh_command("lldpcli show neighbors summary 2>/dev/null || true")
            "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(
+        raw = self._exec_ssh_command("lldpcli show neighbors details 2>/dev/null || true")
            "lldpcli show neighbors details 2>/dev/null || true"
        )
        current_iface = ""
        current_entry: _JsonDict = {}
@@ -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(
+        for m_probe in re.finditer(r"icmp_seq=\d+.*?time=([\d.]+) ms.*?from ([\d.a-fA-F:]+)", raw):
-            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))})
        ):
            probes.append(
                {"ip_address": m_probe.group(2), "rtt": float(m_probe.group(1))}
            )
        return {
            "success": {
@@ -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
+            iface["iface"]: iface for iface in self._get_node_network() if iface.get("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 (
+                if n.get("name") == self._node_name and int(n.get("nodeid", 9999)) == min_id:
                    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 = (
+        name: str = (config.get("name") or f"VLAN{vlan_id}") if config else f"VLAN{vlan_id}"
            (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()
@@ -12,20 +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_proxmox import utils
+from napalm_device_types import APT_UPGRADABLE_COMMAND, parse_apt_upgradable
 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."""
@@ -136,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(
+        raw = self._exec_ssh_command("cat /etc/snmp/snmpd.conf 2>/dev/null || true")
            "cat /etc/snmp/snmpd.conf 2>/dev/null || true"
        )
        communities: dict[str, _JsonDict] = {}
        location = ""
        contact = ""
@@ -187,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]
@@ -197,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": [],
            }
@@ -215,12 +255,39 @@ 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": [],
                }
        return result
    # ------------------------------------------------------------------ #
    # Kernel facts (KernelFactsMixin supplies get_kernel_facts)
    # ------------------------------------------------------------------ #
    def _run_kernel_facts_command(self, command: str) -> str:
        """The transport for ``KernelFactsMixin.get_kernel_facts``: the exec path."""
        return self._exec_ssh_command(command)
    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."""
        return str(self._exec_ssh_command(command))
    # ------------------------------------------------------------------ #
    # Packages (Debian APT)
    # ------------------------------------------------------------------ #
@@ -277,7 +344,8 @@ class ProxmoxSystemMixin:
            if not name or status != "installed":
                continue
            size_bytes = int(size_kb) * 1024 if size_kb.isdigit() else 0
-            result.append({
+            result.append(
                {
                    "name": name,
                    "version": version,
                    "installed": True,
@@ -287,160 +355,102 @@ class ProxmoxSystemMixin:
                    "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]:
+    def get_licenses(self) -> list[_JsonDict]:
-        """Return warnings for the Proxmox node.
+        """The node's subscription, as the API reports it.
-        Currently detects:
+        A fact for the caller: whether running without one is a problem is not
-        - lldpd not installed (LLDP neighbor discovery unavailable)
+        the driver's to say. An unreadable subscription reports nothing.
        - Available package updates (via Proxmox APT API)
        - Missing / invalid subscription
        """
        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", "")
+        raw = str(sub.get("status", ""))
-            if status in ("NotFound", "Invalid", "Expired"):
+        if not raw:
-                warnings.append({"code": "no_subscription", "meta": {"status": status}})
+            return []
-        except Exception as exc:
+        lic: _JsonDict = {
-            logger.debug("Failed to check subscription status: %s", exc)
+            "name": "pve-subscription",
-
+            "status": _SUBSCRIPTION_STATUS.get(raw.lower(), "unknown"),
-        return warnings
+            "detail": sub.get("message") or raw,
    # ------------------------------------------------------------------ #
    # Services (systemd)
    # ------------------------------------------------------------------ #
    def get_services(self) -> list[_JsonDict]:
        """Return systemd services with running and enabled state.
        Uses two ``systemctl`` invocations combined in a single SSH command:
        - ``list-unit-files`` for the static enabled/disabled state
        - ``list-units`` for the live running state
        """
        raw = self._exec_ssh_command(
            "{ systemctl list-unit-files --type=service --no-pager --no-legend --full 2>/dev/null;"
            " echo '---UNITS---';"
            " systemctl list-units --type=service --all --no-pager --no-legend --full 2>/dev/null;"
            " } || true"
        )
        # Parse enabled state from list-unit-files
        enabled_map: dict[str, bool] = {}
        section = "files"
        for line in raw.splitlines():
            if line.strip() == "---UNITS---":
                section = "units"
                continue
            parts = line.strip().split(None, 1)
            if len(parts) < 1:
                continue
            unit = parts[0].lstrip("\u25cf").strip()
            if not unit.endswith(".service"):
                continue
            name = unit[: -len(".service")]
            if section == "files":
                state = parts[1].strip() if len(parts) > 1 else ""
                enabled_map[name] = state in ("enabled", "enabled-runtime", "static")
        # Parse running state from list-units
        running_map: dict[str, bool] = {}
        section = "files"
        for line in raw.splitlines():
            if line.strip() == "---UNITS---":
                section = "units"
                continue
            if section != "units":
                continue
            parts = line.strip().lstrip("\u25cf").strip().split(None, 4)
            if len(parts) < 4:
                continue
            unit = parts[0]
            if not unit.endswith(".service"):
                continue
            name = unit[: -len(".service")]
            sub_state = parts[3]
            running_map[name] = sub_state == "running"
        all_names = sorted(set(enabled_map) | set(running_map))
        return [
            {
                "name": name,
                "running": running_map.get(name, False),
                "enabled": enabled_map.get(name, False),
                "pid": 0,
        }
-            for name in all_names
+        if sub.get("productname"):
-        ]
+            lic["edition"] = sub["productname"]
        if sub.get("nextduedate"):
            lic["expires"] = sub["nextduedate"]
        return [lic]
-    def manage_service(self, name: str, action: str) -> _JsonDict:
+    # ------------------------------------------------------------------ #
-        """Start / stop / restart / enable / disable a systemd service."""
+    # Services (systemd, through napalm-device-types' SystemdServicesMixin)
-        if not re.match(r'^[a-zA-Z0-9_\-\.@]+$', name):
+    # ------------------------------------------------------------------ #
-            raise ValueError(f"Invalid service name: {name!r}")
+
-        if action not in ('start', 'stop', 'restart', 'enable', 'disable'):
+    def _run_service_command(self, command: str, *, privileged: bool, timeout: int) -> str:
-            raise ValueError(f"Invalid action: {action!r}")
+        """The transport for ``SystemdServicesMixin``: the exec path, as root.
-        output = self._exec_ssh_command(f"systemctl {action} {name}.service 2>&1 || true")
+
-        return {"success": True, "output": output}
+        *privileged* needs nothing more on a node the driver reaches as root, and
        the exec path keeps its own timeout.
        """
        return str(self._exec_ssh_command(command))
    # ------------------------------------------------------------------ #
    # Available updates
    # ------------------------------------------------------------------ #
    def get_available_updates(self) -> list[_JsonDict]:
-        """Return list of upgradable packages from the Proxmox APT API."""
+        """Return the node's upgradable packages, with origin and security status.
-        updates: list[_JsonDict] = []
+
        ``apt list --upgradable`` over the exec path names each candidate's suite
        (``trixie-security``); the APT API names only an Origin ("Debian",
        "Proxmox") and is the fallback, with the security status unknown.
        :raises Exception: when neither answers -- never an empty list for
            "could not read".
        """
        try:
-            for upd in self._api.nodes(self._node_name).apt.update.get():
+            updates = parse_apt_upgradable(self._exec_ssh_command(APT_UPGRADABLE_COMMAND) or "")
-                pkg = upd.get("Package", "")
+        except ValueError as exc:
-                if not pkg:
+            logger.debug("apt list over the exec path failed, using the API: %s", exc)
-                    continue
+            updates = self._updates_from_api()
-                updates.append({
+        return sorted(updates, key=lambda u: u["name"])
-                    "name": pkg,
+
    def _updates_from_api(self) -> list[_JsonDict]:
        return [
            {
                "name": upd["Package"],
                "current_version": upd.get("OldVersion", ""),
                "new_version": upd.get("Version", ""),
-                })
+                "origin": upd.get("Origin"),
                "security": None,
            }
            for upd in self._api.nodes(self._node_name).apt.update.get()  # type: ignore[union-attr]
            if upd.get("Package")
        ]
    def refresh_available_updates(self) -> _JsonDict:
        """Resynchronise the node's package index (``POST nodes/{n}/apt/update``)."""
        try:
            task = self._api.nodes(self._node_name).apt.update.post()  # type: ignore[union-attr]
        except Exception as exc:
-            logger.debug("Failed to fetch available updates: %s", exc)
+            return {"success": False, "output": str(exc)}
-        return sorted(updates, key=lambda u: u["name"])
+        return {"success": True, "output": f"Package index refresh started ({task})"}
    def _run_host_status_command(self, command: str) -> str:
        """The transport for ``HostStatusMixin.get_host_status``: the exec path."""
        return str(self._exec_ssh_command(command))
    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}
@@ -470,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()
@@ -488,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."""
+    # Fixes (napalm-device-types SnmpAgentMixin, EnsurePackageMixin): the
-        if action == "fix_snmp":
+    # caller's values. The node's shell runs as root.
-            return self._action_fix_snmp()
+    # ------------------------------------------------------------------
        if action == "install_lldpd":
            return self._action_install_lldpd()
        raise NotImplementedError(f"Unknown action: {action!r}")
-    def _action_install_lldpd(self) -> dict:
+    def _rc(self, command: str) -> tuple[str, int]:
-        """Install, enable and start lldpd on the Proxmox node.
+        """Run *command* and return its output and exit status (-1 when unknown)."""
-
+        out = self._exec_ssh_command(f"{{ {command}; }} 2>&1; echo __PVE_RC=$?")
-        Proxmox runs Debian/Linux underneath, so this is a plain apt install
+        text, _, tail = out.rpartition("__PVE_RC=")
        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 = ""
        try:
-            raw = self._exec_ssh_command(
+            return text.strip(), int(tail.strip())
-                "ss -tnp 2>/dev/null | awk '/sshd/{print $5}' | head -1 | cut -d: -f1"
+        except ValueError:
-            ).strip()
+            return out.strip(), -1
            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)
-        # 3. Write snmpd.conf via /tmp (no permission issues)
+    def _step(self, command: str, done: str) -> dict:
-        conf_str = (
+        output, status = self._rc(command)
-            "agentAddress udp:161\n"
+        return {"ok": status == 0, "output": done if status == 0 else output[-300:]}
            "rocommunity public\n"
            "sysLocation    Managed by netOrk\n"
            "sysContact     netork@localhost\n"
        )
        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"
        )
        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
+    # apt's proxy (napalm-device-types AptProxyMixin): the transports. The node
-        if netork_ip:
+    # 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}",
        )
    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:
-                ufw = self._exec_ssh_command("command -v ufw 2>/dev/null").strip()
+            return str(ipaddress.ip_address(out[0])) if out else None
-                if ufw:
+        except ValueError:
-                    parts = netork_ip.rsplit(".", 1)
+            return None
-                    subnet = f"{parts[0]}.0/24" if len(parts) == 2 else netork_ip
+
-                    fw_out = self._exec_ssh_command(
+    def _install_snmp_agent(self) -> dict:
-                        f"ufw allow from {subnet} to any port 161 proto udp 2>&1"
+        return self._step(
            "DEBIAN_FRONTEND=noninteractive apt-get install -y snmpd snmp", "net-snmp installed"
        )
                    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)
+    def _write_snmp_agent_config(self, settings: dict) -> dict:
-        self._exec_ssh_command(
+        encoded = base64.b64encode(render_snmpd_conf(settings).encode()).decode()
-            "service snmpd stop 2>/dev/null; pkill -9 snmpd 2>/dev/null; true"
+        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)
+    def _allow_snmp_sources(self, settings: dict) -> dict:
-        start_out = self._exec_ssh_command(
+        sources = [_net(s) for s in settings.get("allowed_sources", [])]
-            "service snmpd start 2>&1 || systemctl start snmpd 2>&1 || true"
+        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.
@@ -650,9 +618,11 @@ class ProxmoxSystemMixin:
            fqdn = new_hostname
        # Derive old short hostname for substitution in /etc/hosts
-        old_short = self._exec_ssh_command(
+        old_short = (
-            "cat /etc/hostname 2>/dev/null || hostname -s 2>/dev/null"
+            self._exec_ssh_command("cat /etc/hostname 2>/dev/null || hostname -s 2>/dev/null")
-        ).strip().split(".")[0]
+            .strip()
            .split(".")[0]
        )
        # ── 1. /etc/hostname ─────────────────────────────────────────────
        hostname_b64 = _b64.b64encode(f"{short_hostname}\n".encode()).decode()
@@ -667,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",
@@ -683,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"[ -f /etc/mailname ] && echo {mailname_b64} | base64 -d > /etc/mailname || true"
            f" && echo {mailname_b64} | base64 -d > /etc/mailname"
            f" || true"
        )
        # ── 4. Postfix myhostname ─────────────────────────────────────────
        self._exec_ssh_command(
-            f"command -v postconf >/dev/null 2>&1"
+            f"command -v postconf >/dev/null 2>&1 && postconf -e 'myhostname = {fqdn}' || true"
            f" && postconf -e 'myhostname = {fqdn}'"
            f" || true"
        )
        # ── 5. Apply hostname at runtime ─────────────────────────────────
@@ -735,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,
        )
    # ------------------------------------------------------------------ #
@@ -846,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
@@ -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.
@@ -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(
+    def get_vm_interfaces(self, vmid: int, vm_type: str) -> tuple[dict[str, _JsonDict], bool, bool]:
        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,7 +252,8 @@ class ProxmoxVMMixin:
                )
                disks, onboot = self._get_vm_disk_and_boot(vmid, "qemu")
-                result.append({
+                result.append(
                    {
                        "vmid": str(vmid),
                        "name": name,
                        "type": "vm",
@@ -273,13 +270,14 @@ class ProxmoxVMMixin:
                        "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,7 +298,8 @@ class ProxmoxVMMixin:
                )
                disks, onboot = self._get_vm_disk_and_boot(vmid, "lxc")
-                result.append({
+                result.append(
                    {
                        "vmid": str(vmid),
                        "name": name,
                        "type": "container",
@@ -317,7 +316,8 @@ class ProxmoxVMMixin:
                        "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(
+    def _get_vm_disk_and_boot(self, vmid: int, vm_type: str) -> tuple[list[dict], bool]:
        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_VM = {"start", "stop", "shutdown", "reboot", "reset"}
-    _POWER_ACTIONS_CT = {'start', 'stop', 'shutdown', 'reboot'}
+    _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:
@@ -4,30 +4,136 @@ from __future__ import annotations
 import base64
 import hashlib
 import ipaddress
 import logging
 import re
 import time
-import yaml
+from typing import TYPE_CHECKING, Any
 from typing import Any, Dict, List
 from urllib.parse import quote
 import yaml
 from napalm_device_types.models import (
    NetworkTargetDict,
    StorageTargetDict,
    VMProvisionResultDict,
    VMStatusDict,
 )
 from napalm_device_types.provisioning import (
    QEMU_GUEST_AGENTS,
    GuestAgents,
    network_config,
    split_compression,
 )
 if TYPE_CHECKING:
    # Type-only: VMCpuTypeDict is newer than the napalm_device_types floor in
    # pyproject.toml, and nothing here needs it at runtime.
    from napalm_device_types.models import VMCpuTypeDict
 _logger = logging.getLogger(__name__)
 # The CPU models a new VM may be given. Each lists the /proc/cpuinfo flags it
 # adds on top of QEMU's qemu64 baseline: what the node's CPU must have for the
 # model to start at all, and what a guest can count on. The sets follow
 # Proxmox's own x86-64-v* definitions (qemu-server, PVE/QemuServer/CPUConfig.pm).
 #
 # Leaving the model out of qemu.post is not neutral: Proxmox then falls back to
 # kvm64, which lacks even AES-NI, let alone the AVX MongoDB 5.0+ needs
 # (netOrk#494). The default below is what the Proxmox GUI picks since PVE 8.
 _X86_64_V2_AES_FLAGS = ("aes", "popcnt", "pni", "sse4_1", "sse4_2", "ssse3")
 _X86_64_V3_FLAGS = _X86_64_V2_AES_FLAGS + (
    "avx",
    "avx2",
    "bmi1",
    "bmi2",
    "f16c",
    "fma",
    "abm",
    "movbe",
    "xsave",
 )
 _DEFAULT_CPU_TYPE = "x86-64-v2-AES"
 _CPU_MODELS = (
    (
        "x86-64-v2-AES",
        _X86_64_V2_AES_FLAGS,
        "Proxmox's own default: runs on practically any x86-64 server CPU and "
        "can live-migrate between different ones. No AVX.",
    ),
    (
        "x86-64-v3",
        _X86_64_V3_FLAGS,
        "Adds AVX and AVX2 (which MongoDB 5.0 and later need). Every node the VM "
        "may run on needs an Intel Haswell or AMD Excavator CPU (2013) or newer.",
    ),
 )
 _HOST_CPU_DESCRIPTION = (
    "This node's CPU, passed through unchanged: fastest, with every feature it "
    "has, but the VM can only live-migrate to nodes with the same CPU."
 )
 # Downloaded cloud images are cached here on the hypervisor node, keyed by
 # filename, so provisioning multiple VMs from the same image only pays the
 # download cost once.
 _IMAGE_CACHE_DIR = "/var/lib/vz/template/netork-images"
 # Proxmox reads an import volume's format off its extension and accepts only
 # these. Ubuntu ships its qcow2 cloud images as ".img", so that is stored as
 # qcow2: if an .img is really raw, the import fails loudly, whereas guessing
 # raw for a qcow2 would attach the qcow2 container as a raw disk silently.
 _IMPORT_EXTENSIONS = {"qcow2": "qcow2", "raw": "raw", "vmdk": "vmdk", "img": "qcow2"}
 # The VM shell each guest needs. OpenBSD's qemu-ga cannot use virtio-serial,
 # and over ISA serial it answers only as the second port (cua01): the OpenBSD
 # cloud image keeps its console on com0, which serial0 takes (netork#793).
 _GUEST_VM_SETTINGS: dict[str, dict[str, str]] = {
    "linux": {"ostype": "l26", "agent": "1"},
    "freebsd": {"ostype": "other", "agent": "1"},
    "openbsd": {"ostype": "other", "agent": "1,type=isa", "serial0": "socket"},
 }
 # A NIC as Proxmox reports it: "virtio=BC:24:11:AA:BB:02,bridge=vmbr0".
 _NIC_MAC = re.compile(r"^[a-z0-9]+=([0-9A-Fa-f:]{17})")
 # Characters Proxmox keeps in a content file name (PVE::Storage's
 # SAFE_CHAR_CLASS_RE); anything else it would rewrite behind our back.
 _UNSAFE_FILENAME_CHARS = re.compile(r"[^A-Za-z0-9\-.+=_]")
 def _url_key(image_url: str) -> str:
    """Short hash of the full URL — Ubuntu and others publish a new build
    under the same basename every day, so the basename alone is no cache key."""
    return hashlib.sha256(image_url.encode()).hexdigest()[:12]
 def _split_checksum(image_checksum: str) -> tuple[str, str]:
    """``"<algo>:<hex>"`` as ``(algo, hex)``; the algorithm defaults to sha256."""
    algo, _, expected = image_checksum.partition(":")
    return (algo or "sha256").lower(), expected
 def _import_volume_name(image_url: str) -> str | None:
    """The file name *image_url* gets in an import storage.
    None when Proxmox cannot import the image's type at all (compressed,
    ISO, no extension) — provisioning then downloads it over SSH as before.
    """
    basename = image_url.rstrip("/").rsplit("/", 1)[-1]
    stem, dot, ext = basename.rpartition(".")
    extension = _IMPORT_EXTENSIONS.get(ext.lower()) if dot and stem else None
    if extension is None:
        return None
    safe_stem = _UNSAFE_FILENAME_CHARS.sub("_", stem)
    return f"netork-{_url_key(image_url)}-{safe_stem}.{extension}"
 class ProxmoxVMProvisionMixin:
    """Mixin to add VM provisioning to ProxmoxDriver."""
    #: Every guest QEMU's agent runs on; see _GUEST_VM_SETTINGS for the VM
    #: hardware each one needs.
    GUEST_AGENTS: GuestAgents = QEMU_GUEST_AGENTS
    def _run_node_command(self, command: str, timeout: int) -> str:
        """
        Execute a shell command on the Proxmox node via SSH, raising on failure.
@@ -37,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
+        _, stdout, stderr = self._node_ssh().exec_command(command, timeout=timeout)
        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)
        exit_status = stdout.channel.recv_exit_status()
        out = stdout.read().decode().strip()
        err = stderr.read().decode().strip()
@@ -89,11 +171,9 @@ class ProxmoxVMProvisionMixin:
        before raising.
        """
        filename = image_url.rstrip("/").rsplit("/", 1)[-1]
-        url_hash = hashlib.sha256(image_url.encode()).hexdigest()[:12]
+        local_path = f"{_IMAGE_CACHE_DIR}/{_url_key(image_url)}-{filename}"
        local_path = f"{_IMAGE_CACHE_DIR}/{url_hash}-{filename}"
-        algo, _, expected = (image_checksum or "").partition(":")
+        algo, expected = _split_checksum(image_checksum or "")
        algo = (algo or "sha256").lower()
        max_attempts = 2
        for attempt in range(1, max_attempts + 1):
@@ -133,6 +213,180 @@ class ProxmoxVMProvisionMixin:
        raise AssertionError("unreachable")  # loop always returns or raises above
    def _cloud_image_on_node(self, image_url: str, image_checksum: str | None, timeout: int) -> str:
        """The node path of the image, downloaded, verified and unpacked.
        A packed image (FreeBSD ships .qcow2.xz) is unpacked on the node after
        its checksum, which covers the packed file, has been verified; only
        the unpacked file is kept. Proxmox's download-url unpacks ISOs only,
        so a packed image always comes this way.
        """
        filename = image_url.rstrip("/").rsplit("/", 1)[-1]
        unpacked_name, unpack = split_compression(filename)
        if unpack is None:
            return self._download_cloud_image(image_url, image_checksum, timeout=timeout)
        unpacked = f"{_IMAGE_CACHE_DIR}/{_url_key(image_url)}-{unpacked_name}"
        cached = self._run_node_command(
            f"mkdir -p {_IMAGE_CACHE_DIR} && test -f {unpacked} && echo EXISTS || echo MISSING",
            timeout=30,
        )
        if "EXISTS" in cached:
            return unpacked
        packed = self._download_cloud_image(image_url, image_checksum, timeout=timeout)
        _logger.info(f"Unpacking {packed} -> {unpacked}")
        self._run_node_command(
            f"{unpack} {packed} > {unpacked}.tmp && mv {unpacked}.tmp {unpacked} && rm -f {packed}",
            timeout=timeout,
        )
        return unpacked
    def _write_snippet(self, storage_path: str, filename: str, content: str) -> None:
        """Write a cloud-init snippet into *storage_path*/snippets on the node.
        Proxmox's /storage/{s}/upload API only accepts content in
        {iso, vztmpl, import} — "snippets" is rejected outright ("does not
        have a value in the enumeration"). Snippets can only be written
        directly to the filesystem, so this goes over SSH.
        """
        _logger.debug(f"Writing Cloud-Init snippet {filename} to {storage_path}/snippets")
        encoded = base64.b64encode(content.encode("utf-8")).decode("ascii")
        self._run_node_command(
            f"mkdir -p {storage_path}/snippets && "
            f"echo {encoded} | base64 -d > {storage_path}/snippets/{filename}",
            timeout=30,
        )
    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 = []
        for i, nic in enumerate(nics):
            match = _NIC_MAC.match(config.get(f"net{i}", ""))
            if not match:
                raise RuntimeError(f"VM {vmid} has no MAC address on net{i}")
            macs.append((match.group(1), nic.get("dhcp", i == 0)))
        return macs
    def _find_import_storage(self) -> str | None:
        """The first storage on this node that accepts content "import".
        Such a storage (Proxmox 8.2+) can take a cloud image straight from its
        URL. Inactive storages (typically a share that is not mounted) are
        skipped rather than failed on: the SSH path still works without them.
        Node-scoped for the same reason as _find_default_image_storage.
        """
        for storage in self._node_api().storage.get():
            content = storage.get("content", "").split(",")
            if "import" not in content:
                continue
            if storage.get("enabled", 1) == 0 or storage.get("active", 1) == 0:
                continue
            return storage["storage"]
        return None
    def _import_cloud_image(
        self,
        storage: str,
        filename: str,
        image_url: str,
        image_checksum: str | None,
        timeout: int,
    ) -> str:
        """Have Proxmox download *image_url* into *storage*; returns the volume id.
        Proxmox runs the download as a task and verifies the checksum itself.
        A file already in the storage is reused — the name carries a hash of
        the full URL, and Proxmox only creates it once the download (and its
        checksum check) has succeeded, so an existing file is a complete one.
        """
        volid = f"{storage}:import/{filename}"
        present = self._node_api().storage(storage).content.get(content="import")
        if any(item.get("volid") == volid for item in present):
            _logger.info(f"Cloud image already in {storage}: {volid}")
            return volid
        params: dict[str, Any] = {"url": image_url, "content": "import", "filename": filename}
        if image_checksum:
            algo, expected = _split_checksum(image_checksum)
            params["checksum"] = expected
            params["checksum-algorithm"] = algo
        _logger.info(f"Downloading cloud image {image_url} into {volid}")
        upid = self._node_api().storage(storage)("download-url").post(**params)
        self._wait_for_task(upid, timeout=timeout)
        return volid
    def _import_over_ssh(
        self,
        vmid: int,
        image_storage: str,
        image_url: str,
        image_checksum: str | None,
        download_timeout: int,
        timeout: int,
    ) -> None:
        """Download the image on the node and import it as the root disk.
        The path for nodes without an import storage, or for an image type
        Proxmox cannot import itself.
        """
        local_path = self._cloud_image_on_node(image_url, image_checksum, download_timeout)
        _logger.info(f"Importing {local_path} into VM {vmid} on storage {image_storage}")
        self._run_node_command(
            f"qm importdisk {vmid} {local_path} {image_storage} --format qcow2",
            timeout=timeout,
        )
        # Proxmox leaves the imported disk as an "unusedN" reference — find
        # it and attach it as the boot disk.
        imported_config = self._node_api().qemu(vmid).config.get()
        unused_value = next((v for k, v in imported_config.items() if k.startswith("unused")), None)
        if not unused_value:
            raise RuntimeError(
                f"Disk import for VM {vmid} did not produce an unused disk reference"
            )
        self._node_api().qemu(vmid).config.post(
            scsi0=f"{unused_value},discard=on",
            boot="order=scsi0",
        )
    def _attach_root_disk(
        self,
        vmid: int,
        image_storage: str,
        image_url: str,
        image_checksum: str | None,
        download_timeout: int,
        timeout: int,
    ) -> None:
        """Get the cloud image onto the node and make it the VM's root disk.
        Through an import storage when the node has one — Proxmox downloads,
        verifies and copies the image itself — and over SSH otherwise.
        """
        import_storage = self._find_import_storage()
        filename = _import_volume_name(image_url) if import_storage else None
        if not import_storage or not filename:
            self._import_over_ssh(
                vmid, image_storage, image_url, image_checksum, download_timeout, timeout
            )
            return
        volid = self._import_cloud_image(
            import_storage, filename, image_url, image_checksum, timeout=download_timeout
        )
        _logger.info(f"Importing {volid} into VM {vmid} on storage {image_storage}")
        upid = (
            self._node_api()
            .qemu(vmid)
            .config.post(
                scsi0=f"{image_storage}:0,import-from={volid},discard=on",
                boot="order=scsi0",
            )
        )
        if upid:
            self._wait_for_task(upid, timeout=timeout)
    def _find_default_image_storage(self) -> str:
        """Find a storage suitable for VM root disks (content includes 'images').
@@ -175,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:
@@ -196,6 +450,56 @@ class ProxmoxVMProvisionMixin:
            )
        return targets
    def get_vm_cpu_types(self) -> list[VMCpuTypeDict]:
        """List the CPU models a new VM may be given, judged against this node's CPU."""
        return self._cpu_types_for(self._node_cpu_flags())
    def _node_cpu_flags(self) -> set[str]:
        status = self._node_api().status.get() or {}
        return set(str((status.get("cpuinfo") or {}).get("flags", "")).split())
    @staticmethod
    def _cpu_types_for(node_flags: set[str]) -> list[VMCpuTypeDict]:
        types: list[VMCpuTypeDict] = [
            {
                "name": name,
                "description": description,
                "features": list(flags),
                "available": set(flags) <= node_flags,
                "default": name == _DEFAULT_CPU_TYPE,
            }
            for name, flags, description in _CPU_MODELS
        ]
        types.append(
            {
                "name": "host",
                "description": _HOST_CPU_DESCRIPTION,
                "features": sorted(node_flags),
                "available": True,
                "default": False,
            }
        )
        return types
    def _resolve_cpu_type(self, cpu_type: str | None) -> str:
        """The model to create the VM with. A model asked for by name is checked
        against this node's CPU first: one it cannot run would fail only at VM
        start, after the disk import, leaving a half-built VM behind."""
        if cpu_type is None:
            return _DEFAULT_CPU_TYPE
        node_flags = self._node_cpu_flags()
        offered = {t["name"]: t for t in self._cpu_types_for(node_flags)}
        entry = offered.get(cpu_type)
        if entry is None:
            raise ValueError(f"Unknown CPU type {cpu_type!r}; choose one of {', '.join(offered)}")
        if not entry["available"]:
            missing = sorted(set(entry["features"]) - node_flags)
            raise ValueError(
                f"CPU type {cpu_type!r} needs {', '.join(missing)}, which the CPU of "
                f"node {self._node_name} does not have"
            )
        return cpu_type
    def _wait_for_task(self, upid: str, timeout: int = 120) -> None:
        """
        Poll a Proxmox task until completion.
@@ -234,12 +538,14 @@ class ProxmoxVMProvisionMixin:
        image_url: str,
        cpu: int,
        memory: int,
-        nics: List[Dict[str, Any]],
+        nics: list[dict[str, Any]],
-        cloud_init_config: 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,
        guest_os: str = "linux",
        download_timeout: int = 300,
        timeout: int = 180,
    ) -> VMProvisionResultDict:
@@ -272,6 +578,11 @@ class ProxmoxVMProvisionMixin:
            disk_resize_gb: resize root disk to this size (None = no resize)
            storage: storage pool for the root disk (None = auto-detect first
                enabled, node-available storage with content='images')
            cpu_type: CPU model from get_vm_cpu_types() (None = x86-64-v2-AES)
            guest_os: the OS in the image, a key of GUEST_AGENTS. It sets the
                OS type and how the agent is attached; a guest other than Linux
                also gets a network-config v2 snippet of its own, because
                FreeBSD's nuageinit skips runcmd on the v1 Proxmox writes.
            download_timeout: max seconds for the image download (skipped if cached)
            timeout: max seconds for the remaining provisioning steps
@@ -280,10 +591,18 @@ class ProxmoxVMProvisionMixin:
        Raises:
            RuntimeError: provisioning failure (download, import, config, timeout, etc.)
-            ValueError: invalid storage or configuration
+            ValueError: invalid storage or configuration, or a cpu_type that is
                unknown or that this node's CPU cannot run (raised before
                anything is created)
        """
        if guest_os not in self.GUEST_AGENTS:
            raise ValueError(
                f"Cannot provision a {guest_os!r} guest; this driver provisions "
                f"{', '.join(sorted(self.GUEST_AGENTS))}"
            )
        try:
-            _logger.info(f"Creating VM '{name}' from image {image_url}")
+            _logger.info(f"Creating {guest_os} VM '{name}' from image {image_url}")
            cpu_model = self._resolve_cpu_type(cpu_type)
            # Step 1: Get next VMID
            next_vmid = self._api.cluster.nextid.get()
@@ -297,45 +616,24 @@ class ProxmoxVMProvisionMixin:
                name=name,
                memory=memory,
                cores=cpu,
-                ostype="l26",
+                cpu=cpu_model,
                scsihw="virtio-scsi-pci",
-                # Without this, Proxmox never attaches the virtio-serial
+                # Includes agent: without it Proxmox never attaches the
-                # channel the QEMU guest agent needs — get_vm_status's
+                # channel the QEMU guest agent needs — get_vm_status's agent
-                # agent queries (below) would have nothing to talk to.
+                # queries (below) would have nothing to talk to.
-                agent="1",
+                **_GUEST_VM_SETTINGS[guest_os],
            )
            # Step 3: Download cloud image (cached) and import as root disk
            local_path = self._download_cloud_image(
                image_url, image_checksum, timeout=download_timeout
            )
            image_storage = storage or self._find_default_image_storage()
-
+            self._attach_root_disk(
-            _logger.info(f"Importing {local_path} into VM {vmid} on storage {image_storage}")
+                vmid, image_storage, image_url, image_checksum, download_timeout, timeout
            self._run_node_command(
                f"qm importdisk {vmid} {local_path} {image_storage} --format qcow2",
                timeout=timeout,
            )
            # Proxmox leaves the imported disk as an "unusedN" reference — find
            # it and attach it as the boot disk.
            imported_config = self._node_api().qemu(vmid).config.get()
            unused_value = next(
                (v for k, v in imported_config.items() if k.startswith("unused")), None
            )
            if not unused_value:
                raise RuntimeError(
                    f"Disk import for VM {vmid} did not produce an unused disk reference"
                )
            self._node_api().qemu(vmid).config.post(
                scsi0=f"{unused_value},discard=on",
                boot="order=scsi0",
            )
            # 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):
@@ -386,20 +684,20 @@ class ProxmoxVMProvisionMixin:
            )
            filename = f"{vmid}-user-data.yaml"
            _logger.debug(f"Writing Cloud-Init snippet {filename} to {snippet_storage}")
            # Proxmox's /storage/{s}/upload API only accepts content in
            # {iso, vztmpl, import} — "snippets" is rejected outright
            # ("does not have a value in the enumeration"). Snippets can only
            # be written directly to the filesystem, so resolve the storage's
            # backing path and write the file over SSH instead.
            storage_path = self._get_storage_path(snippet_storage)
-            encoded = base64.b64encode(user_data_yaml.encode("utf-8")).decode("ascii")
+            self._write_snippet(storage_path, filename, user_data_yaml)
-            self._run_node_command(
+            cicustom = f"user={snippet_storage}:snippets/{filename}"
-                f"mkdir -p {storage_path}/snippets && "
+
-                f"echo {encoded} | base64 -d > {storage_path}/snippets/{filename}",
+            # Proxmox writes network-config v1, and FreeBSD's nuageinit fails
-                timeout=30,
+            # on it and then skips runcmd, which starts the guest agent. A
            # guest other than Linux gets the v2 every cloud-init reads.
            network = network_config(self._nic_macs(vmid, nics)) if guest_os != "linux" else None
            if network:
                network_filename = f"{vmid}-network-config.yaml"
                self._write_snippet(
                    storage_path, network_filename, yaml.safe_dump(network, sort_keys=False)
                )
                cicustom += f",network={snippet_storage}:snippets/{network_filename}"
            # Step 7: Configure Cloud-Init references and SSH keys
            _logger.info(f"Setting Cloud-Init config for VM {vmid}")
@@ -413,7 +711,7 @@ class ProxmoxVMProvisionMixin:
                # this points at a snippets-only storage.
                "ide2": f"{image_storage}:cloudinit",
                "citype": "nocloud",
-                "cicustom": f"user={snippet_storage}:snippets/{filename}",
+                "cicustom": cicustom,
            }
            # Configure DHCP for NICs where enabled (default True for index 0, False otherwise)
@@ -501,6 +799,10 @@ class ProxmoxVMProvisionMixin:
            except Exception as e:
                _logger.debug(f"VM {vmid} stop failed (may already be stopped): {e}")
            # The snippets have to be known before the delete: afterwards the
            # VM's config is gone (napalm-proxmox#15).
            snippets = self._cicustom_snippets(vmid_int)
            # Step 2: Delete VM
            # Proxmox's API parameter is hyphenated (destroy-unreferenced-disks),
            # not a valid Python identifier — proxmoxer forwards kwargs to the
@@ -515,21 +817,12 @@ class ProxmoxVMProvisionMixin:
            # Step 3: Clean up Cloud-Init snippets
            # (This is best-effort; snippet files may be unreachable if storage is unavailable)
-            try:
+            for storage, filepath in snippets:
                config = self._node_api().qemu(vmid_int).config.get()
                cicustom = config.get("cicustom", "")
                if "snippets/" in cicustom:
                    parts = cicustom.split("=")
                    if len(parts) >= 2:
                        snippet_ref = parts[1]  # e.g. "snippets:snippets/101-user-data.yaml"
                        storage, filepath = snippet_ref.split(":", 1)
                _logger.debug(f"Deleting snippet {filepath} from {storage}")
                try:
                    self._node_api().storage(storage).content(filepath).delete()
                except Exception as e:
                    _logger.warning(f"Failed to delete snippet {filepath}: {e}")
            except Exception as e:
                _logger.debug(f"Could not clean up snippets for VM {vmid}: {e}")
            _logger.info(f"VM {vmid} destroyed successfully")
@@ -537,6 +830,24 @@ class ProxmoxVMProvisionMixin:
            _logger.exception(f"Failed to destroy VM {vmid}: {e}")
            raise
    def _cicustom_snippets(self, vmid: int) -> list[tuple[str, str]]:
        """``(storage, path)`` of every snippet the VM's cicustom names.
        cicustom reads "user=local:snippets/101-user-data.yaml,network=...".
        Best-effort: a VM whose config cannot be read has none to clean up.
        """
        try:
            cicustom = self._node_api().qemu(vmid).config.get().get("cicustom", "")
        except Exception as e:
            _logger.debug(f"Could not read the snippets of VM {vmid}: {e}")
            return []
        snippets = []
        for entry in cicustom.split(","):
            storage, _, path = entry.partition("=")[2].partition(":")
            if path.startswith("snippets/"):
                snippets.append((storage, path))
        return snippets
    def get_vm_status(
        self,
        vmid: str,
@@ -594,17 +905,18 @@ class ProxmoxVMProvisionMixin:
                    interfaces = (agent_info or {}).get("result", [])
                    # The guest agent does not report interfaces in a fixed order —
-                    # "lo" commonly comes first. Skip it and take the first real
+                    # loopback commonly comes first, named "lo" on Linux and
-                    # NIC that has an IPv4 address.
+                    # "lo0" on the BSDs. Skip loopback addresses, whatever the
                    # interface is called, and take the first IPv4 address.
                    for iface in interfaces:
                        name = iface.get("name", "")
-                        if not name or name == "lo":
+                        if not name:
                            continue
                        for addr in iface.get("ip-addresses", []):
                            if addr.get("ip-address-type") != "ipv4":
                                continue
                            ip_addr = addr.get("ip-address", "")
-                            if ip_addr:
+                            if ip_addr and not ipaddress.ip_address(ip_addr).is_loopback:
                                _logger.info(f"VM {vmid} acquired IP {ip_addr}")
                                return {
                                    "status": "running",
@@ -627,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")
@@ -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>=2.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 = [
+dev = ["pytest>=8", "pytest-cov>=5", "mypy>=1.11", "ruff>=0.6"]
    "pytest>=7.4",
    "pytest-cov>=4.1",
    "pytest-mock>=3.12",
    "responses>=0.24",
    "ruff>=0.1",
    "mypy>=1.7",
 ]
 [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]
+[tool.coverage.run]
-python_version = "3.9"
+source = ["napalm_proxmox"]
-warn_return_any = true
+
-warn_unused_configs = true
+[tool.coverage.report]
 fail_under = 68
@@ -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,
@@ -138,6 +139,7 @@ def _build_mock_api(
    pve_users=None,
    exec_return="",
    sensors=None,
    cluster_status=None,
 ):
    """Build a MagicMock ProxmoxAPI with pre-configured return values."""
    api = MagicMock()
@@ -165,6 +167,7 @@ def _build_mock_api(
    # SDN
    cluster = MagicMock()
    api.cluster = cluster
    cluster.status.get.return_value = cluster_status or []
    cluster.sdn.zones.get.return_value = sdn_zones or SDN_ZONES
    cluster.sdn.vnets.get.return_value = sdn_vnets or SDN_VNETS
@@ -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"
 ]
 }
@@ -2,12 +2,13 @@
 from __future__ import annotations
-import pytest
+from unittest.mock import patch
 from unittest.mock import MagicMock, 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:
@@ -76,3 +77,74 @@ class TestIsAlive:
    def test_not_alive_when_api_fails(self, driver):
        driver._api.version.get.side_effect = Exception("timeout")
        assert driver.is_alive() == {"is_alive": False}
 # A four-node cluster as GET /cluster/status reports it. The node the API
 # session landed on carries local=1 — here the third one, so taking the first
 # entry of /nodes (the old behaviour) picks the wrong host.
 CLUSTER_NODES = [
    {"type": "node", "name": "pve-01", "ip": "172.22.8.101", "local": 0},
    {"type": "node", "name": "pve-02", "ip": "172.22.8.102", "local": 0},
    {"type": "node", "name": "pve-dual", "ip": "172.22.8.120", "local": 1},
    {"type": "node", "name": "pve-garden", "ip": "172.22.8.5", "local": 0},
 ]
 CLUSTER_STATUS = [{"type": "cluster", "name": "home", "nodes": 4}, *CLUSTER_NODES]
 CLUSTER_NODES_LIST = [{"node": n["name"], "status": "online"} for n in CLUSTER_NODES]
 def _open_with(hostname, **api_kwargs):
    drv = ProxmoxDriver(hostname, "root", "secret")
    mock_api = _build_mock_api(**api_kwargs)
    with patch("napalm_proxmox.driver.ProxmoxAPI", return_value=mock_api):
        drv.open()
    return drv
 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)
        assert drv._node_name == "pve-dual"
    def test_matches_by_ip_without_local_flag(self):
        status = [{**n, "local": 0} for n in CLUSTER_STATUS if n["type"] == "node"]
        drv = _open_with("172.22.8.102", nodes=CLUSTER_NODES_LIST, cluster_status=status)
        assert drv._node_name == "pve-02"
    def test_matches_by_name_without_local_flag(self):
        status = [{**n, "local": 0} for n in CLUSTER_STATUS if n["type"] == "node"]
        drv = _open_with("pve-garden", nodes=CLUSTER_NODES_LIST, cluster_status=status)
        assert drv._node_name == "pve-garden"
    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)
        assert drv._node_name == "pve-01"
    def test_single_node_without_cluster_status(self):
        drv = _open_with("10.0.0.9", nodes=[{"node": "solo", "status": "online"}])
        assert drv._node_name == "solo"
    def test_ambiguous_cluster_refuses_to_guess(self):
        status = [{**n, "local": 0} for n in CLUSTER_STATUS if n["type"] == "node"]
        with pytest.raises(ConnectionException, match="node"):
            _open_with("10.9.9.9", nodes=CLUSTER_NODES_LIST, cluster_status=status)
    def test_api_error_fails_open_instead_of_guessing(self):
        # A TLS failure used to be swallowed here: the IP became the node
        # name, open() succeeded and every later call failed quietly.
        drv = ProxmoxDriver("172.22.8.120", "root", "secret")
        mock_api = _build_mock_api()
        mock_api.cluster.status.get.side_effect = Exception("CERTIFICATE_VERIFY_FAILED")
        mock_api.nodes.get.side_effect = Exception("CERTIFICATE_VERIFY_FAILED")
        with patch("napalm_proxmox.driver.ProxmoxAPI", return_value=mock_api):
            with pytest.raises(ConnectionException, match="CERTIFICATE_VERIFY_FAILED"):
                drv.open()
    def test_explicit_node_skips_lookup(self):
        drv = ProxmoxDriver("172.22.8.120", "root", "secret", optional_args={"node": "pve-dual"})
        mock_api = _build_mock_api()
        with patch("napalm_proxmox.driver.ProxmoxAPI", return_value=mock_api):
            drv.open()
        assert drv._node_name == "pve-dual"
        mock_api.cluster.status.get.assert_not_called()
        mock_api.nodes.get.assert_not_called()
@@ -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)
@@ -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": ""}
@@ -0,0 +1,11 @@
 """Proxmox VE says that "upgrade everything" must be a full upgrade on it.
 A plain ``apt-get upgrade`` can leave a node half updated; Proxmox documents
 ``apt full-upgrade``. netOrk reads ``FULL_UPGRADE`` to choose.
 """
 from napalm_proxmox.driver import ProxmoxDriver
 def test_a_full_upgrade_is_declared():
    assert ProxmoxDriver.FULL_UPGRADE is True
@@ -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
@@ -2,11 +2,7 @@
 from __future__ import annotations
-import pytest
+from tests.conftest import NODE_NETWORK
 from tests.conftest import _build_mock_api, NODE_NETWORK
 from unittest.mock import patch
 from napalm_proxmox.driver import ProxmoxDriver
 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_errors",
-            "tx_octets", "rx_octets", "tx_unicast_packets", "rx_unicast_packets",
+            "rx_errors",
-            "tx_multicast_packets", "rx_multicast_packets",
+            "tx_discards",
-            "tx_broadcast_packets", "rx_broadcast_packets",
+            "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}"
@@ -0,0 +1,46 @@
 """`get_kernel_facts`: what the node's kernel has built and loaded.
 A Proxmox node runs its own kernel under every guest, which makes it the host
 where a kernel CVE's preconditions matter most. The command and its parse are
 napalm-device-types'; the driver only carries the command over its exec path.
 """
 from __future__ import annotations
 import base64
 import gzip
 from unittest.mock import patch
 import pytest
 from napalm_device_types import KernelFactsMixin
 from napalm_device_types.kernel import KERNEL_FACTS_COMMAND
 from napalm_proxmox.driver import ProxmoxDriver
 REPORT = (
    "[release]\n6.8.12-4-pve\n[loaded]\nkvm_intel\n[builtin]\nkernel/net/ipv4/tcp_cubic.ko\n"
    "[available]\nkernel/net/tipc/tipc.ko\n[config]\nCONFIG_TIPC=m\n"
 )
 WIRE = "KFACTS_BEGIN\n" + base64.encodebytes(gzip.compress(REPORT.encode())).decode() + "KFACTS_END"
 def test_the_driver_declares_the_contract():
    assert issubclass(ProxmoxDriver, KernelFactsMixin)
 def test_it_runs_the_shared_command(driver):
    with patch.object(driver, "_exec_ssh_command", return_value=WIRE) as exec_:
        facts = driver.get_kernel_facts()
    exec_.assert_called_once_with(KERNEL_FACTS_COMMAND)
    assert facts["release"] == "6.8.12-4-pve"
    assert facts["loaded"] == ["kvm_intel"]
    assert facts["builtin"] == ["tcp_cubic"]
    assert facts["available"] == ["tipc"]
    assert facts["config"] == {"CONFIG_TIPC": "m"}
 def test_output_without_a_report_raises(driver):
    with patch.object(driver, "_exec_ssh_command", return_value=""):
        with pytest.raises(ValueError):
            driver.get_kernel_facts()
@@ -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")
    )
@@ -0,0 +1,38 @@
 """`get_listening_sockets`: what listens on the node, and which service it is.
 Whether pveproxy, a Ceph manager or a guest-facing service is reachable from
 outside the node is decided by the address it listens on. The command and its
 parse are napalm-device-types'; the driver only carries the command over its
 exec path, which already runs as root.
 """
 from __future__ import annotations
 from unittest.mock import patch
 from napalm_device_types import ListeningSocketsMixin
 from napalm_proxmox.driver import ProxmoxDriver
 WIRE = (
    "SOCK_BEGIN\n[ss]\n"
    'tcp LISTEN 0 4096 *:8006 *:* users:(("pveproxy worker",pid=2101,fd=6))\n'
    "__SS_RC=0\n[cgroups]\n"
    "2101 0::/system.slice/pveproxy.service\n"
    "SOCK_END\n"
 )
 def test_the_driver_declares_the_contract():
    assert issubclass(ProxmoxDriver, ListeningSocketsMixin)
 def test_it_reads_as_root_over_the_exec_path(driver):
    with patch.object(driver, "_exec_ssh_command", return_value=WIRE) as exec_:
        reading = driver.get_listening_sockets()
    [command] = [c.args[0] for c in exec_.call_args_list]
    assert command.startswith("sh -c '")
    assert reading["attributed"] is True
    [socket] = reading["sockets"]
    assert (socket["address"], socket["port"], socket["unit"]) == ("*", 8006, "pveproxy")
@@ -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()
@@ -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 = (
+    OVS_APPCTL = " LOCAL     0  aa:bb:cc:dd:ee:03    0\n     1    10  11:22:33:44:55:66  100\n"
        " 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):
@@ -2,8 +2,7 @@
 from __future__ import annotations
-import pytest
+from tests.conftest import _build_mock_api
 from tests.conftest import _build_mock_api, SDN_ZONES, SDN_VNETS
 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})
@@ -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()
@@ -0,0 +1,68 @@
 """Services on a Proxmox node: systemd, through napalm-device-types' mixin.
 The listing used to read ``list-unit-files``' second column, which since
 systemd 245 is followed by a preset column -- so ``enabled`` was false for
 every service on every node (#6). An action ended in ``|| true`` and reported
 success whatever happened. Both now come from the shared mixin; the driver
 only carries the command over its exec path, as root.
 """
 from __future__ import annotations
 from unittest.mock import patch
 import pytest
 from napalm_device_types import SystemdServicesMixin
 from napalm_device_types.systemd import SYSTEMD_SERVICES_COMMAND, service_action_command
 from napalm_proxmox.driver import ProxmoxDriver
 REPORT = (
    "SVC_BEGIN\n[files]\npveproxy.service enabled enabled\n[units]\n"
    "MainPID=1234\nId=pveproxy.service\nNames=pveproxy.service\nLoadState=loaded\n"
    "ActiveState=active\nSubState=running\nUnitFileState=enabled\n"
    "[generated]\nSVC_END\n"
 )
 def test_the_driver_uses_the_shared_mixin():
    assert issubclass(ProxmoxDriver, SystemdServicesMixin)
    assert ProxmoxDriver.get_services is SystemdServicesMixin.get_services
    assert ProxmoxDriver.manage_service is SystemdServicesMixin.manage_service
 def test_listing_runs_the_shared_command(driver):
    with patch.object(driver, "_exec_ssh_command", return_value=REPORT) as exec_:
        services = driver.get_services()
    exec_.assert_called_once_with(SYSTEMD_SERVICES_COMMAND)
    assert services == [{"name": "pveproxy", "running": True, "enabled": True, "pid": 1234}]
 def test_an_unreadable_listing_raises_instead_of_reporting_no_services(driver):
    with patch.object(driver, "_exec_ssh_command", return_value=""):
        with pytest.raises(ValueError):
            driver.get_services()
 def test_an_action_reports_its_real_outcome(driver):
    failed = "Failed to restart nope.service: Unit nope.service not found.\n__SVC_RC=5"
    with patch.object(driver, "_exec_ssh_command", return_value=failed) as exec_:
        result = driver.manage_service("nope", "restart")
    exec_.assert_called_once_with(service_action_command("nope", "restart"))
    assert result["success"] is False
    assert "not found" in result["output"]
 def test_a_successful_action(driver):
    with patch.object(driver, "_exec_ssh_command", return_value="__SVC_RC=0"):
        assert driver.manage_service("pveproxy", "restart") == {"success": True, "output": ""}
 def test_an_invalid_name_never_reaches_the_node(driver):
    with patch.object(driver, "_exec_ssh_command") as exec_:
        with pytest.raises(ValueError):
            driver.manage_service("pveproxy; reboot", "stop")
    exec_.assert_not_called()
@@ -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
@@ -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",
    }
@@ -0,0 +1,120 @@
 """Pending updates on a Proxmox node: from where, whether they are security fixes,
 and whether the node needs a reboot.
 The node's APT API names only an Origin ("Debian", "Proxmox"), the same for the
 main and the security archive. ``apt list --upgradable`` over the exec path
 names the suite (``trixie-security``), so that is read first; the API remains
 the fallback, with the security status left unknown. A reader that cannot read
 raises: an empty list would tell netOrk that nothing is pending.
 """
 from __future__ import annotations
 from unittest.mock import MagicMock, patch
 import pytest
 from napalm_device_types import HostStatusMixin
 from napalm_device_types.host_status import HOST_STATUS_COMMAND
 from napalm_device_types.package_updates import APT_UPGRADABLE_COMMAND
 from napalm_proxmox.driver import ProxmoxDriver
 APT = (
    "libssl3t64/stable-security 3.5.1-1+deb13u2 amd64 [upgradable from: 3.5.1-1+deb13u1]\n"
    "ceph-common/stable 20.2.4-pve5 amd64 [upgradable from: 20.2.4-pve4]\n"
 )
 API_ENTRY = {
    "Package": "librados2",
    "OldVersion": "20.2.4-pve4",
    "Version": "20.2.4-pve5",
    "Origin": "Proxmox",
 }
 def _api_updates(driver, entries=None, error=None):
    api = MagicMock()
    getter = api.nodes.return_value.apt.update.get
    if error:
        getter.side_effect = error
    else:
        getter.return_value = entries or []
    driver._api = api
    return api
 class TestAvailableUpdates:
    def test_apt_over_the_exec_path_names_the_suite(self, driver):
        with patch.object(driver, "_exec_ssh_command", return_value=APT + "__APT_RC=0\n") as exec_:
            updates = {u["name"]: u for u in driver.get_available_updates()}
        exec_.assert_called_once_with(APT_UPGRADABLE_COMMAND)
        assert updates["libssl3t64"]["security"] is True
        assert updates["ceph-common"]["security"] is False
        assert updates["ceph-common"]["origin"] == "stable"
    def test_the_api_is_the_fallback_with_security_unknown(self, driver):
        _api_updates(driver, [API_ENTRY])
        with patch.object(driver, "_exec_ssh_command", return_value=""):
            updates = driver.get_available_updates()
        assert updates == [
            {
                "name": "librados2",
                "current_version": "20.2.4-pve4",
                "new_version": "20.2.4-pve5",
                "origin": "Proxmox",
                "security": None,
            }
        ]
    def test_a_failed_apt_falls_back_too(self, driver):
        _api_updates(driver, [API_ENTRY])
        with patch.object(driver, "_exec_ssh_command", return_value="E: lock\n__APT_RC=100\n"):
            assert [u["name"] for u in driver.get_available_updates()] == ["librados2"]
    def test_nothing_readable_raises_instead_of_reporting_nothing(self, driver):
        _api_updates(driver, error=RuntimeError("API timeout"))
        with patch.object(driver, "_exec_ssh_command", return_value=""):
            with pytest.raises(RuntimeError):
                driver.get_available_updates()
    def test_nothing_pending_is_an_empty_list(self, driver):
        with patch.object(driver, "_exec_ssh_command", return_value="__APT_RC=0\n"):
            assert driver.get_available_updates() == []
 class TestRefresh:
    def test_the_node_refreshes_its_index_through_the_api(self, driver):
        api = _api_updates(driver)
        api.nodes.return_value.apt.update.post.return_value = "UPID:pve1:0001"
        result = driver.refresh_available_updates()
        assert result["success"] is True
        api.nodes.return_value.apt.update.post.assert_called_once()
    def test_a_refused_refresh_says_why(self, driver):
        api = _api_updates(driver)
        api.nodes.return_value.apt.update.post.side_effect = RuntimeError(
            "403 Permission check failed"
        )
        result = driver.refresh_available_updates()
        assert result == {"success": False, "output": "403 Permission check failed"}
 class TestHostStatus:
    def test_the_node_is_read_over_the_exec_path(self, driver):
        report = (
            "HSTAT_BEGIN\n[kernel]\n7.0.14-19-pve\n[modules]\n7.0.14-19-pve\n7.0.2-6-pve\n"
            "[timers]\napt-daily-upgrade.timer enabled\nHSTAT_END\n"
        )
        with patch.object(driver, "_exec_ssh_command", return_value=report) as exec_:
            status = driver.get_host_status()
        exec_.assert_called_once_with(HOST_STATUS_COMMAND)
        assert status["reboot_required"] is False
    def test_the_driver_declares_the_contract(self):
        assert issubclass(ProxmoxDriver, HostStatusMixin)
@@ -2,7 +2,6 @@
 from __future__ import annotations
 import pytest
 from napalm_proxmox import utils
@@ -0,0 +1,153 @@
 """Which virtual CPU model a new VM gets.
 Created without a ``cpu`` argument, a Proxmox VM falls back to ``kvm64``:
 no AVX, no AES-NI. MongoDB 5.0 and later will not even start on it, which is
 how a Graylog provisioned through netOrk failed (netOrk#494). The driver now
 always names a model, defaulting to ``x86-64-v2-AES`` as the Proxmox GUI does,
 and lists the alternatives with what each needs from the node's CPU.
 The flag sets below are trimmed from real ``/nodes/{node}/status`` answers in a
 mixed cluster: a Celeron J3455 (no AVX at all) and an i7-7700 (AVX2).
 """
 from __future__ import annotations
 from unittest.mock import MagicMock, patch
 import pytest
 from napalm_proxmox.vm_provision_mixin import ProxmoxVMProvisionMixin
 CELERON_J3455 = (
    "fpu vme de pse tsc msr pae mce cx8 apic sep mtrr pge mca cmov pat pse36 clflush "
    "mmx fxsr sse sse2 ss ht syscall nx pdpe1gb rdtscp lm constant_tsc pni pclmulqdq "
    "ssse3 cx16 sse4_1 sse4_2 x2apic movbe popcnt aes rdrand lahf_lm 3dnowprefetch "
    "erms mpx rdseed smap clflushopt sha_ni xsaveopt xsavec xgetbv1"
 )
 CORE_I7_7700 = (
    "fpu vme de pse tsc msr pae mce cx8 apic sep mtrr pge mca cmov pat pse36 clflush "
    "mmx fxsr sse sse2 ss ht syscall nx pdpe1gb rdtscp lm constant_tsc pni pclmulqdq "
    "ssse3 fma cx16 sse4_1 sse4_2 x2apic movbe popcnt aes xsave avx f16c rdrand "
    "lahf_lm abm 3dnowprefetch fsgsbase bmi1 hle avx2 smep bmi2 erms invpcid rtm mpx "
    "rdseed adx smap clflushopt xsaveopt xsavec xgetbv1 xsaves"
 )
 def _mixin_on(flags: str) -> tuple[ProxmoxVMProvisionMixin, MagicMock, MagicMock]:
    """A mixin whose node reports `flags` and that can run a full create."""
    mixin = ProxmoxVMProvisionMixin()
    mixin._node_name = "pve1"
    api = MagicMock()
    api.cluster.nextid.get.return_value = 120
    api.storage.return_value.get.return_value = {"path": "/var/lib/vz"}
    node = MagicMock()
    node.status.get.return_value = {"cpuinfo": {"model": "test", "flags": flags}}
    node.storage.get.return_value = [
        {"storage": "local-lvm", "type": "lvmthin", "content": "images,rootdir", "enabled": 1},
        {"storage": "snippets", "type": "dir", "content": "snippets", "enabled": 1},
    ]
    vm = MagicMock()
    vm.config.get.return_value = {
        "unused0": "local-lvm:vm-120-disk-0",
        "scsi0": "local-lvm:vm-120-disk-0",
    }
    vm.status.start.post.return_value = "UPID:pve1:1:start"
    node.qemu.return_value = vm
    task = MagicMock()
    task.status.get.return_value = {"status": "stopped", "exitstatus": "OK"}
    node.tasks.return_value = task
    mixin._api = api
    mixin._node_api = MagicMock(return_value=node)
    mixin._download_cloud_image = MagicMock(return_value="/var/lib/vz/template/x.qcow2")
    mixin._run_node_command = MagicMock(return_value="")
    return mixin, api, node
 def _create(mixin: ProxmoxVMProvisionMixin, **kwargs):
    with patch("time.sleep"):
        return mixin.create_vm_from_cloud_init(
            name="graylog-01",
            image_url="https://cloud-images.ubuntu.com/noble/current/noble-server-cloudimg-amd64.img",
            cpu=2,
            memory=4096,
            nics=[{"bridge": "vmbr0"}],
            cloud_init_config={"hostname": "graylog-01"},
            **kwargs,
        )
 def _by_name(types):
    return {t["name"]: t for t in types}
 class TestListing:
    def test_offers_the_three_models_in_order(self):
        mixin, _, _ = _mixin_on(CORE_I7_7700)
        assert [t["name"] for t in mixin.get_vm_cpu_types()] == [
            "x86-64-v2-AES",
            "x86-64-v3",
            "host",
        ]
    def test_exactly_one_default_and_it_is_what_the_gui_uses(self):
        mixin, _, _ = _mixin_on(CORE_I7_7700)
        defaults = [t["name"] for t in mixin.get_vm_cpu_types() if t["default"]]
        assert defaults == ["x86-64-v2-AES"]
    def test_v3_gives_avx_and_runs_on_a_core_i7(self):
        mixin, _, _ = _mixin_on(CORE_I7_7700)
        v3 = _by_name(mixin.get_vm_cpu_types())["x86-64-v3"]
        assert v3["available"] is True
        assert {"avx", "avx2"} <= set(v3["features"])
    def test_v3_is_unavailable_on_a_celeron_without_avx(self):
        mixin, _, _ = _mixin_on(CELERON_J3455)
        types = _by_name(mixin.get_vm_cpu_types())
        assert types["x86-64-v3"]["available"] is False
        assert types["x86-64-v2-AES"]["available"] is True
    def test_v2_aes_has_no_avx(self):
        mixin, _, _ = _mixin_on(CORE_I7_7700)
        assert "avx" not in _by_name(mixin.get_vm_cpu_types())["x86-64-v2-AES"]["features"]
    def test_host_passes_the_nodes_own_flags_through(self):
        """On a CPU without AVX, `host` gives none either -- a role that needs
        AVX must not be told `host` would help there."""
        mixin, _, _ = _mixin_on(CELERON_J3455)
        host = _by_name(mixin.get_vm_cpu_types())["host"]
        assert host["available"] is True
        assert "avx" not in host["features"]
        assert "aes" in host["features"]
    def test_every_entry_explains_itself(self):
        mixin, _, _ = _mixin_on(CORE_I7_7700)
        assert all(t["description"] for t in mixin.get_vm_cpu_types())
 class TestCreate:
    def test_names_the_default_model_instead_of_leaving_kvm64(self):
        mixin, _, node = _mixin_on(CORE_I7_7700)
        _create(mixin)
        assert node.qemu.post.call_args[1]["cpu"] == "x86-64-v2-AES"
    def test_passes_a_chosen_model_through(self):
        mixin, _, node = _mixin_on(CORE_I7_7700)
        _create(mixin, cpu_type="host")
        assert node.qemu.post.call_args[1]["cpu"] == "host"
    def test_refuses_a_model_the_node_cannot_run_before_creating_anything(self):
        mixin, api, node = _mixin_on(CELERON_J3455)
        with pytest.raises(ValueError, match="avx"):
            _create(mixin, cpu_type="x86-64-v3")
        api.cluster.nextid.get.assert_not_called()
        node.qemu.post.assert_not_called()
    def test_refuses_an_unknown_model_before_creating_anything(self):
        mixin, api, node = _mixin_on(CORE_I7_7700)
        with pytest.raises(ValueError, match="kvm64"):
            _create(mixin, cpu_type="kvm64")
        api.cluster.nextid.get.assert_not_called()
        node.qemu.post.assert_not_called()
@@ -0,0 +1,276 @@
 """Provisioning guests other than Linux: FreeBSD and OpenBSD (NetOrk/netork#794).
 What each guest needs was found on FreeBSD 15.1 and OpenBSD 7.9 cloud images
 (NetOrk/netork#793).
 """
 from __future__ import annotations
 import base64
 from unittest.mock import MagicMock, patch
 import pytest
 import yaml
 from napalm_device_types.provisioning import QEMU_GUEST_AGENTS
 from napalm_proxmox.driver import ProxmoxDriver
 from napalm_proxmox.vm_provision_mixin import ProxmoxVMProvisionMixin
 _FREEBSD_URL = (
    "https://download.freebsd.org/releases/VM-IMAGES/15.1-RELEASE/amd64/Latest/"
    "FreeBSD-15.1-RELEASE-amd64-BASIC-CLOUDINIT-ufs.qcow2.xz"
 )
 _DEBIAN_URL = (
    "https://cloud.debian.org/images/cloud/trixie/latest/debian-13-genericcloud-amd64.qcow2"
 )
 def _mixin(net0: str = "virtio=BC:24:11:AA:BB:02,bridge=vmbr0") -> tuple:
    """A mixin whose node answers like Proxmox; the VM config carries *net0*."""
    mixin = ProxmoxVMProvisionMixin()
    mixin._node_name = "pve1"
    api = MagicMock()
    api.cluster.nextid.get.return_value = 101
    api.storage.return_value.get.return_value = {"path": "/var/lib/vz"}
    node = MagicMock()
    node.storage.get.return_value = [
        {"storage": "local-lvm", "type": "lvmthin", "content": "images,rootdir", "enabled": 1},
        {"storage": "local", "type": "dir", "content": "snippets,iso", "enabled": 1},
    ]
    vm = MagicMock()
    node.qemu.return_value = vm
    vm.config.get.return_value = {"unused0": "local-lvm:vm-101-disk-0", "net0": net0}
    vm.status.start.post.return_value = "UPID:pve1:start"
    node.tasks.return_value.status.get.return_value = {"status": "stopped", "exitstatus": "OK"}
    mixin._api = api
    mixin._node_api = MagicMock(return_value=node)
    mixin._download_cloud_image = MagicMock(
        return_value="/var/lib/vz/template/netork-images/x.qcow2"
    )
    mixin._run_node_command = MagicMock(return_value="")
    return mixin, node, vm
 def _create(mixin, guest_os: str, image_url: str = _DEBIAN_URL, **kwargs):
    with patch("time.sleep"):
        return mixin.create_vm_from_cloud_init(
            name="bsd-vm",
            image_url=image_url,
            cpu=2,
            memory=2048,
            nics=[{"bridge": "vmbr0"}],
            cloud_init_config={"hostname": "bsd-vm"},
            guest_os=guest_os,
            **kwargs,
        )
 def _all_config_posts(vm) -> dict:
    merged: dict = {}
    for call in vm.config.post.call_args_list:
        merged.update(call.kwargs)
    return merged
 def _written_snippet(mixin, name: str) -> str | None:
    """The content of the snippet *name* the driver wrote over SSH, if any."""
    for call in mixin._run_node_command.call_args_list:
        command = call.args[0]
        if f"snippets/{name}" in command and "base64 -d" in command:
            encoded = command.split("echo ", 1)[1].split(" |", 1)[0]
            return base64.b64decode(encoded).decode()
    return None
 class TestWhichGuestsProxmoxProvisions:
    def test_all_qemu_guests(self):
        assert ProxmoxDriver.GUEST_AGENTS == QEMU_GUEST_AGENTS
    def test_an_unknown_guest_is_refused_before_anything_is_created(self):
        mixin, node, _ = _mixin()
        with pytest.raises(ValueError, match="windows"):
            _create(mixin, "windows")
        mixin._api.cluster.nextid.get.assert_not_called()
        node.qemu.post.assert_not_called()
 class TestVmShellPerGuest:
    @pytest.mark.parametrize(
        ("guest_os", "ostype"), [("linux", "l26"), ("freebsd", "other"), ("openbsd", "other")]
    )
    def test_ostype(self, guest_os, ostype):
        mixin, node, _ = _mixin()
        _create(mixin, guest_os)
        assert node.qemu.post.call_args.kwargs["ostype"] == ostype
    @pytest.mark.parametrize("guest_os", ["linux", "freebsd"])
    def test_agent_over_virtio_serial(self, guest_os):
        mixin, node, _ = _mixin()
        _create(mixin, guest_os)
        shell = node.qemu.post.call_args.kwargs
        assert shell["agent"] == "1"
        assert "serial0" not in shell
    def test_openbsd_agent_on_the_second_isa_serial_port(self):
        """OpenBSD's qemu-ga cannot use virtio-serial, and the image keeps its
        console on com0: the agent only answers as cua01, behind serial0."""
        mixin, node, _ = _mixin()
        _create(mixin, "openbsd")
        shell = node.qemu.post.call_args.kwargs
        assert shell["agent"] == "1,type=isa"
        assert shell["serial0"] == "socket"
 class TestNetworkConfigPerGuest:
    def test_linux_keeps_proxmoxs_own_network_config(self):
        mixin, _, vm = _mixin()
        _create(mixin, "linux")
        config = _all_config_posts(vm)
        assert config["cicustom"] == "user=local:snippets/101-user-data.yaml"
        assert config["ipconfig0"] == "ip=dhcp"
        assert _written_snippet(mixin, "101-network-config.yaml") is None
    @pytest.mark.parametrize("guest_os", ["freebsd", "openbsd"])
    def test_a_bsd_guest_gets_a_v2_network_config_matched_by_its_mac(self, guest_os):
        """FreeBSD's nuageinit fails on the v1 Proxmox writes and then skips
        runcmd, which is what starts the guest agent."""
        mixin, _, vm = _mixin(net0="virtio=BC:24:11:AA:BB:02,bridge=vmbr0")
        _create(mixin, guest_os)
        config = _all_config_posts(vm)
        assert config["cicustom"] == (
            "user=local:snippets/101-user-data.yaml,network=local:snippets/101-network-config.yaml"
        )
        assert yaml.safe_load(_written_snippet(mixin, "101-network-config.yaml")) == {
            "version": 2,
            "ethernets": {"nic0": {"match": {"macaddress": "bc:24:11:aa:bb:02"}, "dhcp4": True}},
        }
 class TestPackedImages:
    """Proxmox's download-url unpacks ISOs only, so a packed image goes over SSH."""
    def test_an_xz_image_is_unpacked_on_the_node_and_the_unpacked_file_imported(self):
        mixin, _, _ = _mixin()
        mixin._run_node_command = MagicMock(
            side_effect=lambda cmd, timeout: "MISSING" if "test -f" in cmd else ""
        )
        mixin._download_cloud_image = MagicMock(
            return_value="/var/lib/vz/template/netork-images/k-FreeBSD-15.1-ufs.qcow2.xz"
        )
        _create(mixin, "freebsd", image_url=_FREEBSD_URL, image_checksum="sha256:abc")
        mixin._download_cloud_image.assert_called_once()
        assert mixin._download_cloud_image.call_args.args[1] == "sha256:abc"  # the .xz is verified
        commands = [c.args[0] for c in mixin._run_node_command.call_args_list]
        unpack = next(c for c in commands if c.startswith("xz -dc "))
        assert "k-FreeBSD-15.1-ufs.qcow2.xz" in unpack
        importdisk = next(c for c in commands if c.startswith("qm importdisk"))
        assert ".qcow2.xz" not in importdisk
        assert ".qcow2 " in importdisk
    def test_an_unpacked_image_already_on_the_node_is_not_downloaded_again(self):
        mixin, _, _ = _mixin()
        mixin._run_node_command = MagicMock(
            side_effect=lambda cmd, timeout: "EXISTS" if "test -f" in cmd else ""
        )
        _create(mixin, "freebsd", image_url=_FREEBSD_URL)
        mixin._download_cloud_image.assert_not_called()
        commands = [c.args[0] for c in mixin._run_node_command.call_args_list]
        assert not any(c.startswith("xz -dc ") for c in commands)
    def test_a_packed_image_never_goes_through_an_import_storage(self):
        mixin, node, _ = _mixin()
        node.storage.get.return_value = [
            {"storage": "local-lvm", "content": "images,rootdir", "enabled": 1},
            {"storage": "local", "content": "snippets,import", "enabled": 1},
        ]
        mixin._run_node_command = MagicMock(
            side_effect=lambda cmd, timeout: "MISSING" if "test -f" in cmd else ""
        )
        _create(mixin, "freebsd", image_url=_FREEBSD_URL)
        # storage(name)("download-url").post(...) is the import-storage download.
        node.storage.return_value.return_value.post.assert_not_called()
        mixin._download_cloud_image.assert_called_once()
 class TestLoopbackIsNeverTheVmsAddress:
    """OpenBSD's agent lists lo0 first (#793); Linux names it lo."""
    def _status(self, interfaces: list) -> dict:
        mixin = ProxmoxVMProvisionMixin()
        node = MagicMock()
        mixin._node_api = MagicMock(return_value=node)
        node.qemu.return_value.config.get.return_value = {"net0": "virtio,bridge=vmbr0"}
        node.qemu.return_value.agent.return_value.get.return_value = {"result": interfaces}
        with patch("time.sleep"):
            return mixin.get_vm_status("101", wait_for_ip=True, timeout=30, poll_interval=1)
    def test_openbsd_answer(self):
        """The raw answer of OpenBSD 7.9's qemu-ga, from the spike."""
        result = self._status(
            [
                {
                    "name": "lo0",
                    "ip-addresses": [
                        {"ip-address-type": "ipv6", "ip-address": "::1", "prefix": 128},
                        {"ip-address-type": "ipv6", "ip-address": "fe80:3::1", "prefix": 64},
                        {"ip-address-type": "ipv4", "ip-address": "127.0.0.1", "prefix": 8},
                    ],
                    "hardware-address": "00:00:00:00:00:00",
                },
                {
                    "name": "vio0",
                    "ip-addresses": [
                        {"ip-address-type": "ipv4", "ip-address": "10.0.2.15", "prefix": 24}
                    ],
                    "hardware-address": "bc:24:11:aa:bb:04",
                },
                {"name": "enc0", "hardware-address": "00:00:00:00:00:00"},
                {"name": "pflog0", "hardware-address": "00:00:00:00:00:00"},
            ]
        )
        assert result["ip_address"] == "10.0.2.15"
        assert result["mac_address"] == "bc:24:11:aa:bb:04"
    def test_any_loopback_address_is_skipped_whatever_the_interface_is_called(self):
        result = self._status(
            [
                {
                    "name": "lo1",
                    "ip-addresses": [{"ip-address-type": "ipv4", "ip-address": "127.0.0.2"}],
                },
                {
                    "name": "vtnet0",
                    "ip-addresses": [{"ip-address-type": "ipv4", "ip-address": "10.0.0.5"}],
                },
            ]
        )
        assert result["ip_address"] == "10.0.0.5"
 class TestDestroyRemovesEverySnippet:
    def test_user_and_network_snippets_are_read_before_the_vm_is_deleted(self):
        """After the delete the config is gone, and the snippets stayed behind
        (napalm-proxmox#15)."""
        mixin = ProxmoxVMProvisionMixin()
        node = MagicMock()
        mixin._node_api = MagicMock(return_value=node)
        vm = node.qemu.return_value
        order: list[str] = []
        vm.config.get.side_effect = lambda: (
            order.append("config")
            or {
                "cicustom": "user=local:snippets/101-user-data.yaml,"
                "network=local:snippets/101-network-config.yaml"
            }
        )
        vm.delete.side_effect = lambda **kw: order.append("delete")
        node.tasks.return_value.status.get.return_value = {"status": "stopped", "exitstatus": "OK"}
        with patch("time.sleep"):
            mixin.destroy_vm("101")
        assert order.index("config") < order.index("delete")
        deleted = [c.args[0] for c in node.storage.return_value.content.call_args_list]
        assert deleted == ["snippets/101-user-data.yaml", "snippets/101-network-config.yaml"]
@@ -0,0 +1,241 @@
 """Tests for provisioning through a storage with content type "import".
 Proxmox (8.2+) can download a disk image into such a storage itself
 (``download-url``, with checksum verification) and attach it to a VM with
 ``import-from``. When the node has one, provisioning uses it instead of
 downloading over SSH into the node's root filesystem.
 """
 from __future__ import annotations
 from unittest.mock import MagicMock, patch
 from napalm_proxmox.vm_provision_mixin import (
    ProxmoxVMProvisionMixin,
    _import_volume_name,
 )
 _UBUNTU = (
    "https://cloud-images.ubuntu.com/releases/26.04/release/ubuntu-26.04-server-cloudimg-amd64.img"
 )
 _DEBIAN = "https://cloud.debian.org/images/cloud/bookworm/latest/debian-12-genericcloud-amd64.qcow2"
 def _mixin_with_storages(storages: list[dict]) -> tuple[ProxmoxVMProvisionMixin, MagicMock]:
    mixin = ProxmoxVMProvisionMixin()
    mixin._node_name = "pve1"
    node = MagicMock()
    node.storage.get.return_value = storages
    mixin._node_api = MagicMock(return_value=node)
    return mixin, node
 # ── which storage ─────────────────────────────────────────────────────────────
 def test_find_import_storage_picks_a_storage_that_accepts_import():
    mixin, _ = _mixin_with_storages(
        [
            {"storage": "local", "content": "iso,backup,vztmpl,snippets"},
            {"storage": "software", "content": "import,iso", "active": 1},
        ]
    )
    assert mixin._find_import_storage() == "software"
 def test_find_import_storage_does_not_mistake_images_for_import():
    mixin, _ = _mixin_with_storages([{"storage": "local-zfs", "content": "images,rootdir"}])
    assert mixin._find_import_storage() is None
 def test_find_import_storage_skips_disabled_and_inactive_storages():
    """An inactive storage is typically an unmounted share; Proxmox cannot
    download into it, and the SSH path still works without it."""
    mixin, _ = _mixin_with_storages(
        [
            {"storage": "off", "content": "import", "enabled": 0},
            {"storage": "unmounted", "content": "import", "active": 0},
        ]
    )
    assert mixin._find_import_storage() is None
 # ── what the file is called ───────────────────────────────────────────────────
 def test_import_volume_name_keeps_a_qcow2_extension():
    name = _import_volume_name(_DEBIAN)
    assert name is not None
    assert name.endswith("-debian-12-genericcloud-amd64.qcow2")
 def test_import_volume_name_stores_an_img_as_qcow2():
    """Proxmox reads the format off the extension and does not accept .img.
    Ubuntu's .img is qcow2; guessing qcow2 for a raw .img fails loudly at
    import, while the opposite guess would import a qcow2 container as a raw
    disk without complaint."""
    name = _import_volume_name(_UBUNTU)
    assert name is not None
    assert name.endswith("-ubuntu-26.04-server-cloudimg-amd64.qcow2")
 def test_import_volume_name_is_none_for_types_proxmox_cannot_import():
    assert _import_volume_name("https://example.org/image.qcow2.xz") is None
    assert _import_volume_name("https://example.org/installer.iso") is None
    assert _import_volume_name("https://example.org/noextension") is None
 def test_import_volume_name_differs_for_the_same_basename_under_another_url():
    """Ubuntu publishes daily builds under one basename; a cache keyed on the
    basename alone would serve yesterday's build."""
    a = _import_volume_name("https://x/release-20260927/ubuntu-26.04-server-cloudimg-amd64.img")
    b = _import_volume_name("https://x/release-20260928/ubuntu-26.04-server-cloudimg-amd64.img")
    assert a != b
 def test_import_volume_name_uses_only_characters_proxmox_keeps():
    name = _import_volume_name("https://x/My Image (beta)+1.qcow2")
    assert name is not None
    assert all(c.isalnum() or c in "-.+=_" for c in name)
 # ── the download ──────────────────────────────────────────────────────────────
 def test_import_cloud_image_reuses_a_file_already_in_the_storage():
    mixin, node = _mixin_with_storages([])
    name = _import_volume_name(_DEBIAN)
    node.storage.return_value.content.get.return_value = [
        {"volid": f"software:import/{name}", "content": "import"}
    ]
    volid = mixin._import_cloud_image("software", name, _DEBIAN, None, timeout=300)
    assert volid == f"software:import/{name}"
    node.storage.return_value.return_value.post.assert_not_called()
 def test_import_cloud_image_has_proxmox_download_and_verify_it():
    mixin, node = _mixin_with_storages([])
    name = _import_volume_name(_UBUNTU)
    node.storage.return_value.content.get.return_value = []
    download = node.storage.return_value.return_value
    download.post.return_value = "UPID:pve1:download"
    mixin._wait_for_task = MagicMock()
    volid = mixin._import_cloud_image("software", name, _UBUNTU, "sha256:abc123", timeout=300)
    assert volid == f"software:import/{name}"
    node.storage.assert_any_call("software")
    node.storage.return_value.assert_called_with("download-url")
    download.post.assert_called_once_with(
        url=_UBUNTU,
        content="import",
        filename=name,
        checksum="abc123",
        **{"checksum-algorithm": "sha256"},
    )
    mixin._wait_for_task.assert_called_once_with("UPID:pve1:download", timeout=300)
 def test_import_cloud_image_without_checksum_sends_none():
    mixin, node = _mixin_with_storages([])
    name = _import_volume_name(_DEBIAN)
    node.storage.return_value.content.get.return_value = []
    mixin._wait_for_task = MagicMock()
    mixin._import_cloud_image("software", name, _DEBIAN, None, timeout=300)
    kwargs = node.storage.return_value.return_value.post.call_args.kwargs
    assert "checksum" not in kwargs
    assert "checksum-algorithm" not in kwargs
 # ── provisioning end to end ───────────────────────────────────────────────────
 def _provisioning_mixin(storages: list[dict]) -> tuple[ProxmoxVMProvisionMixin, MagicMock]:
    mixin, node = _mixin_with_storages(storages)
    api = MagicMock()
    api.cluster.nextid.get.return_value = 101
    api.storage.return_value.get.return_value = {"path": "/var/lib/vz"}
    mixin._api = api
    mixin._run_node_command = MagicMock(return_value="")
    mixin._download_cloud_image = MagicMock(return_value="/var/lib/vz/template/x.qcow2")
    vm = MagicMock()
    node.qemu.return_value = vm
    vm.config.get.return_value = {"unused0": "local-zfs:vm-101-disk-0"}
    vm.config.post.return_value = None
    node.tasks.return_value.status.get.return_value = {"status": "stopped", "exitstatus": "OK"}
    node.storage.return_value.content.get.return_value = []
    return mixin, node
 def _provision(mixin: ProxmoxVMProvisionMixin, image_url: str) -> None:
    with patch("time.sleep"):
        mixin.create_vm_from_cloud_init(
            name="vm",
            image_url=image_url,
            cpu=1,
            memory=1024,
            nics=[{"bridge": "vmbr0"}],
            cloud_init_config={"hostname": "vm"},
            storage="local-zfs",
            timeout=60,
        )
 _WITH_IMPORT = [
    {"storage": "local", "content": "iso,backup,vztmpl,snippets"},
    {"storage": "software", "content": "import,iso"},
    {"storage": "local-zfs", "content": "images,rootdir"},
 ]
 def test_provisioning_downloads_through_the_import_storage_when_there_is_one():
    mixin, node = _provisioning_mixin(_WITH_IMPORT)
    name = _import_volume_name(_UBUNTU)
    _provision(mixin, _UBUNTU)
    mixin._download_cloud_image.assert_not_called()
    assert not any("importdisk" in c.args[0] for c in mixin._run_node_command.call_args_list), (
        "no SSH download/import when Proxmox can do it"
    )
    disk = next(
        c.kwargs for c in node.qemu.return_value.config.post.call_args_list if "scsi0" in c.kwargs
    )
    assert disk["scsi0"] == f"local-zfs:0,import-from=software:import/{name},discard=on"
    assert disk["boot"] == "order=scsi0"
 def test_provisioning_waits_for_the_import_task():
    """Attaching with import-from copies the image — a task, which has to
    finish before the cloud-init drive and the resize touch the VM."""
    mixin, node = _provisioning_mixin(_WITH_IMPORT)
    node.qemu.return_value.config.post.side_effect = lambda **kw: (
        "UPID:pve1:import" if "scsi0" in kw else None
    )
    mixin._wait_for_task = MagicMock()
    _provision(mixin, _DEBIAN)
    waited = [c.args[0] for c in mixin._wait_for_task.call_args_list]
    assert "UPID:pve1:import" in waited
 def test_provisioning_falls_back_to_ssh_without_an_import_storage():
    mixin, _ = _provisioning_mixin([s for s in _WITH_IMPORT if s["storage"] != "software"])
    _provision(mixin, _UBUNTU)
    mixin._download_cloud_image.assert_called_once()
    assert any("qm importdisk 101" in c.args[0] for c in mixin._run_node_command.call_args_list)
 def test_provisioning_falls_back_to_ssh_for_an_image_type_proxmox_cannot_import():
    mixin, _ = _provisioning_mixin(_WITH_IMPORT)
    _provision(mixin, "https://example.org/image.qcow2.xz")
    mixin._download_cloud_image.assert_called_once()
@@ -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
@@ -459,7 +458,13 @@ def test_destroy_vm_success():
 def test_destroy_vm_already_stopped():
-    """destroy_vm succeeds even if VM already stopped."""
+    """destroy_vm succeeds even if VM already stopped.
    Proxmox refuses to stop a VM that is not running, so the stop call raises.
    (A bare MagicMock as the stop task never reports "stopped": _wait_for_task
    then spun for the full timeout with sleep patched out, and every recorded
    mock call made the test take 60 s and several GB, enough for CI to kill it.)
    """
    mixin = ProxmoxVMProvisionMixin()
    mock_node = MagicMock()
@@ -468,6 +473,7 @@ def test_destroy_vm_already_stopped():
    # Mock VM operations
    mock_vm = MagicMock()
    mock_node.qemu.return_value = mock_vm
    mock_vm.status.stop.post.side_effect = Exception("VM 101 not running")
    mock_vm.config.get.return_value = {}
    mock_vm.delete.return_value = None
@@ -609,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()
+    ssh = MagicMock()
-    mixin._ssh_client.exec_command.return_value = _mock_ssh_exec(0, stdout_text="hello\n")
+    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)
@@ -619,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()
+    ssh = MagicMock()
-    mixin._ssh_client.exec_command.return_value = _mock_ssh_exec(
+    ssh.exec_command.return_value = _mock_ssh_exec(1, stderr_text="No such file or directory")
-        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)