Merge pull request 'feat(vm-provision): name a CPU model for every new VM, and list the choices' (#5) from feat/vm-cpu-type into master

A VM created without a cpu argument gets Proxmox's API default, kvm64: no
AES-NI and no AVX. MongoDB 5.0 and later exit with "Illegal instruction" on
it, which is how a Graylog provisioned through netOrk failed (netork#494).
Every VM built by hand in the same cluster uses host or x86-64-v2-AES; only
the ones this driver created were left on kvm64.

create_vm_from_cloud_init now always passes cpu=, defaulting to x86-64-v2-AES
as the Proxmox GUI does since PVE 8, and takes cpu_type for a different one.

get_vm_cpu_types() lists x86-64-v2-AES (default), x86-64-v3 and host. The
named models carry the cpuinfo flags they add over qemu64, following
Proxmox's own definitions (CPUConfig.pm), and are available when the node's
CPU has all of them, read from /nodes/{node}/status. host lists the node's
own flags, so on a CPU without AVX it does not pretend to offer any.

A model asked for by name is checked against the node before a VMID is
allocated: one the CPU cannot run would only fail at VM start, after the disk
import, leaving a half-built VM behind. The default is not checked, so a
plain create costs no extra API call. VMCpuTypeDict is imported for type
checking only, so this works with an older napalm_device_types.

CHANGELOG.md

@@ -7,6 +7,28 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Added
+- 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

napalm_proxmox/driver.py

@@ -49,6 +49,8 @@ from napalm_proxmox.sdn_mixin import ProxmoxSDNMixin
 from napalm_proxmox.lldp_mixin import ProxmoxLLDPMixin
 from napalm_proxmox.config_mixin import ProxmoxConfigMixin
 from napalm_proxmox.vm_mixin import ProxmoxVMMixin
+from napalm_proxmox.vm_contract_mixin import ProxmoxVMContractMixin
+from napalm_proxmox.vm_snapshot_mixin import ProxmoxVMSnapshotMixin
 from napalm_proxmox.vm_provision_mixin import ProxmoxVMProvisionMixin
 from napalm_proxmox.routing_mixin import ProxmoxRoutingMixin
 from napalm_proxmox.system_mixin import ProxmoxSystemMixin
@@ -69,6 +71,8 @@ class ProxmoxDriver(
     ProxmoxLLDPMixin,
     ProxmoxConfigMixin,
     ProxmoxVMMixin,
+    ProxmoxVMContractMixin,
+    ProxmoxVMSnapshotMixin,
     ProxmoxVMProvisionMixin,
     ProxmoxRoutingMixin,
     ProxmoxSystemMixin,
@@ -78,6 +82,10 @@ class ProxmoxDriver(
 
     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
     PORT_SPECS = [
         PortSpec("https", 8006, weight=8.0),
     ]
@@ -137,9 +145,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,
@@ -170,31 +187,45 @@ class ProxmoxDriver(
             ) 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
-        try:
-            nodes = self._api.nodes.get()
-            # Match by hostname or IP
-            for n in nodes:
-                n_node = n.get("node", "")
-                if n_node:
-                    # First match: the node exists in the cluster
-                    self._node_name = n_node
+        members = [
+            s for s in (self._api.cluster.status.get() or []) if s.get("type") == "node"
+        ]
+        short_host = self.hostname.split(".")[0]
+        for pick in (
+            lambda s: s.get("local"),
+            lambda s: s.get("ip") == self.hostname,
+            lambda s: s.get("name") in (self.hostname, short_host),
+        ):
+            match = next((s for s in members if pick(s)), None)
+            if match:
+                self._node_name = match["name"]
                 return self._node_name
-        except Exception:
-            pass
 
-        # Fallback: use the configured hostname as node name
-        # (may not match the PVE node name — SSH-based methods will fail,
-        # but API methods that target a specific node name require correct
-        # resolution)
-        self._node_name = self.hostname
+        names = [n.get("node") for n in (self._api.nodes.get() or []) if n.get("node")]
+        if len(names) == 1:
+            self._node_name = names[0]
             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
@@ -206,11 +237,17 @@ class ProxmoxDriver(
             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
@@ -412,6 +449,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 +466,7 @@ class ProxmoxDriver(
             "os_version": os_version,
             "serial_number": serial,
             "interface_list": iface_list,
+            "running_kernel": running_kernel,
         }
 
     # ------------------------------------------------------------------ #

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.
 

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 {
-            vid: entry for vid, entry in result.items()
-            if entry.get("tagged") or entry.get("untagged")
-        }
+        # Deliberately unfiltered. This used to drop entries with no member
+        # ports, which hid every configured SDN VNet that no VM happened to be
+        # 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.

napalm_proxmox/system_mixin.py

@@ -244,7 +244,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,6 +258,22 @@ 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
@@ -266,6 +284,8 @@ class ProxmoxSystemMixin:
                 "description": "",
                 "size": size_bytes,
                 "source": "pve",
+                "source_package": source_package,
+                "source_version": source_version,
                 "upgrade_version": upgradable.get(name, ""),
             })
         return result

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

@@ -216,7 +216,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.
@@ -257,7 +257,7 @@ class ProxmoxVMMixin:
 
                 disks, onboot = self._get_vm_disk_and_boot(vmid, "qemu")
                 result.append({
-                    "vmid": vmid,
+                    "vmid": str(vmid),
                     "name": name,
                     "type": "vm",
                     "status": status,
@@ -301,7 +301,7 @@ class ProxmoxVMMixin:
 
                 disks, onboot = self._get_vm_disk_and_boot(vmid, "lxc")
                 result.append({
-                    "vmid": vmid,
+                    "vmid": str(vmid),
                     "name": name,
                     "type": "container",
                     "status": status,
@@ -321,7 +321,7 @@ class ProxmoxVMMixin:
         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+$")

napalm_proxmox/vm_provision_mixin.py

@@ -5,9 +5,10 @@ from __future__ import annotations
 import base64
 import hashlib
 import logging
+import re
 import time
 import yaml
-from typing import Any, Dict, List
+from typing import TYPE_CHECKING, Any, Dict, List
 from urllib.parse import quote
 
 from napalm_device_types.models import (
@@ -17,13 +18,95 @@ from napalm_device_types.models import (
     VMStatusDict,
 )
 
+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"}
+
+# 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."""
@@ -89,11 +172,9 @@ class ProxmoxVMProvisionMixin:
         before raising.
         """
         filename = image_url.rstrip("/").rsplit("/", 1)[-1]
-        url_hash = hashlib.sha256(image_url.encode()).hexdigest()[:12]
-        local_path = f"{_IMAGE_CACHE_DIR}/{url_hash}-{filename}"
+        local_path = f"{_IMAGE_CACHE_DIR}/{_url_key(image_url)}-{filename}"
 
-        algo, _, expected = (image_checksum or "").partition(":")
-        algo = (algo or "sha256").lower()
+        algo, expected = _split_checksum(image_checksum or "")
 
         max_attempts = 2
         for attempt in range(1, max_attempts + 1):
@@ -133,6 +214,125 @@ class ProxmoxVMProvisionMixin:
 
         raise AssertionError("unreachable")  # loop always returns or raises above
 
+    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._download_cloud_image(image_url, image_checksum, timeout=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').
 
@@ -196,6 +396,56 @@ class ProxmoxVMProvisionMixin:
             )
         return targets
 
+    def get_vm_cpu_types(self) -> list[VMCpuTypeDict]:
+        """List the CPU models a new VM may be given, judged against this node's CPU."""
+        return self._cpu_types_for(self._node_cpu_flags())
+
+    def _node_cpu_flags(self) -> set[str]:
+        status = self._node_api().status.get() or {}
+        return set(str((status.get("cpuinfo") or {}).get("flags", "")).split())
+
+    @staticmethod
+    def _cpu_types_for(node_flags: set[str]) -> list[VMCpuTypeDict]:
+        types: list[VMCpuTypeDict] = [
+            {
+                "name": name,
+                "description": description,
+                "features": list(flags),
+                "available": set(flags) <= node_flags,
+                "default": name == _DEFAULT_CPU_TYPE,
+            }
+            for name, flags, description in _CPU_MODELS
+        ]
+        types.append(
+            {
+                "name": "host",
+                "description": _HOST_CPU_DESCRIPTION,
+                "features": sorted(node_flags),
+                "available": True,
+                "default": False,
+            }
+        )
+        return types
+
+    def _resolve_cpu_type(self, cpu_type: str | None) -> str:
+        """The model to create the VM with. A model asked for by name is checked
+        against this node's CPU first: one it cannot run would fail only at VM
+        start, after the disk import, leaving a half-built VM behind."""
+        if cpu_type is None:
+            return _DEFAULT_CPU_TYPE
+        node_flags = self._node_cpu_flags()
+        offered = {t["name"]: t for t in self._cpu_types_for(node_flags)}
+        entry = offered.get(cpu_type)
+        if entry is None:
+            raise ValueError(f"Unknown CPU type {cpu_type!r}; choose one of {', '.join(offered)}")
+        if not entry["available"]:
+            missing = sorted(set(entry["features"]) - node_flags)
+            raise ValueError(
+                f"CPU type {cpu_type!r} needs {', '.join(missing)}, which the CPU of "
+                f"node {self._node_name} does not have"
+            )
+        return cpu_type
+
     def _wait_for_task(self, upid: str, timeout: int = 120) -> None:
         """
         Poll a Proxmox task until completion.
@@ -240,6 +490,7 @@ class ProxmoxVMProvisionMixin:
         ssh_public_keys: List[str] | None = None,
         disk_resize_gb: int | None = None,
         storage: str | None = None,
+        cpu_type: str | None = None,
         download_timeout: int = 300,
         timeout: int = 180,
     ) -> VMProvisionResultDict:
@@ -272,6 +523,7 @@ class ProxmoxVMProvisionMixin:
             disk_resize_gb: resize root disk to this size (None = no resize)
             storage: storage pool for the root disk (None = auto-detect first
                 enabled, node-available storage with content='images')
+            cpu_type: CPU model from get_vm_cpu_types() (None = x86-64-v2-AES)
             download_timeout: max seconds for the image download (skipped if cached)
             timeout: max seconds for the remaining provisioning steps
 
@@ -280,10 +532,13 @@ class ProxmoxVMProvisionMixin:
 
         Raises:
             RuntimeError: provisioning failure (download, import, config, timeout, etc.)
-            ValueError: invalid storage or configuration
+            ValueError: invalid storage or configuration, or a cpu_type that is
+                unknown or that this node's CPU cannot run (raised before
+                anything is created)
         """
         try:
             _logger.info(f"Creating VM '{name}' from image {image_url}")
+            cpu_model = self._resolve_cpu_type(cpu_type)
 
             # Step 1: Get next VMID
             next_vmid = self._api.cluster.nextid.get()
@@ -297,6 +552,7 @@ class ProxmoxVMProvisionMixin:
                 name=name,
                 memory=memory,
                 cores=cpu,
+                cpu=cpu_model,
                 ostype="l26",
                 scsihw="virtio-scsi-pci",
                 # Without this, Proxmox never attaches the virtio-serial
@@ -306,30 +562,9 @@ class ProxmoxVMProvisionMixin:
             )
 
             # Step 3: Download cloud image (cached) and import as root disk
-            local_path = self._download_cloud_image(
-                image_url, image_checksum, timeout=download_timeout
-            )
             image_storage = storage or self._find_default_image_storage()
-
-            _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",
+            self._attach_root_disk(
+                vmid, image_storage, image_url, image_checksum, download_timeout, timeout
             )
 
             # Step 4: Configure network interfaces (CPU/memory already set at shell creation)

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

@@ -25,7 +25,7 @@ classifiers = [
 requires-python = ">=3.9"
 dependencies = [
     "napalm>=5.0.0",
-    "napalm_device_types>=0.1.0",
+    "napalm_device_types>=2.0.0",
     "paramiko>=5.0.0",  # CVE-2026-44405; imported directly for SSH fallback (driver.py)
     "proxmoxer>=2.0.0",
     "netaddr>=0.9.0",

tests/conftest.py

@@ -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"}
 
@@ -135,6 +138,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 +166,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/test_connection.py

@@ -34,7 +34,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 +76,78 @@ 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_misc.py

@@ -263,17 +263,28 @@ class TestGetRouteTo:
 
 
 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

@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 import pytest
+from unittest.mock import patch
 
 from napalm_proxmox import utils
 
@@ -126,6 +127,11 @@ 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"
@@ -142,9 +148,7 @@ class TestGetMACAddressTable:
                 return self.BRIDGE_FDB
             return ""
 
-        driver._node_api().execute.post.side_effect = lambda command: {
-            "data": _exec(command)
-        }
+        with patch.object(driver, "_exec_ssh_command", side_effect=_exec):
             result = driver.get_mac_address_table()
         macs = {e["mac"] for e in result}
         assert "aa:bb:cc:dd:ee:01" in macs
@@ -155,9 +159,7 @@ class TestGetMACAddressTable:
                 return self.BRIDGE_FDB
             return ""
 
-        driver._node_api().execute.post.side_effect = lambda command: {
-            "data": _exec(command)
-        }
+        with patch.object(driver, "_exec_ssh_command", side_effect=_exec):
             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

@@ -22,19 +22,33 @@ class TestGetVlans:
         result = driver.get_vlans()
         assert "100000" in result
 
-    def test_bridge_vlan_show_parsing(self, driver):
-        # Simulate bridge vlan output
-        bridge_output = (
-            "vmbr0   1\n"
-            "        10\n"
-            "        20\n"
-            "eth0    1\n"
-        )
-        driver._node_api().execute.post.return_value = {"data": bridge_output}
+    def test_membership_derived_from_vm_configs(self, driver):
+        """Without OVS ports, VLAN membership comes from each VM's netN config.
+
+        This replaces a test for a `bridge vlan show` fallback that no longer
+        exists — it also asserted an "interfaces" key this method has never
+        produced, so it could not have passed against any version of the code.
+        """
+        node = driver._node_api()
+        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
-        entry = result.get("1", {})
-        assert "vmbr0" in entry.get("interfaces", [])
+        assert "vmbr0" in result["10"]["untagged"]
+
+    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

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