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

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

.gitea/workflows/ci.yml

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

CHANGELOG.md

@@ -7,6 +7,87 @@ 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
+  for the Proxmox task to finish. `power_vm` is unchanged.
+- Snapshot methods `get_vm_snapshots`, `create_vm_snapshot`,
+  `delete_vm_snapshot`, `rollback_vm_snapshot` for VMs and containers
+  (containers never save RAM state).
+- `reboot_host()` restarts the node through the API instead of SSH.
+
+### Changed
+- `get_vms()` reports `vmid` as a string (`"100"`), following
+  napalm-device-types 2.0. Ordering stays numeric.
+
 ## [0.1.0] - 2024-01-01
 
 ### Added

README.md

@@ -12,8 +12,9 @@ It supports all three networking domains of Proxmox VE:
 
 ## Requirements
 
-- Python ≥ 3.9
+- Python ≥ 3.10
 - NAPALM ≥ 5.0.0
+- napalm-device-types ≥ 4.22.0
 - proxmoxer ≥ 2.0.0
 - netaddr ≥ 0.9.0
 - requests ≥ 2.31.0
@@ -27,7 +28,7 @@ pip install napalm-proxmox
 Or directly from source:
 
 ```bash
-git clone https://github.com/example/napalm-proxmox.git
+git clone https://git.netork.io/NAPALM/napalm-proxmox.git
 cd napalm-proxmox
 pip install -e .
 ```

napalm_proxmox/config_mixin.py

@@ -19,7 +19,6 @@ from __future__ import annotations
 import re
 from typing import Any
 
-
 _JsonDict = dict[str, Any]
 
 
@@ -44,13 +43,9 @@ class ProxmoxConfigMixin:
         configs: _JsonDict = {"running": "", "candidate": "", "startup": ""}
 
         if retrieve in ("running", "all", "startup"):
-            raw = self._exec_ssh_command(
+            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

napalm_proxmox/driver.py

@@ -32,26 +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.interfaces_mixin import ProxmoxInterfaceMixin
+from napalm_proxmox.lldp_mixin import ProxmoxLLDPMixin
+from napalm_proxmox.routing_mixin import ProxmoxRoutingMixin
+from napalm_proxmox.sdn_mixin import ProxmoxSDNMixin
+from napalm_proxmox.system_mixin import ProxmoxSystemMixin
+from napalm_proxmox.vm_contract_mixin import ProxmoxVMContractMixin
 from napalm_proxmox.vm_mixin import ProxmoxVMMixin
 from napalm_proxmox.vm_provision_mixin import ProxmoxVMProvisionMixin
-from napalm_proxmox.routing_mixin import ProxmoxRoutingMixin
+from napalm_proxmox.vm_snapshot_mixin import ProxmoxVMSnapshotMixin
-from napalm_proxmox.system_mixin import ProxmoxSystemMixin
+
+logger = logging.getLogger(__name__)
 
 # --------------------------------------------------------------------------- #
 # Type aliases
@@ -69,15 +84,35 @@ class ProxmoxDriver(
     ProxmoxLLDPMixin,
     ProxmoxConfigMixin,
     ProxmoxVMMixin,
+    ProxmoxVMContractMixin,
+    ProxmoxVMSnapshotMixin,
     ProxmoxVMProvisionMixin,
     ProxmoxRoutingMixin,
     ProxmoxSystemMixin,
+    KernelFactsMixin,
+    ServicePackagesMixin,
+    ListeningSocketsMixin,
+    SystemIdentityMixin,
+    SystemdServicesMixin,
+    HostStatusMixin,
+    LicenseMixin,
+    SnmpAgentMixin,
+    EnsurePackageMixin,
+    AptProxyMixin,
     HypervisorDriver,
 ):
     """NAPALM driver for Proxmox VE nodes."""
 
     VENDOR = "Proxmox"
     DRIVER_NAME = "proxmox"
+    # Everything runs over the Proxmox REST API; there is no SSH session.
+    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),
     ]
@@ -107,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")
@@ -120,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 = ""
@@ -137,9 +174,18 @@ class ProxmoxDriver(
             # The openssh backend tunnels all kwargs through to
             # openssh_wrapper.CommandBaseSession, which does not
             # accept password/verify_ssl/token params.
+            # Proxmox authenticates against "<user>@<realm>" and rejects a bare
+            # username outright. A caller who typed a realm keeps it; one who
+            # did not gets self._realm, which is what the documented `realm`
+            # optional_arg is for -- it was read in __init__ and then never
+            # used, so the option had no effect and a bare username failed.
+            user = self.username or ""
+            if user and "@" not in user:
+                user = f"{user}@{self._realm}"
+
             kwargs: _JsonDict = {
                 "host": self.hostname,
-                "user": self.username,
+                "user": user,
                 "password": self.password,
                 "port": self._port,
                 "verify_ssl": self._verify_ssl,
@@ -165,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
@@ -202,18 +258,24 @@ 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:
-        """Return connection liveness."""
+        """Return connection liveness.
+
+        Probes ``GET /version``, the cheapest endpoint that proves the session
+        still authenticates. It used to call ``_resolve_node()``, which returns
+        early without touching the API whenever a node was configured via
+        optional_args — so a dead connection reported itself alive.
+        """
         alive = False
         if self._api:
             try:
-                self._resolve_node()
+                self._api.version.get()
                 alive = True
             except Exception:
-                pass
+                logger.debug("%s: liveness probe failed", self.hostname, exc_info=True)
         return {"is_alive": alive}
 
     # ------------------------------------------------------------------ #
@@ -239,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:
@@ -257,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
     # ------------------------------------------------------------------ #
@@ -360,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", "")
@@ -389,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:
@@ -412,6 +482,14 @@ class ProxmoxDriver(
         except Exception as exc:
             logger.debug("Failed to read DMI info via SSH: %s", exc)
 
+        # Currently-booted kernel release (uname -r) — distinct from any newer
+        # kernel that is merely installed and pending a reboot.
+        running_kernel = ""
+        try:
+            running_kernel = self._exec_ssh_command("uname -r").strip()
+        except Exception as exc:
+            logger.debug("Failed to read running kernel via SSH: %s", exc)
+
         return {
             "uptime": uptime,
             "vendor": vendor or "Proxmox Server Solutions GmbH",
@@ -421,6 +499,7 @@ class ProxmoxDriver(
             "os_version": os_version,
             "serial_number": serial,
             "interface_list": iface_list,
+            "running_kernel": running_kernel,
         }
 
     # ------------------------------------------------------------------ #
@@ -430,5 +509,3 @@ class ProxmoxDriver(
     def cli(self, commands: list[str], encoding: str = "text") -> dict[str, str]:
         """Execute a list of shell commands and return their outputs."""
         return {cmd: self._exec_ssh_command(cmd) for cmd in commands}
-
-

napalm_proxmox/interfaces_mixin.py

@@ -17,6 +17,7 @@
 from __future__ import annotations
 
 import logging
+import re
 from typing import Any
 
 from napalm_proxmox import utils
@@ -113,6 +114,47 @@ class ProxmoxInterfaceMixin:
             }
         return result
 
+    def get_ipv6_neighbors_table(self) -> list[_JsonDict]:
+        """Return the IPv6 neighbour table, read via ``ip -6 neigh show``.
+
+        The IPv6 counterpart to :meth:`get_arp_table`. Proxmox exposes no REST
+        endpoint for it, so it goes through the node exec helper like the ARP
+        table does.
+
+        Entries in FAILED state are dropped: they record an address the kernel
+        could not resolve, so there is no neighbour to report.
+        """
+        raw = self._exec_ssh_command("ip -6 neigh show 2>/dev/null || true")
+        if not raw:
+            return []
+
+        entries: list[_JsonDict] = []
+        for line in raw.splitlines():
+            parts = line.split()
+            # "<ip> dev <iface> lladdr <mac> <STATE>" — an entry without lladdr
+            # never resolved and carries no neighbour.
+            if len(parts) < 6 or "lladdr" not in parts:
+                continue
+            state = parts[-1].upper()
+            if state == "FAILED":
+                continue
+            try:
+                iface = parts[parts.index("dev") + 1]
+                mac = parts[parts.index("lladdr") + 1]
+            except (ValueError, IndexError):
+                continue
+            entries.append(
+                {
+                    "interface": iface,
+                    "mac": utils.normalize_mac(mac),
+                    "ip": parts[0],
+                    # `ip neigh` reports no age; NAPALM's shape requires the key.
+                    "age": 0.0,
+                    "state": state,
+                }
+            )
+        return entries
+
     def get_arp_table(self, vrf: str = "") -> list[_JsonDict]:
         """Return ARP table.
 
@@ -132,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
@@ -160,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:
@@ -198,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:

napalm_proxmox/lldp_mixin.py

@@ -19,7 +19,6 @@ from __future__ import annotations
 import re
 from typing import Any
 
-
 _JsonDict = dict[str, Any]
 
 
@@ -29,9 +28,7 @@ class ProxmoxLLDPMixin:
     def get_lldp_neighbors(self) -> dict[str, list[_JsonDict]]:
         """Return LLDP neighbours (requires lldpd on the Proxmox node)."""
         result: dict[str, list[_JsonDict]] = {}
-        raw = self._exec_ssh_command(
+        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 = {}
 

napalm_proxmox/routing_mixin.py

@@ -21,7 +21,6 @@ from typing import Any
 
 import napalm.base.constants as C
 
-
 _JsonDict = dict[str, Any]
 
 
@@ -92,12 +91,8 @@ class ProxmoxRoutingMixin:
 
         # Individual probe lines
         probes = []
-        for m_probe in re.finditer(
+        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": {

napalm_proxmox/sdn_mixin.py

@@ -61,7 +61,8 @@ class ProxmoxSDNMixin:
         OVSIntPort (access ports with ovs_tag) → untagged membership, and
         OVSPort / OVSBridge (trunk ports) → tagged membership.
 
-        Falls back to ``bridge vlan show`` for classic Linux-bridge nodes.
+        Without OVS ports, VLAN membership is derived from the ``tag=`` values
+        in each VM's and container's ``netN`` config (:meth:`_get_vm_vlan_tags`).
         """
         result: dict[str, _JsonDict] = {}
         node_network = self._get_node_network()
@@ -113,10 +114,12 @@ class ProxmoxSDNMixin:
                 if bridge not in entry["untagged"]:
                     entry["untagged"].append(bridge)
 
-        return {
+        # Deliberately unfiltered. This used to drop entries with no member
-            vid: entry for vid, entry in result.items()
+        # ports, which hid every configured SDN VNet that no VM happened to be
-            if entry.get("tagged") or entry.get("untagged")
+        # attached to — and contradicted the OVS branch above, which returns
-        }
+        # empty-membership VLANs. A VNet exists on the node whether or not
+        # anything currently uses it, and netOrk's VLAN discovery reads this.
+        return result
 
     def _get_vm_vlan_tags(self) -> dict[str, set[str]]:
         """Return ``{vlan_tag: {bridge_names}}`` derived from VM/container net configs.
@@ -150,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 {}
@@ -161,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 {}
@@ -235,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:
@@ -306,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:
@@ -358,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()

napalm_proxmox/system_mixin.py

@@ -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)
     # ------------------------------------------------------------------ #
@@ -244,7 +311,9 @@ class ProxmoxSystemMixin:
 
         # Installed packages via SSH dpkg-query
         raw = self._exec_ssh_command(
-            "dpkg-query -W -f='${Package}\\t${Version}\\t${db:Status-Status}\\t${Installed-Size}\\n'"
+            "dpkg-query -W -f="
+            "'${Package}\\t${Version}\\t${db:Status-Status}\\t${Installed-Size}"
+            "\\t${source:Package}\\t${source:Version}\\n'"
             " 2>/dev/null"
         )
         result: list[_JsonDict] = []
@@ -256,187 +325,132 @@ class ProxmoxSystemMixin:
             version = parts[1] if len(parts) > 1 else ""
             status = parts[2] if len(parts) > 2 else "installed"
             size_kb = parts[3] if len(parts) > 3 else "0"
+            # Debian source package (differs from the binary for split packages,
+            # e.g. openssh-server → openssh). Needed for accurate OSV matching.
+            source_package = parts[4] if len(parts) > 4 and parts[4] else name
+            # And its version, which is a different number from this package's.
+            #
+            # OSV states Debian ranges in *source* versions, so a consumer given
+            # the source package and only the binary version compares two
+            # unrelated numbers: libldb2 is 2:2.11.0+samba4.22.11+dfsg-… while
+            # its source, samba, is 2:4.22.11+dfsg-…. Reporting the source
+            # without its version is worse than reporting neither, because it
+            # looks usable. Every package on all three Proxmox nodes was in that
+            # state — the format string never asked for the field.
+            #
+            # dpkg leaves it empty when it equals `Version`, and so does an
+            # older dpkg that does not know the field at all.
+            source_version = parts[5] if len(parts) > 5 and parts[5] else version
             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,
                     "description": "",
                     "size": size_bytes,
                     "source": "pve",
+                    "source_package": source_package,
+                    "source_version": source_version,
                     "upgrade_version": upgradable.get(name, ""),
-            })
+                }
+            )
         return result
 
     # ------------------------------------------------------------------ #
-    # Device warnings
+    # Licenses (napalm-device-types' LicenseMixin)
     # ------------------------------------------------------------------ #
 
-    def get_device_warnings(self) -> list[_JsonDict]:
+    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",
-                    "severity": "warning",
-                    "action": "install_lldpd",
-                })
-        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",
-                    "severity": "warning",
-                    "title": (
-                        f"{len(updates)} package update"
-                        f"{'s' if len(updates) != 1 else ''} available"
-                    ),
-                    "message": None,
-                    "action": None,
-                    "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({
+            return []
-                    "code": "no_subscription",
+        lic: _JsonDict = {
-                    "severity": "warning",
+            "name": "pve-subscription",
-                    "action": None,
+            "status": _SUBSCRIPTION_STATUS.get(raw.lower(), "unknown"),
-                    "meta": {"status": status},
+            "detail": sub.get("message") or raw,
-                })
-        except Exception as exc:
-            logger.debug("Failed to check subscription status: %s", exc)
-
-        return warnings
-
-    # ------------------------------------------------------------------ #
-    # 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}
 
@@ -466,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()
@@ -484,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.
@@ -646,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()
@@ -663,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",
@@ -679,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 ─────────────────────────────────
@@ -731,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,
         )
 
     # ------------------------------------------------------------------ #
@@ -842,7 +815,8 @@ class ProxmoxSystemMixin:
             "timezone": None,
             "ntp_servers": [],
             "ssh_port": 22,
-            "ssh_password_auth": False,
+            # A flag, not a password.
+            "ssh_password_auth": False,  # nosec B105
         }
 
         # Cluster name and node list from /cluster/status

napalm_proxmox/utils.py

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

napalm_proxmox/vm_contract_mixin.py

@@ -0,0 +1,193 @@
+"""HypervisorDriver contract methods for Proxmox VE: power actions and VM config.
+
+``power_vm`` stays for callers that already use it; these are what a
+hypervisor-neutral caller talks to. They raise instead of returning a
+``{"success": ...}`` dict, and block until Proxmox reports the task finished.
+"""
+
+from __future__ import annotations
+
+import re
+from typing import Any
+
+from napalm_device_types.models import (
+    VMConfigDict,
+    VMDiskDict,
+    VMNICDict,
+    VMPassthroughDict,
+)
+
+_JsonDict = dict[str, Any]
+
+_POWER_TIMEOUT = 120
+_VM_DISK_KEY = re.compile(r"^(scsi|ide|virtio|sata)\d+$|^efidisk\d+$|^tpmstate\d+$")
+_CT_DISK_KEY = re.compile(r"^rootfs$|^mp\d+$")
+_NET_KEY = re.compile(r"^net\d+$")
+_PASSTHROUGH_KEY = re.compile(r"^(hostpci|usb)\d+$")
+_NIC_MODELS = {"virtio", "e1000", "e1000e", "vmxnet3", "rtl8139", "ne2k_pci"}
+_SIZE = re.compile(r"^(\d+(?:\.\d+)?)([KMGT]?)$", re.I)
+_GB_PER_UNIT = {"K": 1 / 1024**2, "M": 1 / 1024, "G": 1, "T": 1024, "": 1}
+
+
+def _options(value: str) -> tuple[str, dict[str, str]]:
+    """Split ``"volume,key=val,..."`` into the leading bare part and its options."""
+    head = ""
+    opts: dict[str, str] = {}
+    for part in str(value).split(","):
+        if "=" in part:
+            k, v = part.split("=", 1)
+            opts[k.strip().lower()] = v.strip()
+        elif not head:
+            head = part.strip()
+    return head, opts
+
+
+def _size_gb(raw: str) -> int:
+    m = _SIZE.match(raw or "")
+    if not m:
+        return 0
+    return int(float(m.group(1)) * _GB_PER_UNIT[m.group(2).upper()])
+
+
+def _boot_order(cfg: _JsonDict) -> list[str]:
+    boot = str(cfg.get("boot", "") or "")
+    if boot.startswith("order="):
+        return [d for d in boot[len("order=") :].split(";") if d]
+    bootdisk = cfg.get("bootdisk")
+    return [bootdisk] if bootdisk else []
+
+
+def _disks(cfg: _JsonDict, vm_type: str, boot_order: list[str]) -> list[VMDiskDict]:
+    key_re = _VM_DISK_KEY if vm_type == "vm" else _CT_DISK_KEY
+    disks: list[VMDiskDict] = []
+    for key in sorted(cfg, key=lambda k: (k != "rootfs", k)):
+        if not key_re.match(key):
+            continue
+        value = str(cfg[key] or "")
+        head, opts = _options(value)
+        if opts.get("media") == "cdrom" or head in ("none", "0", ""):
+            continue
+        disks.append(
+            {
+                "device": key,
+                "storage": head.split(":", 1)[0],
+                "size": _size_gb(opts.get("size", "")),
+                "format": opts.get("format", ""),
+                "bootable": key in boot_order,
+            }
+        )
+    return disks
+
+
+def _nic(key: str, value: str) -> VMNICDict:
+    _, opts = _options(value)
+    model = next((m for m in _NIC_MODELS if m in opts), opts.get("type", ""))
+    mac = opts.get(model, "") if model in _NIC_MODELS else opts.get("hwaddr", "")
+    tag = opts.get("tag", "")
+    return {
+        "device": key,
+        "mac": mac.upper(),
+        "model": model,
+        "bridge": opts.get("bridge", ""),
+        "vlan_id": int(tag) if tag.isdigit() else 0,
+    }
+
+
+def _passthrough(cfg: _JsonDict) -> list[VMPassthroughDict]:
+    return [
+        {"slot": key, "kind": "pci" if key.startswith("hostpci") else "usb", "config": str(val)}
+        for key, val in sorted(cfg.items())
+        if _PASSTHROUGH_KEY.match(key)
+    ]
+
+
+def parse_vm_config(vmid: str, vm_type: str, cfg: _JsonDict) -> VMConfigDict:
+    """Turn a raw ``/qemu/{id}/config`` or ``/lxc/{id}/config`` into a VMConfigDict."""
+    is_vm = vm_type == "vm"
+    cores = int(cfg.get("cores", 1) or 1)
+    sockets = int(cfg.get("sockets", 1) or 1) if is_vm else 1
+    boot_order = _boot_order(cfg)
+    tags = str(cfg.get("tags", "") or "")
+    name_key = "name" if is_vm else "hostname"
+    result: VMConfigDict = {
+        "name": cfg.get(name_key) or f"{'vm' if is_vm else 'ct'}-{vmid}",
+        "vmid": vmid,
+        "vcpus": cores * sockets,
+        "memory": int(cfg.get("memory", 0) or 0),
+        "os_type": cfg.get("ostype", ""),
+        "boot_order": boot_order,
+        "disks": _disks(cfg, vm_type, boot_order),
+        "nics": [_nic(k, str(v)) for k, v in sorted(cfg.items()) if _NET_KEY.match(k)],
+        "description": cfg.get("description", ""),
+        "tags": [t for t in re.split(r"[;,\s]+", tags) if t],
+        "passthrough": _passthrough(cfg),
+    }
+    if is_vm:
+        result["cpu_type"] = str(cfg.get("cpu", "kvm64")).split(",")[0].removeprefix("cputype=")
+        result["sockets"] = sockets
+        result["cores_per_socket"] = cores
+        result["firmware"] = "efi" if cfg.get("bios") == "ovmf" else "bios"
+        if cfg.get("machine"):
+            result["machine"] = cfg["machine"]
+    return result
+
+
+class ProxmoxVMContractMixin:
+    """HypervisorDriver's VM methods on top of the Proxmox node API."""
+
+    def _resolve_vm(self, name: str) -> tuple[int, str]:
+        """Find a guest by vmid or display name; return ``(vmid, "vm"|"container")``."""
+        node = self._node_api()
+        for vm_type, listing in (("vm", node.qemu), ("container", node.lxc)):
+            for guest in listing.get() or []:
+                if str(guest.get("vmid")) == name or guest.get("name") == name:
+                    return int(guest["vmid"]), vm_type
+        raise ValueError(f"No VM or container named or numbered {name!r}")
+
+    def _guest_api(self, vmid: int, vm_type: str) -> Any:
+        node = self._node_api()
+        return node.qemu(vmid) if vm_type == "vm" else node.lxc(vmid)
+
+    def _run_power(self, vmid: int, vm_type: str, action: str) -> None:
+        try:
+            upid = getattr(self._guest_api(vmid, vm_type).status, action).post()
+        except Exception as exc:
+            raise RuntimeError(f"{action} of {vm_type} {vmid} failed: {exc}") from exc
+        if upid:
+            self._wait_for_task(upid, timeout=_POWER_TIMEOUT)
+
+    def start_vm(self, name: str) -> None:
+        self._run_power(*self._resolve_vm(name), "start")
+
+    def stop_vm(self, name: str, force: bool = False) -> None:
+        self._run_power(*self._resolve_vm(name), "stop" if force else "shutdown")
+
+    def reboot_vm(self, name: str, force: bool = False) -> None:
+        vmid, vm_type = self._resolve_vm(name)
+        if not force:
+            self._run_power(vmid, vm_type, "reboot")
+        elif vm_type == "vm":
+            self._run_power(vmid, vm_type, "reset")
+        else:
+            self._run_power(vmid, vm_type, "stop")
+            self._run_power(vmid, vm_type, "start")
+
+    def suspend_vm(self, name: str) -> None:
+        vmid, vm_type = self._resolve_vm(name)
+        if vm_type != "vm":
+            raise RuntimeError(f"Proxmox cannot suspend container {vmid}")
+        self._run_power(vmid, vm_type, "suspend")
+
+    def get_vm_config(self, name: str) -> VMConfigDict:
+        vmid, vm_type = self._resolve_vm(name)
+        cfg = self._guest_api(vmid, vm_type).config.get() or {}
+        return parse_vm_config(str(vmid), vm_type, cfg)
+
+    # -- the node itself --------------------------------------------------------
+
+    def reboot_host(self) -> None:
+        """Restart this Proxmox node through the API (no SSH involved)."""
+        try:
+            self._node_api().status.post(command="reboot")
+        except Exception as exc:
+            raise RuntimeError(f"Reboot of node {self._node_name!r} refused: {exc}") from exc

napalm_proxmox/vm_mixin.py

@@ -20,8 +20,6 @@ import logging
 import re
 from typing import Any
 
-from napalm_proxmox import utils
-
 logger = logging.getLogger(__name__)
 
 _JsonDict = dict[str, Any]
@@ -30,9 +28,7 @@ _JsonDict = dict[str, Any]
 class ProxmoxVMMixin:
     """Mixin providing VM and container NAPALM methods."""
 
-    def get_vm_interfaces(
+    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
@@ -216,7 +212,7 @@ class ProxmoxVMMixin:
         """Return all VMs (QEMU) and containers (LXC) on this node.
 
         Each entry contains:
-        * vmid (int)        - Proxmox VM/container ID
+        * vmid (str)        - Proxmox VM/container ID, e.g. ``"100"``
         * name (str)        - display name
         * type (str)        - ``"vm"`` or ``"container"``
         * status (str)      - ``"running"``, ``"stopped"``, etc.
@@ -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,8 +252,9 @@ class ProxmoxVMMixin:
                 )
 
                 disks, onboot = self._get_vm_disk_and_boot(vmid, "qemu")
-                result.append({
+                result.append(
-                    "vmid": vmid,
+                    {
+                        "vmid": str(vmid),
                         "name": name,
                         "type": "vm",
                         "status": status,
@@ -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,8 +298,9 @@ class ProxmoxVMMixin:
                 )
 
                 disks, onboot = self._get_vm_disk_and_boot(vmid, "lxc")
-                result.append({
+                result.append(
-                    "vmid": vmid,
+                    {
+                        "vmid": str(vmid),
                         "name": name,
                         "type": "container",
                         "status": status,
@@ -317,20 +316,19 @@ 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)
 
-        return sorted(result, key=lambda x: x["vmid"])
+        return sorted(result, key=lambda x: int(x["vmid"]))
 
     # Disk-key prefixes for QEMU: scsi, virtio, ide, sata (exclude cdrom/none entries)
     _DISK_KEYS_VM = re.compile(r"^(scsi|virtio|ide|sata)\d+$")
     # 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:

napalm_proxmox/vm_provision_mixin.py

@@ -2,20 +2,504 @@
 
 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
 
-from napalm_device_types.models import NetworkTargetDict, VMProvisionResultDict, VMStatusDict
+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.
+
+        Unlike ``_exec_ssh_command`` (best-effort, fixed timeout, swallows
+        errors), this is for critical provisioning steps — image download,
+        disk import — where a non-zero exit or a caller-specific timeout must
+        surface as a hard failure rather than an empty string.
+        """
+        _, stdout, stderr = self._node_ssh().exec_command(command, timeout=timeout)
+        exit_status = stdout.channel.recv_exit_status()
+        out = stdout.read().decode().strip()
+        err = stderr.read().decode().strip()
+        if exit_status != 0:
+            raise RuntimeError(f"Command failed (exit {exit_status}): {command}\n{err or out}")
+        return out
+
+    def _download_cloud_image(
+        self, image_url: str, image_checksum: str | None, timeout: int
+    ) -> str:
+        """
+        Download image_url to the node's image cache dir if not already present.
+
+        Returns the local path on the hypervisor node. The cache filename is
+        prefixed with a hash of the *full* URL, not just its basename —
+        Ubuntu (and others) publish per-build URLs that change daily under a
+        stable basename (e.g. .../release-20260713/ubuntu-26.04-server-
+        cloudimg-amd64.img), so keying the cache on the basename alone let a
+        stale previous-day build satisfy the "already cached" check and fail
+        checksum verification against today's expected hash.
+
+        Verifies image_checksum (format "<algo>:<hex>", e.g. "sha256:abc123...")
+        if given. On mismatch, removes the bad file and retries the download
+        once (covers a corrupted/partial transfer or a stale same-keyed file)
+        before raising.
+        """
+        filename = image_url.rstrip("/").rsplit("/", 1)[-1]
+        local_path = f"{_IMAGE_CACHE_DIR}/{_url_key(image_url)}-{filename}"
+
+        algo, expected = _split_checksum(image_checksum or "")
+
+        max_attempts = 2
+        for attempt in range(1, max_attempts + 1):
+            exists = self._run_node_command(
+                f"mkdir -p {_IMAGE_CACHE_DIR} && test -f {local_path} "
+                f"&& echo EXISTS || echo MISSING",
+                timeout=30,
+            )
+            if "EXISTS" not in exists:
+                _logger.info(f"Downloading cloud image {image_url} -> {local_path}")
+                self._run_node_command(
+                    f"wget -q -O {local_path}.tmp '{image_url}' "
+                    f"&& mv {local_path}.tmp {local_path}",
+                    timeout=timeout,
+                )
+
+            if not image_checksum:
+                return local_path
+
+            actual = self._run_node_command(
+                f"{algo}sum {local_path} | awk '{{print $1}}'", timeout=60
+            )
+            if actual.lower() == expected.lower():
+                return local_path
+
+            # Remove the bad file so the next attempt re-downloads instead of
+            # reusing it.
+            self._run_node_command(f"rm -f {local_path}", timeout=30)
+            if attempt == max_attempts:
+                raise RuntimeError(
+                    f"Checksum mismatch for {image_url}: expected {expected}, got {actual}"
+                )
+            _logger.warning(
+                f"Checksum mismatch for {image_url} on attempt {attempt}/{max_attempts} "
+                "— retrying download"
+            )
+
+        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').
+
+        Proxmox's /storage API omits the "enabled" field entirely for storages
+        that were never explicitly toggled — it is not present-and-falsy, it is
+        just absent, defaulting to enabled. Only an explicit 0 means disabled.
+
+        Queries the node-scoped /nodes/{node}/storage endpoint, not the
+        cluster-wide /storage one: a storage can be configured with a "nodes"
+        restriction limiting it to other cluster members, and the cluster-wide
+        list doesn't reflect that — it would happily return a storage this
+        node can't actually see, and "qm importdisk" would fail with
+        "storage 'X' is not available on node 'Y'" after the VM shell was
+        already created.
+        """
+        for storage in self._node_api().storage.get():
+            content = storage.get("content", "")
+            if "images" in content and storage.get("enabled", 1) != 0:
+                return storage["storage"]
+        raise ValueError(
+            "No storage with content='images' found. Configure a storage for VM disks."
+        )
+
+    def _get_storage_path(self, storage: str) -> str:
+        """Resolve a storage's filesystem path on the node.
+
+        Needed to write Cloud-Init snippets directly: Proxmox's
+        /storage/{s}/upload API only accepts content in {iso, vztmpl,
+        import} — "snippets" is rejected outright, so snippets must be
+        written straight to the filesystem instead. Only dir-backed storages
+        (dir, nfs, cifs, cephfs) expose "path"; those are also the only
+        storage types Proxmox itself allows content='snippets' on.
+        """
+        config = self._api.storage(storage).get()
+        path = config.get("path")
+        if not path:
+            raise ValueError(
+                f"Storage '{storage}' has no filesystem path (content='snippets' "
+                "requires a dir/nfs/cifs/cephfs-backed storage)"
+            )
+        return path
+
+    def get_image_storages(self) -> list[StorageTargetDict]:
+        """List node-available storage pools suitable for a new VM's root disk."""
+        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:
+                continue
+            if storage.get("active", 1) == 0:
+                continue
+            total = storage.get("total") or 0
+            avail = storage.get("avail") or 0
+            targets.append(
+                {
+                    "name": storage["storage"],
+                    "type": storage.get("type", ""),
+                    "total_gb": round(total / (1024**3), 1),
+                    "available_gb": round(avail / (1024**3), 1),
+                }
+            )
+        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.
@@ -51,68 +535,105 @@ class ProxmoxVMProvisionMixin:
         self,
         name: str,
         *,
-        template: str,
+        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],
-        ssh_public_keys: List[str] | None = None,
+        image_checksum: 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:
         """
-        Create a new VM from a Cloud-Init template via Proxmox API.
+        Create a new VM from a downloaded cloud image via Proxmox API.
 
         Steps:
           1. Get next available VMID from cluster
-          2. Clone template VM (full clone, new VMID)
+          2. Create an empty VM shell (no clone — no pre-existing template needed)
-          3. Configure CPU, memory, and network interfaces
+          3. Download the cloud image on the node (cached by filename) and
-          4. Verify snippet storage exists
+             import it as the VM's root disk
-          5. Render cloud-init config to YAML and upload
+          4. Configure CPU, memory, and network interfaces
-          6. Set Cloud-Init config references and SSH keys
+          5. Verify snippet storage exists
-          7. Optionally resize root disk
+          6. Render cloud-init config to YAML and upload
-          8. Start the VM
+          7. Set Cloud-Init config references and SSH keys
-          9. Return VMID, name, node
+          8. Optionally resize root disk
+          9. Start the VM
+          10. Return VMID, name, node
 
         Args:
             name: new VM display name
-            template: template VMID/name to clone from
+            image_url: URL of the cloud image to download and use as root disk
             cpu: number of vCPUs
             memory: RAM in MB
             nics: list of NIC config dicts (bridge, vlan_tag/trunk_vlan_tags, dhcp flag)
             cloud_init_config: user-data dict (will be YAML-rendered)
+            image_checksum: expected "<algo>:<hex>" checksum of the image, verified
+                after download (None = no verification)
             ssh_public_keys: SSH public keys to inject
             disk_resize_gb: resize root disk to this size (None = no resize)
-            timeout: max seconds for provisioning
+            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
 
         Returns:
             {"vmid": str, "name": str, "node": str}
 
         Raises:
-            RuntimeError: provisioning failure (clone, config, timeout, etc.)
+            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 template {template}")
+            _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()
             vmid = int(next_vmid)
             _logger.info(f"Allocated VMID {vmid}")
 
-            # Step 2: Clone template
+            # Step 2: Create empty VM shell (no disks yet)
-            _logger.info(f"Cloning template {template} → VMID {vmid}")
+            _logger.info(f"Creating VM shell {vmid}")
-            clone_upid = self._node_api().qemu(template).clone.post(
+            self._node_api().qemu.post(
-                newid=vmid,
+                vmid=vmid,
-                full=1,
                 name=name,
+                memory=memory,
+                cores=cpu,
+                cpu=cpu_model,
+                scsihw="virtio-scsi-pci",
+                # Includes agent: without it Proxmox never attaches the
+                # channel the QEMU guest agent needs — get_vm_status's agent
+                # queries (below) would have nothing to talk to.
+                **_GUEST_VM_SETTINGS[guest_os],
             )
-            self._wait_for_task(clone_upid, timeout=timeout)
 
-            # Step 3: Configure CPU, memory, and NICs
+            # Step 3: Download cloud image (cached) and import as root disk
-            _logger.info(f"Configuring VM {vmid}: {cpu} CPU, {memory}MB RAM, {len(nics)} NIC(s)")
+            image_storage = storage or self._find_default_image_storage()
+            self._attach_root_disk(
+                vmid, image_storage, image_url, image_checksum, download_timeout, timeout
+            )
 
-            config_args = {"cores": cpu, "memory": memory}
+            # 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] = {}
 
             # Build NIC config strings generically
             for i, nic in enumerate(nics):
@@ -120,8 +641,9 @@ class ProxmoxVMProvisionMixin:
                 if not bridge:
                     raise ValueError(f"NIC {i}: bridge is required")
 
-                # Build base config: model + bridge
+                # Build base config: model[=mac] + bridge
-                net_config = f"virtio,bridge={bridge}"
+                mac = nic.get("mac")
+                net_config = f"virtio={mac},bridge={bridge}" if mac else f"virtio,bridge={bridge}"
 
                 # Add VLAN configuration (access vs trunk)
                 if "trunk_vlan_tags" in nic and nic["trunk_vlan_tags"]:
@@ -134,13 +656,15 @@ class ProxmoxVMProvisionMixin:
 
             self._node_api().qemu(vmid).config.post(**config_args)
 
-            # Step 4: Verify snippet storage exists
+            # Step 5: Verify snippet storage exists
+            # (enabled is absent-not-falsy, and node-scoping matters — see
+            # _find_default_image_storage)
             _logger.info("Checking for snippet storage...")
-            storages = self._api.storage.get()
+            storages = self._node_api().storage.get()
             snippet_storage = None
             for storage in storages:
                 content = storage.get("content", "")
-                if "snippets" in content and storage.get("enabled"):
+                if "snippets" in content and storage.get("enabled", 1) != 0:
                     snippet_storage = storage["storage"]
                     break
 
@@ -152,7 +676,7 @@ class ProxmoxVMProvisionMixin:
                 )
             _logger.info(f"Using snippet storage: {snippet_storage}")
 
-            # Step 5: Render and upload Cloud-Init config
+            # Step 6: Render and upload Cloud-Init config
             _logger.info(f"Rendering Cloud-Init config for VMID {vmid}")
 
             user_data_yaml = "#cloud-config\n" + yaml.dump(
@@ -160,22 +684,34 @@ class ProxmoxVMProvisionMixin:
             )
 
             filename = f"{vmid}-user-data.yaml"
-            _logger.debug(f"Uploading Cloud-Init snippet {filename} to {snippet_storage}")
+            storage_path = self._get_storage_path(snippet_storage)
+            self._write_snippet(storage_path, filename, user_data_yaml)
+            cicustom = f"user={snippet_storage}:snippets/{filename}"
 
-            # Upload to snippet storage
+            # Proxmox writes network-config v1, and FreeBSD's nuageinit fails
-            self._node_api().storage(snippet_storage).upload.post(
+            # on it and then skips runcmd, which starts the guest agent. A
-                content="snippets",
+            # guest other than Linux gets the v2 every cloud-init reads.
-                filename=filename,
+            network = network_config(self._nic_macs(vmid, nics)) if guest_os != "linux" else None
-                data=user_data_yaml,
+            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 6: Configure Cloud-Init references and SSH keys
+            # Step 7: Configure Cloud-Init references and SSH keys
             _logger.info(f"Setting Cloud-Init config for VM {vmid}")
 
             cloud_init_args = {
-                "ide2": f"{snippet_storage}:cloudinit",
+                # The cloud-init drive is a disk image — it needs a storage
+                # with content='images' (same requirement as the root disk),
+                # NOT the snippet storage (content='snippets'). These are
+                # often different storages; Proxmox fails at VM start with
+                # "storage 'X' does not support content-type 'images'" if
+                # 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)
@@ -191,7 +727,7 @@ class ProxmoxVMProvisionMixin:
 
             self._node_api().qemu(vmid).config.post(**cloud_init_args)
 
-            # Step 7: Optionally resize root disk
+            # Step 8: Optionally resize root disk
             if disk_resize_gb is not None:
                 _logger.info(f"Resizing root disk to {disk_resize_gb}GB")
                 # Find root disk (scsi0, virtio0, ide0, sata0 — whichever is first)
@@ -212,7 +748,7 @@ class ProxmoxVMProvisionMixin:
                 except Exception as e:
                     _logger.warning(f"Failed to resize disk: {e}, continuing anyway")
 
-            # Step 8: Start the VM
+            # Step 9: Start the VM
             _logger.info(f"Starting VM {vmid}")
             start_upid = self._node_api().qemu(vmid).status.start.post()
             self._wait_for_task(start_upid, timeout=timeout)
@@ -263,30 +799,30 @@ 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
+            # request verbatim with no underscore-to-hyphen translation, so this
+            # must be built as a dict and unpacked rather than passed as a kwarg.
             _logger.debug(f"Deleting VM {vmid} configuration and disks")
-            self._node_api().qemu(vmid_int).delete(
+            delete_params = {
-                purge=1,
+                "purge": 1,
-                destroy_unreferenced_disks=1 if remove_disk else 0,
+                "destroy-unreferenced-disks": 1 if remove_disk else 0,
-            )
+            }
+            self._node_api().qemu(vmid_int).delete(**delete_params)
 
             # 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")
 
@@ -294,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,
@@ -325,51 +879,50 @@ class ProxmoxVMProvisionMixin:
             vmid_int = int(vmid)
             _logger.debug(f"Getting status for VM {vmid}")
 
-            # Get VM config to infer net0 MAC (for matching guest-agent results)
+            # VM must exist / be readable before we start polling.
             try:
-                config = self._node_api().qemu(vmid_int).config.get()
+                self._node_api().qemu(vmid_int).config.get()
             except Exception:
                 # VM may not exist yet or config not readable
                 return {"status": "unknown"}
 
-            # Parse net0 MAC from config (if present)
-            net0_line = config.get("net0", "")
-            expected_mac = None
-            # Example: "virtio,bridge=vmbr0,tag=10" — no explicit MAC
-            # Proxmox auto-generates MACs in a deterministic pattern, but we'll
-            # match by looking for the first NIC's IP in guest-agent results
-
             # Polling loop
             start_time = time.time()
             while True:
                 elapsed = time.time() - start_time
                 if wait_for_ip and elapsed > timeout:
-                    raise RuntimeError(
+                    raise RuntimeError(f"VM {vmid} failed to acquire IP within {timeout}s")
-                        f"VM {vmid} failed to acquire IP within {timeout}s"
-                    )
 
                 try:
-                    # Query guest-agent network interfaces
+                    # Query guest-agent network interfaces. Proxmox's REST path is
-                    agent_info = self._node_api().qemu(vmid_int).agent.network_get_interfaces.get()
+                    # "network-get-interfaces" (hyphens) — it must be passed as a
-                    interfaces = agent_info.get("result", [])
+                    # resource id via __call__, not dotted attribute access (which
+                    # would silently build a non-existent "network_get_interfaces"
+                    # path and 404 on every poll).
+                    agent_info = (
+                        self._node_api().qemu(vmid_int).agent("network-get-interfaces").get()
+                    )
+                    interfaces = (agent_info or {}).get("result", [])
 
-                    # Find net0 (first interface with IP)
+                    # The guest agent does not report interfaces in a fixed order —
-                    if interfaces:
+                    # loopback commonly comes first, named "lo" on Linux and
-                        net0_iface = interfaces[0]  # Assumes net0 is first in list
+                    # "lo0" on the BSDs. Skip loopback addresses, whatever the
-                        net0_mac = net0_iface.get("hardware-address", "")
+                    # interface is called, and take the first IPv4 address.
-                        ip_addresses = net0_iface.get("ip-addresses", [])
+                    for iface in interfaces:
-
+                        name = iface.get("name", "")
-                        if ip_addresses:
+                        if not name:
-                            # Found IP
+                            continue
-                            ip_info = ip_addresses[0]
+                        for addr in iface.get("ip-addresses", []):
-                            ip_addr = ip_info.get("ip-address", "")
+                            if addr.get("ip-address-type") != "ipv4":
-                            if ip_addr:
+                                continue
+                            ip_addr = addr.get("ip-address", "")
+                            if ip_addr and not ipaddress.ip_address(ip_addr).is_loopback:
                                 _logger.info(f"VM {vmid} acquired IP {ip_addr}")
                                 return {
                                     "status": "running",
                                     "ip_address": ip_addr,
-                                    "hostname": net0_iface.get("name", ""),
+                                    "hostname": name,
-                                    "mac_address": net0_mac,
+                                    "mac_address": iface.get("hardware-address", ""),
                                 }
 
                 except Exception as e:
@@ -386,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")
@@ -415,6 +968,15 @@ class ProxmoxVMProvisionMixin:
             if not name:
                 continue
             # A vnet's VLAN is already fixed by its zone/tag — no separate vlan_tag applies.
-            targets.append({"name": name, "kind": "vnet", "vlan_aware": False})
+            tag = vnet.get("tag")
+            fixed_vlan_tag = int(tag) if tag is not None else None
+            targets.append(
+                {
+                    "name": name,
+                    "kind": "vnet",
+                    "vlan_aware": False,
+                    "fixed_vlan_tag": fixed_vlan_tag,
+                }
+            )
 
         return targets

napalm_proxmox/vm_snapshot_mixin.py

@@ -0,0 +1,74 @@
+"""HypervisorDriver snapshot methods for Proxmox VE guests (QEMU and LXC)."""
+
+from __future__ import annotations
+
+from typing import Any
+
+from napalm_device_types.models import SnapshotDict
+
+#: Proxmox lists the live state as a pseudo-snapshot of this name.
+_CURRENT = "current"
+#: A RAM snapshot of a large VM takes a while to write out.
+_SNAPSHOT_TIMEOUT = 600
+
+
+class ProxmoxVMSnapshotMixin:
+    """Relies on ``_resolve_vm``/``_guest_api`` from ProxmoxVMContractMixin."""
+
+    _resolve_vm: Any
+    _guest_api: Any
+    _wait_for_task: Any
+
+    def _snapshots(self, name: str) -> tuple[Any, str, str, list[dict[str, Any]]]:
+        vmid, vm_type = self._resolve_vm(name)
+        api = self._guest_api(vmid, vm_type)
+        raw = [s for s in api.snapshot.get() or [] if s.get("name") != _CURRENT]
+        return api, str(vmid), vm_type, raw
+
+    def _run_task(self, call: Any, *args: Any, **kwargs: Any) -> None:
+        try:
+            upid = call(*args, **kwargs)
+        except Exception as exc:
+            raise RuntimeError(str(exc)) from exc
+        if upid:
+            self._wait_for_task(upid, timeout=_SNAPSHOT_TIMEOUT)
+
+    @staticmethod
+    def _require(raw: list[dict[str, Any]], snapshot: str, vm: str) -> None:
+        if not any(s.get("name") == snapshot for s in raw):
+            raise ValueError(f"VM {vm!r} has no snapshot named {snapshot!r}")
+
+    def get_vm_snapshots(self, name: str) -> list[SnapshotDict]:
+        _, _, _, raw = self._snapshots(name)
+        return [
+            {
+                "name": s["name"],
+                "vm": name,
+                "created": float(s.get("snaptime", 0)),
+                "description": s.get("description", ""),
+                "has_memory": bool(s.get("vmstate")),
+                "parent": s.get("parent", ""),
+            }
+            for s in raw
+        ]
+
+    def create_vm_snapshot(
+        self, name: str, snapshot: str, description: str = "", include_memory: bool = False
+    ) -> None:
+        api, _, vm_type, raw = self._snapshots(name)
+        if any(s.get("name") == snapshot for s in raw):
+            raise ValueError(f"VM {name!r} already has a snapshot named {snapshot!r}")
+        kwargs: dict[str, Any] = {"snapname": snapshot, "description": description}
+        if vm_type == "vm":  # containers have no RAM state to save
+            kwargs["vmstate"] = 1 if include_memory else 0
+        self._run_task(api.snapshot.post, **kwargs)
+
+    def delete_vm_snapshot(self, name: str, snapshot: str) -> None:
+        api, _, _, raw = self._snapshots(name)
+        self._require(raw, snapshot, name)
+        self._run_task(api.snapshot(snapshot).delete)
+
+    def rollback_vm_snapshot(self, name: str, snapshot: str) -> None:
+        api, _, _, raw = self._snapshots(name)
+        self._require(raw, snapshot, name)
+        self._run_task(api.snapshot(snapshot).rollback.post)

pyproject.toml

@@ -1,31 +1,29 @@
 [build-system]
-requires = ["setuptools>=61", "wheel"]
+requires = ["setuptools>=77"]
 build-backend = "setuptools.build_meta"
 
 [project]
 name = "napalm-proxmox"
-version = "0.1.0"
+version = "0.8.0"
 description = "NAPALM driver for Proxmox VE (classic Linux networking, SDN, OVS)"
 readme = "README.md"
-license = { text = "Apache-2.0" }
+license = "Apache-2.0"
 authors = [{ name = "Christian Manivong" }]
 keywords = ["napalm", "proxmox", "network", "sdn", "ovs", "automation"]
 classifiers = [
     "Development Status :: 3 - Alpha",
     "Intended Audience :: Developers",
     "Intended Audience :: System Administrators",
-    "License :: OSI Approved :: Apache Software License",
     "Programming Language :: Python :: 3",
-    "Programming Language :: Python :: 3.9",
     "Programming Language :: Python :: 3.10",
     "Programming Language :: Python :: 3.11",
     "Programming Language :: Python :: 3.12",
     "Topic :: System :: Networking",
 ]
-requires-python = ">=3.9"
+requires-python = ">=3.10"
 dependencies = [
     "napalm>=5.0.0",
-    "napalm_device_types>=0.1.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

tests/conftest.py

@@ -2,11 +2,11 @@
 
 from __future__ import annotations
 
-import pytest
 from unittest.mock import MagicMock, patch
 
-from napalm_proxmox.driver import ProxmoxDriver
+import pytest
 
+from napalm_proxmox.driver import ProxmoxDriver
 
 # ---------------------------------------------------------------------------
 # Sample API payloads
@@ -91,7 +91,10 @@ SDN_SUBNETS_VNET1 = [
 
 SDN_SUBNETS_VNET2: list = []
 
-DNS_INFO = {"search": "pve1.example.com", "dns1": "8.8.8.8"}
+# A DNS *search domain*, not an FQDN. It used to read "pve1.example.com",
+# which made get_facts build "pve1.pve1.example.com" and looked like a
+# driver bug rather than bad test data.
+DNS_INFO = {"search": "example.com", "dns1": "8.8.8.8"}
 
 NTP_INFO = {"server": "pool.ntp.org,time.cloudflare.com"}
 
@@ -121,6 +124,7 @@ PVE_USERS = [
 # Driver fixture
 # ---------------------------------------------------------------------------
 
+
 def _build_mock_api(
     nodes=None,
     node_status=None,
@@ -135,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()
@@ -162,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
 

tests/quality_baseline.json

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

tests/test_connection.py

@@ -2,12 +2,13 @@
 
 from __future__ import annotations
 
-import pytest
+from unittest.mock import 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:
@@ -34,7 +35,10 @@ class TestOpen:
         with patch("napalm_proxmox.driver.ProxmoxAPI", return_value=mock_api) as mock_cls:
             drv.open()
         call_kwargs = mock_cls.call_args.kwargs
-        assert call_kwargs["user"] == "napalm@pam!mytoken"
+        # proxmoxer wants the two halves separately, not the combined
+        # "<user>!<tokenid>" string that Proxmox's UI displays.
+        assert call_kwargs["user"] == "napalm@pam"
+        assert call_kwargs["token_name"] == "mytoken"
         assert call_kwargs["token_value"] == "super-secret"
 
     def test_open_connection_error(self):
@@ -73,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()

tests/test_environment.py

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

tests/test_fix_templates.py

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

tests/test_full_upgrade.py

@@ -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

tests/test_get_facts.py

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

tests/test_interfaces.py

@@ -2,11 +2,7 @@
 
 from __future__ import annotations
 
-import pytest
+from tests.conftest import 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}"
 

tests/test_kernel_facts.py

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

tests/test_licenses.py

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

tests/test_listening_sockets.py

@@ -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")

tests/test_misc.py

@@ -2,8 +2,9 @@
 
 from __future__ import annotations
 
+from unittest.mock import patch
+
 import pytest
-from unittest.mock import MagicMock, patch
 
 
 class TestGetNTPServers:
@@ -85,7 +86,7 @@ class TestGetUsers:
     def test_user_has_required_keys(self, driver):
         driver._node_api().execute.post.return_value = {"data": ""}
         result = driver.get_users()
-        for user, data in result.items():
+        for _user, data in result.items():
             assert "level" in data
             assert "password" in data
             assert "sshkeys" in data
@@ -248,7 +249,7 @@ class TestGetRouteTo:
     def test_route_fields(self, driver):
         driver._node_api().execute.post.return_value = {"data": self.ROUTE_OUTPUT}
         result = driver.get_route_to()
-        for prefix, routes in result.items():
+        for _prefix, routes in result.items():
             for route in routes:
                 assert "protocol" in route
                 assert "next_hop" in route
@@ -257,23 +258,34 @@ 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()
 
 
 class TestLLDPNeighbors:
+    # Real `lldpcli show neighbors summary` output. The interface line carries
+    # ", via: LLDP, ..." after the name, which is what the parser matches on —
+    # the previous fixture stopped at the name and matched nothing.
     LLDP_SUMMARY = (
-        "  Interface: eth0\n"
+        "LLDP neighbors:\n"
+        "-------------------------------------------------------------------------------\n"
+        "Interface:    eth0, via: LLDP, RID: 1, Time: 0 day, 00:11:22\n"
+        "  Chassis:\n"
         "    SysName:      sw01.example.com\n"
+        "  Port:\n"
         "    PortID:       ifname GigabitEthernet1/0/1\n"
-        "  Interface: eth1\n"
+        "-------------------------------------------------------------------------------\n"
+        "Interface:    eth1, via: LLDP, RID: 2, Time: 0 day, 00:11:22\n"
+        "  Chassis:\n"
         "    SysName:      sw02.example.com\n"
+        "  Port:\n"
         "    PortID:       ifname GigabitEthernet1/0/2\n"
+        "-------------------------------------------------------------------------------\n"
     )
 
     def test_neighbors_found(self, driver):
-        driver._node_api().execute.post.return_value = {"data": self.LLDP_SUMMARY}
+        with patch.object(driver, "_exec_ssh_command", return_value=self.LLDP_SUMMARY):
             result = driver.get_lldp_neighbors()
         assert "eth0" in result
         assert result["eth0"][0]["hostname"] == "sw01.example.com"

tests/test_ovs_and_arp.py

@@ -2,11 +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
@@ -126,15 +125,17 @@ class TestGetARPTable:
 
 
 class TestGetMACAddressTable:
+    # Mocked at _exec_ssh_command, the driver's own seam. Mocking the API call
+    # underneath it broke twice over: that helper passes two positional
+    # arguments where these doubles accepted one, and it base64-wraps the
+    # command, so a fixture keyed on "bridge fdb" appearing in the text never
+    # matched.
     BRIDGE_FDB = (
         "aa:bb:cc:dd:ee:01 dev eth0 vlan 10 master vmbr0 permanent\n"
         "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):
@@ -142,9 +143,7 @@ class TestGetMACAddressTable:
                 return self.BRIDGE_FDB
             return ""
 
-        driver._node_api().execute.post.side_effect = lambda command: {
+        with patch.object(driver, "_exec_ssh_command", side_effect=_exec):
-            "data": _exec(command)
-        }
             result = driver.get_mac_address_table()
         macs = {e["mac"] for e in result}
         assert "aa:bb:cc:dd:ee:01" in macs
@@ -155,9 +154,7 @@ class TestGetMACAddressTable:
                 return self.BRIDGE_FDB
             return ""
 
-        driver._node_api().execute.post.side_effect = lambda command: {
+        with patch.object(driver, "_exec_ssh_command", side_effect=_exec):
-            "data": _exec(command)
-        }
             result = driver.get_mac_address_table()
         static_entries = [e for e in result if e["mac"] == "aa:bb:cc:dd:ee:01"]
         assert static_entries[0]["static"] is True

tests/test_packages.py

@@ -0,0 +1,67 @@
+"""`get_packages` and the source coordinate OSV matching needs.
+
+A Proxmox node is a Debian host, so its packages are matched against Debian
+advisories — and OSV states those ranges in *source* package versions. Reporting
+the source package without its version leaves a consumer holding two numbers on
+different axes: `libldb2` is `2:2.11.0+samba4.22.11+dfsg-…` while its source,
+samba, is `2:4.22.11+dfsg-…`, and comparing the first against a samba range is
+meaningless.
+
+Measured before this was fixed: on three Proxmox nodes, **every one** of their
+2 349 packages carried a source package and no source version — 802 of 802,
+774 of 774, 773 of 773 — while twenty non-Proxmox hosts had both. The format
+string simply never asked for the field.
+"""
+
+from __future__ import annotations
+
+import pytest
+
+# `${Package}\t${Version}\t${db:Status-Status}\t${Installed-Size}\t${source:Package}\t${source:Version}`
+DPKG = (
+    "openssh-server\t1:9.2p1-2\tinstalled\t1024\topenssh\t1:9.2p1-2\n"
+    "libldb2\t2:2.11.0+samba4.22.11+dfsg-0+deb13u1\tinstalled\t512\tsamba\t2:4.22.11+dfsg-0+deb13u1\n"
+    "curl\t7.88.1-10\tinstalled\t256\t\t\n"
+)
+
+
+@pytest.mark.parametrize("driver_with_exec", [DPKG], indirect=True)
+def test_the_source_version_is_reported(driver_with_exec):
+    """The field the whole fix is about."""
+    packages = {p["name"]: p for p in driver_with_exec.get_packages()}
+
+    assert packages["libldb2"]["source_package"] == "samba"
+    assert packages["libldb2"]["source_version"] == "2:4.22.11+dfsg-0+deb13u1"
+
+
+@pytest.mark.parametrize("driver_with_exec", [DPKG], indirect=True)
+def test_the_binary_version_is_kept_beside_it(driver_with_exec):
+    """Both are wanted: one says what is installed, the other is the axis the
+    advisory's range is stated on."""
+    libldb2 = {p["name"]: p for p in driver_with_exec.get_packages()}["libldb2"]
+
+    assert libldb2["version"] == "2:2.11.0+samba4.22.11+dfsg-0+deb13u1"
+
+
+@pytest.mark.parametrize("driver_with_exec", [DPKG], indirect=True)
+def test_an_empty_source_falls_back_to_the_package_itself(driver_with_exec):
+    """dpkg leaves both fields empty when the source is the package — and an
+    older dpkg leaves them empty because it does not know the field at all.
+    Neither may produce a package with no coordinate."""
+    curl = {p["name"]: p for p in driver_with_exec.get_packages()}["curl"]
+
+    assert curl["source_package"] == "curl"
+    assert curl["source_version"] == "7.88.1-10"
+
+
+def test_the_query_asks_for_the_field():
+    """The defect was in the format string, not in the parsing: the driver
+    reported a source package it had asked for and a source version it had
+    not, so no amount of parsing could have produced one."""
+    import inspect
+
+    from napalm_proxmox import system_mixin
+
+    source = inspect.getsource(system_mixin.ProxmoxSystemMixin.get_packages)
+
+    assert "source:Version" in source

tests/test_reboot_host.py

@@ -0,0 +1,16 @@
+"""reboot_host: restart the Proxmox node itself through the API, not over SSH."""
+
+from __future__ import annotations
+
+import pytest
+
+
+def test_posts_reboot_to_the_node(driver):
+    driver.reboot_host()
+    driver._node_api().status.post.assert_called_once_with(command="reboot")
+
+
+def test_api_refusal_is_a_runtime_error(driver):
+    driver._node_api().status.post.side_effect = Exception("Permission check failed")
+    with pytest.raises(RuntimeError, match="Permission check failed"):
+        driver.reboot_host()

tests/test_sdn.py

@@ -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:
@@ -22,24 +21,38 @@ class TestGetVlans:
         result = driver.get_vlans()
         assert "100000" in result
 
-    def test_bridge_vlan_show_parsing(self, driver):
+    def test_membership_derived_from_vm_configs(self, driver):
-        # Simulate bridge vlan output
+        """Without OVS ports, VLAN membership comes from each VM's netN config.
-        bridge_output = (
+
-            "vmbr0   1\n"
+        This replaces a test for a `bridge vlan show` fallback that no longer
-            "        10\n"
+        exists — it also asserted an "interfaces" key this method has never
-            "        20\n"
+        produced, so it could not have passed against any version of the code.
-            "eth0    1\n"
+        """
-        )
+        node = driver._node_api()
-        driver._node_api().execute.post.return_value = {"data": bridge_output}
+        node.qemu.get.return_value = [{"vmid": 100}]
+        node.lxc.get.return_value = []
+        node.qemu.return_value.config.get.return_value = {
+            "net0": "virtio=AA:BB:CC:DD:EE:FF,bridge=vmbr0,tag=10",
+        }
+
         result = driver.get_vlans()
-        # Interface vmbr0 should appear in vlan 1
+        assert "vmbr0" in result["10"]["untagged"]
-        entry = result.get("1", {})
+
-        assert "vmbr0" in entry.get("interfaces", [])
+    def test_configured_vnets_appear_even_without_members(self, driver):
+        """An SDN VNet exists on the node whether or not anything is attached.
+
+        Entries with no member ports used to be filtered out of this one return
+        path while the OVS path returned them, so a configured VLAN was visible
+        or invisible depending on which branch ran.
+        """
+        result = driver.get_vlans()
+        assert result["20"]["name"] == "vnet1"
+        assert result["20"]["untagged"] == []
 
     def test_empty_sdn_returns_dict(self):
         from unittest.mock import patch
+
         from napalm_proxmox.driver import ProxmoxDriver
-        from tests.conftest import _build_mock_api
 
         mock_api = _build_mock_api(sdn_vnets=[], sdn_zones=[])
         drv = ProxmoxDriver("pve1", "root", "secret", optional_args={"verify_ssl": False})

tests/test_service_packages.py

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

tests/test_services.py

@@ -0,0 +1,68 @@
+"""Services on a Proxmox node: systemd, through napalm-device-types' mixin.
+
+The listing used to read ``list-unit-files``' second column, which since
+systemd 245 is followed by a preset column -- so ``enabled`` was false for
+every service on every node (#6). An action ended in ``|| true`` and reported
+success whatever happened. Both now come from the shared mixin; the driver
+only carries the command over its exec path, as root.
+"""
+
+from __future__ import annotations
+
+from unittest.mock import patch
+
+import pytest
+from napalm_device_types import SystemdServicesMixin
+from napalm_device_types.systemd import SYSTEMD_SERVICES_COMMAND, service_action_command
+
+from napalm_proxmox.driver import ProxmoxDriver
+
+REPORT = (
+    "SVC_BEGIN\n[files]\npveproxy.service enabled enabled\n[units]\n"
+    "MainPID=1234\nId=pveproxy.service\nNames=pveproxy.service\nLoadState=loaded\n"
+    "ActiveState=active\nSubState=running\nUnitFileState=enabled\n"
+    "[generated]\nSVC_END\n"
+)
+
+
+def test_the_driver_uses_the_shared_mixin():
+    assert issubclass(ProxmoxDriver, SystemdServicesMixin)
+    assert ProxmoxDriver.get_services is SystemdServicesMixin.get_services
+    assert ProxmoxDriver.manage_service is SystemdServicesMixin.manage_service
+
+
+def test_listing_runs_the_shared_command(driver):
+    with patch.object(driver, "_exec_ssh_command", return_value=REPORT) as exec_:
+        services = driver.get_services()
+
+    exec_.assert_called_once_with(SYSTEMD_SERVICES_COMMAND)
+    assert services == [{"name": "pveproxy", "running": True, "enabled": True, "pid": 1234}]
+
+
+def test_an_unreadable_listing_raises_instead_of_reporting_no_services(driver):
+    with patch.object(driver, "_exec_ssh_command", return_value=""):
+        with pytest.raises(ValueError):
+            driver.get_services()
+
+
+def test_an_action_reports_its_real_outcome(driver):
+    failed = "Failed to restart nope.service: Unit nope.service not found.\n__SVC_RC=5"
+    with patch.object(driver, "_exec_ssh_command", return_value=failed) as exec_:
+        result = driver.manage_service("nope", "restart")
+
+    exec_.assert_called_once_with(service_action_command("nope", "restart"))
+    assert result["success"] is False
+    assert "not found" in result["output"]
+
+
+def test_a_successful_action(driver):
+    with patch.object(driver, "_exec_ssh_command", return_value="__SVC_RC=0"):
+        assert driver.manage_service("pveproxy", "restart") == {"success": True, "output": ""}
+
+
+def test_an_invalid_name_never_reaches_the_node(driver):
+    with patch.object(driver, "_exec_ssh_command") as exec_:
+        with pytest.raises(ValueError):
+            driver.manage_service("pveproxy; reboot", "stop")
+
+    exec_.assert_not_called()

tests/test_ssh_host_key.py

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

tests/test_system_identity.py

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

tests/test_updates.py

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

tests/test_utils.py

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

tests/test_vm_contract.py

@@ -0,0 +1,239 @@
+"""HypervisorDriver contract methods: VM lookup, power actions, get_vm_config.
+
+netOrk used to call Proxmox's own ``power_vm`` and reach into ``_node_api()``
+for a VM's hardware. Both are Proxmox-only, so a second hypervisor could not
+serve the same endpoints. These pin the contract methods that replace them.
+"""
+
+from __future__ import annotations
+
+from unittest.mock import MagicMock
+
+import pytest
+
+from napalm_proxmox.vm_contract_mixin import parse_vm_config
+
+QEMU_LIST = [{"vmid": 100, "name": "web01", "status": "running"}]
+LXC_LIST = [{"vmid": 200, "name": "dns01", "status": "running"}]
+
+
+@pytest.fixture
+def api(driver):
+    node = driver._node_api()
+    node.qemu.get.return_value = QEMU_LIST
+    node.lxc.get.return_value = LXC_LIST
+    node.qemu.return_value.status.start.post.return_value = "UPID:start"
+    driver._wait_for_task = MagicMock()
+    return node
+
+
+class TestGetVmsReportsStringIds:
+    def test_vmid_is_a_string(self, driver, api):
+        driver.get_vm_interfaces = MagicMock(return_value=({}, False, False))
+        driver._get_vm_disk_and_boot = MagicMock(return_value=([], False))
+        assert [vm["vmid"] for vm in driver.get_vms()] == ["100", "200"]
+
+    def test_ordered_numerically_not_lexically(self, driver, api):
+        api.qemu.get.return_value = [{"vmid": 1000, "name": "a"}, {"vmid": 99, "name": "b"}]
+        api.lxc.get.return_value = []
+        driver.get_vm_interfaces = MagicMock(return_value=({}, False, False))
+        driver._get_vm_disk_and_boot = MagicMock(return_value=([], False))
+        assert [vm["vmid"] for vm in driver.get_vms()] == ["99", "1000"]
+
+
+class TestResolveVm:
+    def test_by_vmid_string(self, driver, api):
+        assert driver._resolve_vm("100") == (100, "vm")
+
+    def test_by_name(self, driver, api):
+        assert driver._resolve_vm("dns01") == (200, "container")
+
+    def test_unknown_raises_value_error(self, driver, api):
+        with pytest.raises(ValueError, match="nope"):
+            driver._resolve_vm("nope")
+
+
+class TestPowerActions:
+    def test_start_posts_and_waits_for_the_task(self, driver, api):
+        driver.start_vm("web01")
+        api.qemu.return_value.status.start.post.assert_called_once()
+        driver._wait_for_task.assert_called_once_with("UPID:start", timeout=120)
+
+    @pytest.mark.parametrize(("force", "action"), [(False, "shutdown"), (True, "stop")])
+    def test_stop_graceful_or_forced(self, driver, api, force, action):
+        driver.stop_vm("100", force=force)
+        getattr(api.qemu.return_value.status, action).post.assert_called_once()
+
+    @pytest.mark.parametrize(("force", "action"), [(False, "reboot"), (True, "reset")])
+    def test_reboot_graceful_or_forced(self, driver, api, force, action):
+        driver.reboot_vm("100", force=force)
+        getattr(api.qemu.return_value.status, action).post.assert_called_once()
+
+    def test_forced_reboot_of_a_container_is_a_stop_and_start(self, driver, api):
+        """LXC has no reset; stop + start is the closest thing to pulling the plug."""
+        driver.reboot_vm("200", force=True)
+        status = api.lxc.return_value.status
+        status.stop.post.assert_called_once()
+        status.start.post.assert_called_once()
+
+    def test_suspend_vm(self, driver, api):
+        driver.suspend_vm("100")
+        api.qemu.return_value.status.suspend.post.assert_called_once()
+
+    def test_suspend_container_is_refused(self, driver, api):
+        with pytest.raises(RuntimeError, match="container"):
+            driver.suspend_vm("200")
+
+    def test_api_error_becomes_runtime_error(self, driver, api):
+        api.qemu.return_value.status.start.post.side_effect = Exception("locked")
+        with pytest.raises(RuntimeError, match="locked"):
+            driver.start_vm("100")
+
+
+QEMU_CONFIG = {
+    "name": "web01",
+    "cores": 2,
+    "sockets": 2,
+    "memory": "8192",
+    "ostype": "l26",
+    "cpu": "host,flags=+aes",
+    "bios": "ovmf",
+    "machine": "q35",
+    "boot": "order=scsi0;ide2;net0",
+    "scsi0": "local-lvm:vm-100-disk-0,size=32G,format=raw",
+    "virtio1": "tank:vm-100-disk-1,size=512M",
+    "ide2": "local:iso/debian.iso,media=cdrom",
+    "efidisk0": "local-lvm:vm-100-disk-2,size=4M",
+    "net0": "virtio=BC:24:11:AA:BB:CC,bridge=vmbr0,tag=10,firewall=1",
+    "net1": "e1000=BC:24:11:AA:BB:DD,bridge=vmbr1",
+    "hostpci0": "0000:01:00.0,pcie=1",
+    "usb0": "host=1234:5678",
+    "description": "Production web server",
+    "tags": "prod;web",
+}
+
+LXC_CONFIG = {
+    "hostname": "dns01",
+    "cores": 1,
+    "memory": 512,
+    "ostype": "debian",
+    "rootfs": "local-lvm:vm-200-disk-0,size=8G",
+    "mp0": "tank:subvol-200-disk-1,mp=/data,size=1T",
+    "net0": "name=eth0,bridge=vmbr0,hwaddr=BC:24:11:00:00:01,ip=dhcp,tag=20,type=veth",
+}
+
+
+class TestParseQemuConfig:
+    @pytest.fixture
+    def cfg(self):
+        return parse_vm_config("100", "vm", QEMU_CONFIG)
+
+    def test_core_fields(self, cfg):
+        assert cfg["vmid"] == "100"
+        assert cfg["name"] == "web01"
+        assert cfg["vcpus"] == 4
+        assert cfg["memory"] == 8192
+        assert cfg["os_type"] == "l26"
+        assert cfg["description"] == "Production web server"
+        assert cfg["tags"] == ["prod", "web"]
+
+    def test_boot_order(self, cfg):
+        assert cfg["boot_order"] == ["scsi0", "ide2", "net0"]
+
+    def test_disks_skip_cdrom_and_normalise_size(self, cfg):
+        by_dev = {d["device"]: d for d in cfg["disks"]}
+        assert set(by_dev) == {"scsi0", "virtio1", "efidisk0"}
+        assert by_dev["scsi0"] == {
+            "device": "scsi0",
+            "storage": "local-lvm",
+            "size": 32,
+            "format": "raw",
+            "bootable": True,
+        }
+        assert by_dev["virtio1"]["size"] == 0  # 512M rounds down to 0 GB
+        assert by_dev["virtio1"]["bootable"] is False
+
+    def test_nics(self, cfg):
+        assert cfg["nics"] == [
+            {
+                "device": "net0",
+                "mac": "BC:24:11:AA:BB:CC",
+                "model": "virtio",
+                "bridge": "vmbr0",
+                "vlan_id": 10,
+            },
+            {
+                "device": "net1",
+                "mac": "BC:24:11:AA:BB:DD",
+                "model": "e1000",
+                "bridge": "vmbr1",
+                "vlan_id": 0,
+            },
+        ]
+
+    def test_hardware_details(self, cfg):
+        assert cfg["cpu_type"] == "host"
+        assert cfg["sockets"] == 2
+        assert cfg["cores_per_socket"] == 2
+        assert cfg["firmware"] == "efi"
+        assert cfg["machine"] == "q35"
+
+    def test_passthrough(self, cfg):
+        assert cfg["passthrough"] == [
+            {"slot": "hostpci0", "kind": "pci", "config": "0000:01:00.0,pcie=1"},
+            {"slot": "usb0", "kind": "usb", "config": "host=1234:5678"},
+        ]
+
+    def test_defaults_for_a_bare_config(self):
+        cfg = parse_vm_config("101", "vm", {})
+        assert cfg["name"] == "vm-101"
+        assert cfg["vcpus"] == 1
+        assert cfg["cpu_type"] == "kvm64"
+        assert cfg["firmware"] == "bios"
+        assert "machine" not in cfg
+        assert cfg["boot_order"] == []
+
+    def test_legacy_bootdisk(self):
+        cfg = parse_vm_config("101", "vm", {"boot": "cdn", "bootdisk": "scsi0"})
+        assert cfg["boot_order"] == ["scsi0"]
+
+
+class TestParseLxcConfig:
+    @pytest.fixture
+    def cfg(self):
+        return parse_vm_config("200", "container", LXC_CONFIG)
+
+    def test_core_fields(self, cfg):
+        assert cfg["name"] == "dns01"
+        assert cfg["vcpus"] == 1
+        assert cfg["memory"] == 512
+        assert cfg["tags"] == []
+
+    def test_rootfs_and_mountpoint(self, cfg):
+        assert [(d["device"], d["storage"], d["size"]) for d in cfg["disks"]] == [
+            ("rootfs", "local-lvm", 8),
+            ("mp0", "tank", 1024),
+        ]
+
+    def test_veth_nic(self, cfg):
+        assert cfg["nics"] == [
+            {
+                "device": "net0",
+                "mac": "BC:24:11:00:00:01",
+                "model": "veth",
+                "bridge": "vmbr0",
+                "vlan_id": 20,
+            }
+        ]
+
+    def test_no_vm_only_hardware_fields(self, cfg):
+        assert "firmware" not in cfg
+        assert "sockets" not in cfg
+
+
+class TestGetVmConfig:
+    def test_fetches_the_right_config(self, driver, api):
+        api.lxc.return_value.config.get.return_value = LXC_CONFIG
+        cfg = driver.get_vm_config("dns01")
+        assert cfg["vmid"] == "200"
+        assert cfg["name"] == "dns01"

tests/test_vm_cpu_types.py

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

tests/test_vm_guest_os.py

@@ -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"]

tests/test_vm_import_download.py

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

tests/test_vm_snapshots.py

@@ -0,0 +1,95 @@
+"""HypervisorDriver snapshot methods on Proxmox (QEMU and LXC)."""
+
+from __future__ import annotations
+
+from unittest.mock import MagicMock
+
+import pytest
+
+SNAPSHOTS = [
+    {"name": "base", "description": "clean install", "snaptime": 1700000000, "vmstate": 0},
+    {"name": "upgrade", "description": "", "snaptime": 1700000100, "parent": "base", "vmstate": 1},
+    {"name": "current", "description": "You are here!", "parent": "upgrade", "running": 1},
+]
+
+
+@pytest.fixture
+def api(driver):
+    node = driver._node_api()
+    node.qemu.get.return_value = [{"vmid": 100, "name": "web01"}]
+    node.lxc.get.return_value = [{"vmid": 200, "name": "dns01"}]
+    node.qemu.return_value.snapshot.get.return_value = SNAPSHOTS
+    driver._wait_for_task = MagicMock()
+    return node
+
+
+class TestList:
+    def test_flattens_and_skips_the_current_marker(self, driver, api):
+        assert driver.get_vm_snapshots("web01") == [
+            {
+                "name": "base",
+                "vm": "web01",
+                "created": 1700000000.0,
+                "description": "clean install",
+                "has_memory": False,
+                "parent": "",
+            },
+            {
+                "name": "upgrade",
+                "vm": "web01",
+                "created": 1700000100.0,
+                "description": "",
+                "has_memory": True,
+                "parent": "base",
+            },
+        ]
+
+    def test_unknown_vm(self, driver, api):
+        with pytest.raises(ValueError):
+            driver.get_vm_snapshots("nope")
+
+
+class TestCreate:
+    def test_vm_with_memory(self, driver, api):
+        api.qemu.return_value.snapshot.post.return_value = "UPID:snap"
+        driver.create_vm_snapshot("100", "pre", description="d", include_memory=True)
+        api.qemu.return_value.snapshot.post.assert_called_once_with(
+            snapname="pre", description="d", vmstate=1
+        )
+        driver._wait_for_task.assert_called_once_with("UPID:snap", timeout=600)
+
+    def test_container_never_saves_memory(self, driver, api):
+        api.lxc.return_value.snapshot.get.return_value = []
+        driver.create_vm_snapshot("200", "pre", include_memory=True)
+        api.lxc.return_value.snapshot.post.assert_called_once_with(snapname="pre", description="")
+
+    def test_duplicate_name(self, driver, api):
+        with pytest.raises(ValueError, match="already"):
+            driver.create_vm_snapshot("web01", "base")
+
+    def test_api_refusal(self, driver, api):
+        api.qemu.return_value.snapshot.post.side_effect = Exception(
+            "snapshot feature is not available"
+        )
+        with pytest.raises(RuntimeError, match="not available"):
+            driver.create_vm_snapshot("web01", "new")
+
+
+class TestDeleteAndRollback:
+    def test_delete(self, driver, api):
+        driver.delete_vm_snapshot("web01", "base")
+        api.qemu.return_value.snapshot.assert_called_with("base")
+        api.qemu.return_value.snapshot.return_value.delete.assert_called_once_with()
+
+    def test_rollback(self, driver, api):
+        driver.rollback_vm_snapshot("web01", "upgrade")
+        api.qemu.return_value.snapshot.return_value.rollback.post.assert_called_once_with()
+
+    @pytest.mark.parametrize("method", ["delete_vm_snapshot", "rollback_vm_snapshot"])
+    def test_unknown_snapshot(self, driver, api, method):
+        with pytest.raises(ValueError, match="no snapshot"):
+            getattr(driver, method)("web01", "nope")
+
+    def test_current_is_not_a_snapshot(self, driver, api):
+        with pytest.raises(ValueError):
+            driver.rollback_vm_snapshot("web01", "current")