feat: VM snapshots and reboot_host through the API

get_vm_snapshots, create_vm_snapshot, delete_vm_snapshot and
rollback_vm_snapshot for VMs and containers, so netOrk's snapshot view
works on Proxmox as it does on VMware. Proxmox lists the live state as a
pseudo-snapshot named "current"; it is never reported or addressable.
Containers have no RAM state, so include_memory is ignored for them.

reboot_host() restarts the node with POST /nodes/{node}/status
command=reboot instead of /sbin/reboot over SSH.

CHANGELOG.md

@@ -7,6 +7,20 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Added
+- `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,
@@ -206,11 +223,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 +435,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 +452,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 {
+        # 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.

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
@@ -288,11 +308,7 @@ class ProxmoxSystemMixin:
         try:
             lldpd_path = self._exec_ssh_command("command -v lldpd 2>/dev/null").strip()
             if not lldpd_path:
-                warnings.append({
+                warnings.append({"code": "lldpd_not_installed"})
-                    "code": "lldpd_not_installed",
-                    "severity": "warning",
-                    "action": "install_lldpd",
-                })
         except Exception as exc:
             logger.debug("Failed to check for lldpd: %s", exc)
 
@@ -302,13 +318,6 @@ class ProxmoxSystemMixin:
             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],
@@ -322,12 +331,7 @@ class ProxmoxSystemMixin:
             sub = self._get_node_subscription()
             status = sub.get("status", "")
             if status in ("NotFound", "Invalid", "Expired"):
-                warnings.append({
+                warnings.append({"code": "no_subscription", "meta": {"status": status}})
-                    "code": "no_subscription",
-                    "severity": "warning",
-                    "action": None,
-                    "meta": {"status": status},
-                })
         except Exception as exc:
             logger.debug("Failed to check subscription status: %s", exc)
 

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

@@ -2,7 +2,8 @@
 
 from __future__ import annotations
 
-import io
+import base64
+import hashlib
 import logging
 import time
 import yaml
@@ -74,38 +75,63 @@ class ProxmoxVMProvisionMixin:
         """
         Download image_url to the node's image cache dir if not already present.
 
-        Returns the local path on the hypervisor node. Verifies image_checksum
+        Returns the local path on the hypervisor node. The cache filename is
-        (format "<algo>:<hex>", e.g. "sha256:abc123...") if given, re-downloading
+        prefixed with a hash of the *full* URL, not just its basename —
-        is left to the caller's next attempt if verification fails.
+        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}/{filename}"
+        url_hash = hashlib.sha256(image_url.encode()).hexdigest()[:12]
+        local_path = f"{_IMAGE_CACHE_DIR}/{url_hash}-{filename}"
 
-        exists = self._run_node_command(
+        algo, _, expected = (image_checksum or "").partition(":")
-            f"mkdir -p {_IMAGE_CACHE_DIR} && test -f {local_path} && echo EXISTS || echo MISSING",
+        algo = (algo or "sha256").lower()
-            timeout=30,
+
-        )
+        max_attempts = 2
-        if "EXISTS" not in exists:
+        for attempt in range(1, max_attempts + 1):
-            _logger.info(f"Downloading cloud image {image_url} -> {local_path}")
+            exists = self._run_node_command(
-            self._run_node_command(
+                f"mkdir -p {_IMAGE_CACHE_DIR} && test -f {local_path} "
-                f"wget -q -O {local_path}.tmp '{image_url}' && mv {local_path}.tmp {local_path}",
+                f"&& echo EXISTS || echo MISSING",
-                timeout=timeout,
+                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
 
-        if image_checksum:
-            algo, _, expected = image_checksum.partition(":")
-            algo = (algo or "sha256").lower()
             actual = self._run_node_command(
                 f"{algo}sum {local_path} | awk '{{print $1}}'", timeout=60
             )
-            if actual.lower() != expected.lower():
+            if actual.lower() == expected.lower():
-                # Remove the bad file so a retry re-downloads instead of reusing it.
+                return local_path
-                self._run_node_command(f"rm -f {local_path}", timeout=30)
+
+            # 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"
+            )
 
-        return local_path
+        raise AssertionError("unreachable")  # loop always returns or raises above
 
     def _find_default_image_storage(self) -> str:
         """Find a storage suitable for VM root disks (content includes 'images').
@@ -130,6 +156,25 @@ class ProxmoxVMProvisionMixin:
             "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] = []
@@ -254,6 +299,10 @@ class ProxmoxVMProvisionMixin:
                 cores=cpu,
                 ostype="l26",
                 scsihw="virtio-scsi-pci",
+                # Without this, Proxmox never attaches the virtio-serial
+                # channel the QEMU guest agent needs — get_vm_status's
+                # agent queries (below) would have nothing to talk to.
+                agent="1",
             )
 
             # Step 3: Download cloud image (cached) and import as root disk
@@ -294,8 +343,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"]:
@@ -336,27 +386,32 @@ class ProxmoxVMProvisionMixin:
             )
 
             filename = f"{vmid}-user-data.yaml"
-            _logger.debug(f"Uploading Cloud-Init snippet {filename} to {snippet_storage}")
+            _logger.debug(f"Writing Cloud-Init snippet {filename} to {snippet_storage}")
 
-            # Upload to snippet storage. Proxmox's upload endpoint expects the
+            # Proxmox's /storage/{s}/upload API only accepts content in
-            # "filename" parameter to BE the file (multipart), not a name
+            # {iso, vztmpl, import} — "snippets" is rejected outright
-            # string with separate content — proxmoxer only builds a
+            # ("does not have a value in the enumeration"). Snippets can only
-            # multipart request when the value is an io.IOBase instance,
+            # be written directly to the filesystem, so resolve the storage's
-            # otherwise it silently sends everything as a plain
+            # backing path and write the file over SSH instead.
-            # form-urlencoded POST, which real Proxmox rejects by dropping
+            storage_path = self._get_storage_path(snippet_storage)
-            # the connection (RemoteDisconnected, no HTTP response at all).
+            encoded = base64.b64encode(user_data_yaml.encode("utf-8")).decode("ascii")
-            file_obj = io.BytesIO(user_data_yaml.encode("utf-8"))
+            self._run_node_command(
-            file_obj.name = filename
+                f"mkdir -p {storage_path}/snippets && "
-            self._node_api().storage(snippet_storage).upload.post(
+                f"echo {encoded} | base64 -d > {storage_path}/snippets/{filename}",
-                content="snippets",
+                timeout=30,
-                filename=file_obj,
             )
 
             # 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}",
             }
@@ -447,11 +502,16 @@ class ProxmoxVMProvisionMixin:
                 _logger.debug(f"VM {vmid} stop failed (may already be stopped): {e}")
 
             # 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)
@@ -508,51 +568,49 @@ 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:
+                    # "lo" commonly comes first. Skip it and take the first real
-                        net0_iface = interfaces[0]  # Assumes net0 is first in list
+                    # NIC that has an IPv4 address.
-                        net0_mac = net0_iface.get("hardware-address", "")
+                    for iface in interfaces:
-                        ip_addresses = net0_iface.get("ip-addresses", [])
+                        name = iface.get("name", "")
-
+                        if not name or name == "lo":
-                        if ip_addresses:
+                            continue
-                            # Found IP
+                        for addr in iface.get("ip-addresses", []):
-                            ip_info = ip_addresses[0]
+                            if addr.get("ip-address-type") != "ipv4":
-                            ip_addr = ip_info.get("ip-address", "")
+                                continue
+                            ip_addr = addr.get("ip-address", "")
                             if ip_addr:
                                 _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:

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"}
 

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

tests/test_misc.py

@@ -263,18 +263,29 @@ 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"
-        "    SysName: sw01.example.com\n"
+        "-------------------------------------------------------------------------------\n"
-        "    PortID: ifname GigabitEthernet1/0/1\n"
+        "Interface:    eth0, via: LLDP, RID: 1, Time: 0 day, 00:11:22\n"
-        "  Interface: eth1\n"
+        "  Chassis:\n"
-        "    SysName: sw02.example.com\n"
+        "    SysName:      sw01.example.com\n"
-        "    PortID: ifname GigabitEthernet1/0/2\n"
+        "  Port:\n"
+        "    PortID:       ifname GigabitEthernet1/0/1\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()
+            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,10 +148,8 @@ 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()
-        }
-        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: {
+        with patch.object(driver, "_exec_ssh_command", side_effect=_exec):
-            "data": _exec(command)
+            result = driver.get_mac_address_table()
-        }
-        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):
+    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

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_provision_mixin.py

@@ -2,7 +2,7 @@
 
 from __future__ import annotations
 
-import io
+import base64
 
 import pytest
 from unittest.mock import MagicMock, patch
@@ -76,6 +76,7 @@ def test_create_vm_from_cloud_init_single_nic():
     # Mock API hierarchy
     mock_api = MagicMock()
     mock_api.cluster.nextid.get.return_value = 101
+    mock_api.storage.return_value.get.return_value = {"path": "/var/lib/vz"}
 
     mock_node = MagicMock()
     mock_node.storage.get.return_value = [
@@ -84,7 +85,9 @@ def test_create_vm_from_cloud_init_single_nic():
     ]
     mixin._api = mock_api
     mixin._node_api = MagicMock(return_value=mock_node)
-    mixin._download_cloud_image = MagicMock(return_value="/var/lib/vz/template/netork-images/debian-12.qcow2")
+    mixin._download_cloud_image = MagicMock(
+        return_value="/var/lib/vz/template/netork-images/debian-12.qcow2"
+    )
     mixin._run_node_command = MagicMock(return_value="")
 
     # Mock VM operations
@@ -103,11 +106,6 @@ def test_create_vm_from_cloud_init_single_nic():
     mock_task.status.get.return_value = {"status": "stopped", "exitstatus": "OK"}
     mock_node.tasks.return_value = mock_task
 
-    # Mock storage upload
-    mock_storage = MagicMock()
-    mock_storage.upload.post.return_value = {"filename": "snippets:snippets/101-user-data.yaml"}
-    mock_node.storage.return_value = mock_storage
-
     with patch("time.sleep"):
         result = mixin.create_vm_from_cloud_init(
             name="test-vm",
@@ -125,21 +123,27 @@ def test_create_vm_from_cloud_init_single_nic():
     assert mock_node.qemu.post.called
     mixin._download_cloud_image.assert_called_once()
 
+    # Regression: without agent="1" Proxmox never attaches the virtio-serial
+    # channel the QEMU guest agent needs, so get_vm_status(wait_for_ip=True)
+    # can never succeed no matter what's installed inside the guest.
+    assert mock_node.qemu.post.call_args[1]["agent"] == "1"
+
     # Verify NIC config was set correctly: net0 with tag=10, DHCP enabled
-    net_call_args = next(
+    net_call_args = next(c for c in mock_vm.config.post.call_args_list if "net0" in c[1])
-        c for c in mock_vm.config.post.call_args_list if "net0" in c[1]
-    )
     assert "tag=10" in net_call_args[1]["net0"]
     assert "vmbr0" in net_call_args[1]["net0"]
 
-    # Regression: the snippet must be uploaded as an actual file (io.IOBase),
+    # Regression: the snippet must be written directly to the filesystem via
-    # not a plain filename string with a separate "data" field — proxmoxer
+    # SSH, not uploaded via the /storage/upload API — real Proxmox rejects
-    # only builds a real multipart request for io.IOBase values, and real
+    # content='snippets' on that endpoint outright (see
-    # Proxmox drops the connection outright for anything else (see
+    # test_create_vm_writes_snippet_via_ssh_with_correct_content for the
-    # test_create_vm_uploads_snippet_as_file_object for the dedicated check).
+    # dedicated check).
-    upload_kwargs = mock_storage.upload.post.call_args[1]
+    write_cmd = next(
-    assert "data" not in upload_kwargs
+        c[0][0]
-    assert isinstance(upload_kwargs["filename"], io.IOBase)
+        for c in mixin._run_node_command.call_args_list
+        if "snippets/101-user-data.yaml" in c[0][0]
+    )
+    assert "/var/lib/vz/snippets" in write_cmd
 
 
 def test_create_vm_from_cloud_init_dual_nic_trunk():
@@ -150,6 +154,7 @@ def test_create_vm_from_cloud_init_dual_nic_trunk():
     # Mock API hierarchy
     mock_api = MagicMock()
     mock_api.cluster.nextid.get.return_value = 102
+    mock_api.storage.return_value.get.return_value = {"path": "/var/lib/vz"}
 
     mock_node = MagicMock()
     mock_node.storage.get.return_value = [
@@ -158,7 +163,9 @@ def test_create_vm_from_cloud_init_dual_nic_trunk():
     ]
     mixin._api = mock_api
     mixin._node_api = MagicMock(return_value=mock_node)
-    mixin._download_cloud_image = MagicMock(return_value="/var/lib/vz/template/netork-images/debian-12.qcow2")
+    mixin._download_cloud_image = MagicMock(
+        return_value="/var/lib/vz/template/netork-images/debian-12.qcow2"
+    )
     mixin._run_node_command = MagicMock(return_value="")
 
     # Mock VM operations
@@ -177,11 +184,6 @@ def test_create_vm_from_cloud_init_dual_nic_trunk():
     mock_task.status.get.return_value = {"status": "stopped", "exitstatus": "OK"}
     mock_node.tasks.return_value = mock_task
 
-    # Mock storage upload
-    mock_storage = MagicMock()
-    mock_storage.upload.post.return_value = {"filename": "snippets:snippets/102-user-data.yaml"}
-    mock_node.storage.return_value = mock_storage
-
     with patch("time.sleep"):
         result = mixin.create_vm_from_cloud_init(
             name="wireshark-sat-1",
@@ -190,7 +192,11 @@ def test_create_vm_from_cloud_init_dual_nic_trunk():
             memory=4096,
             nics=[
                 {"bridge": "vmbr0", "vlan_tag": 10},  # net0: mgmt with DHCP
-                {"bridge": "vmbr1", "trunk_vlan_tags": [20, 30], "dhcp": False},  # net1: trunk, no DHCP
+                {
+                    "bridge": "vmbr1",
+                    "trunk_vlan_tags": [20, 30],
+                    "dhcp": False,
+                },  # net1: trunk, no DHCP
             ],
             cloud_init_config={"hostname": "sat-1", "runcmd": ["custom cmd"]},
             timeout=120,
@@ -200,9 +206,7 @@ def test_create_vm_from_cloud_init_dual_nic_trunk():
     assert result["name"] == "wireshark-sat-1"
 
     # Verify both NICs configured
-    net_call_args = next(
+    net_call_args = next(c for c in mock_vm.config.post.call_args_list if "net0" in c[1])
-        c for c in mock_vm.config.post.call_args_list if "net0" in c[1]
-    )
     assert "net1" in net_call_args[1]
     assert "tag=10" in net_call_args[1]["net0"]
     assert "trunks=20;30" in net_call_args[1]["net1"]
@@ -232,7 +236,9 @@ def test_create_vm_missing_snippet_storage():
         {"storage": "local-lvm", "type": "lvmthin", "content": "images,rootdir", "enabled": 1}
     ]
     mixin._node_api = MagicMock(return_value=mock_node)
-    mixin._download_cloud_image = MagicMock(return_value="/var/lib/vz/template/netork-images/debian-12.qcow2")
+    mixin._download_cloud_image = MagicMock(
+        return_value="/var/lib/vz/template/netork-images/debian-12.qcow2"
+    )
     mixin._run_node_command = MagicMock(return_value="")
     mock_vm = MagicMock()
     mock_node.qemu.return_value = mock_vm
@@ -265,6 +271,7 @@ def test_create_vm_with_disk_resize():
     # Mock API hierarchy
     mock_api = MagicMock()
     mock_api.cluster.nextid.get.return_value = 103
+    mock_api.storage.return_value.get.return_value = {"path": "/var/lib/vz"}
 
     mock_node = MagicMock()
     mock_node.storage.get.return_value = [
@@ -273,7 +280,9 @@ def test_create_vm_with_disk_resize():
     ]
     mixin._api = mock_api
     mixin._node_api = MagicMock(return_value=mock_node)
-    mixin._download_cloud_image = MagicMock(return_value="/var/lib/vz/template/netork-images/debian-12.qcow2")
+    mixin._download_cloud_image = MagicMock(
+        return_value="/var/lib/vz/template/netork-images/debian-12.qcow2"
+    )
     mixin._run_node_command = MagicMock(return_value="")
 
     # Mock VM operations
@@ -293,11 +302,6 @@ def test_create_vm_with_disk_resize():
     mock_task.status.get.return_value = {"status": "stopped", "exitstatus": "OK"}
     mock_node.tasks.return_value = mock_task
 
-    # Mock storage upload
-    mock_storage = MagicMock()
-    mock_storage.upload.post.return_value = {"filename": "snippets:snippets/103-user-data.yaml"}
-    mock_node.storage.return_value = mock_storage
-
     with patch("time.sleep"):
         result = mixin.create_vm_from_cloud_init(
             name="big-vm",
@@ -331,9 +335,12 @@ def test_get_vm_status_with_wait_for_ip():
     mock_node.qemu.return_value = mock_vm
     mock_vm.config.get.return_value = {"net0": "virtio,bridge=vmbr0,tag=10"}
 
-    # Mock guest-agent: first no IP, then with IP
+    # The real Proxmox REST path is "network-get-interfaces" (hyphens),
-    mock_agent = MagicMock()
+    # reached via agent(...) as a callable resource, not dotted attribute
-    mock_agent.network_get_interfaces.get.side_effect = [
+    # access — mock that exact call shape.
+    mock_agent_call = MagicMock()
+    mock_vm.agent.return_value = mock_agent_call
+    mock_agent_call.get.side_effect = [
         {"result": [{"name": "eth0", "hardware-address": "aa:bb:cc:dd:ee:00"}]},
         {
             "result": [
@@ -345,16 +352,51 @@ def test_get_vm_status_with_wait_for_ip():
             ]
         },
     ]
-    mock_vm.agent = mock_agent
 
     with patch("time.sleep"):
         result = mixin.get_vm_status("101", wait_for_ip=True, timeout=30, poll_interval=1)
 
+    mock_vm.agent.assert_called_with("network-get-interfaces")
     assert result["status"] == "running"
     assert result["ip_address"] == "10.0.0.100"
     assert result["mac_address"] == "aa:bb:cc:dd:ee:00"
 
 
+def test_get_vm_status_skips_loopback_interface():
+    """The guest agent commonly reports "lo" before the real NIC — it must
+    be skipped rather than mistaken for the VM's address."""
+    mixin = ProxmoxVMProvisionMixin()
+
+    mock_node = MagicMock()
+    mixin._node_api = MagicMock(return_value=mock_node)
+
+    mock_vm = MagicMock()
+    mock_node.qemu.return_value = mock_vm
+    mock_vm.config.get.return_value = {"net0": "virtio,bridge=vmbr0"}
+
+    mock_agent_call = MagicMock()
+    mock_vm.agent.return_value = mock_agent_call
+    mock_agent_call.get.return_value = {
+        "result": [
+            {
+                "name": "lo",
+                "hardware-address": "00:00:00:00:00:00",
+                "ip-addresses": [{"ip-address": "127.0.0.1", "ip-address-type": "ipv4"}],
+            },
+            {
+                "name": "eth0",
+                "hardware-address": "aa:bb:cc:dd:ee:00",
+                "ip-addresses": [{"ip-address": "10.0.0.101", "ip-address-type": "ipv4"}],
+            },
+        ]
+    }
+
+    with patch("time.sleep"):
+        result = mixin.get_vm_status("101", wait_for_ip=True, timeout=30, poll_interval=1)
+
+    assert result["ip_address"] == "10.0.0.101"
+
+
 def test_get_vm_status_timeout_waiting_for_ip():
     """get_vm_status raises RuntimeError if IP acquisition times out."""
     mixin = ProxmoxVMProvisionMixin()
@@ -368,11 +410,11 @@ def test_get_vm_status_timeout_waiting_for_ip():
     mock_vm.config.get.return_value = {"net0": "virtio,bridge=vmbr0"}
 
     # Mock guest-agent that never returns IP
-    mock_agent = MagicMock()
+    mock_agent_call = MagicMock()
-    mock_agent.network_get_interfaces.get.return_value = {
+    mock_vm.agent.return_value = mock_agent_call
+    mock_agent_call.get.return_value = {
         "result": [{"name": "eth0", "hardware-address": "aa:bb:cc:dd:ee:00"}]
     }
-    mock_vm.agent = mock_agent
 
     with pytest.raises(RuntimeError, match="timeout|IP"):
         with patch("time.sleep"):
@@ -409,6 +451,11 @@ def test_destroy_vm_success():
         mixin.destroy_vm("101", remove_disk=True, timeout=60)
 
     assert mock_vm.delete.called
+    # Proxmox's API parameter is hyphenated; passing the underscore form
+    # (a Python-identifier-friendly typo) gets silently rejected by Proxmox
+    # with a 400 "property is not defined in schema" instead of deleting.
+    _, delete_kwargs = mock_vm.delete.call_args
+    assert delete_kwargs == {"purge": 1, "destroy-unreferenced-disks": 1}
 
 
 def test_destroy_vm_already_stopped():
@@ -480,9 +527,7 @@ def test_get_network_targets_linux_bridge_vlan_aware_flag():
 def test_get_network_targets_ovs_bridge_always_vlan_aware():
     """An OVS bridge is always vlan_aware, regardless of bridge_vlan_aware."""
     mixin = ProxmoxVMProvisionMixin()
-    mixin._get_node_network = MagicMock(
+    mixin._get_node_network = MagicMock(return_value=[{"iface": "vmbr1", "type": "OVSBridge"}])
-        return_value=[{"iface": "vmbr1", "type": "OVSBridge"}]
-    )
     mixin._get_sdn_vnets = MagicMock(return_value=[])
 
     targets = mixin.get_network_targets()
@@ -526,9 +571,7 @@ def test_get_network_targets_vnet_without_tag_has_none_fixed_vlan_tag():
     """A vnet with no tag (e.g. VXLAN/EVPN zone) reports fixed_vlan_tag=None."""
     mixin = ProxmoxVMProvisionMixin()
     mixin._get_node_network = MagicMock(return_value=[])
-    mixin._get_sdn_vnets = MagicMock(
+    mixin._get_sdn_vnets = MagicMock(return_value=[{"vnet": "vnet2", "zone": "zone-vxlan"}])
-        return_value=[{"vnet": "vnet2", "zone": "zone-vxlan"}]
-    )
 
     targets = mixin.get_network_targets()
 
@@ -635,10 +678,73 @@ def test_download_cloud_image_verifies_matching_checksum():
     assert path.endswith("debian-12-genericcloud-amd64.qcow2")
 
 
-def test_download_cloud_image_checksum_mismatch_raises_and_removes_file():
+def test_download_cloud_image_cache_key_differs_for_same_basename_different_url():
+    """Regression: Ubuntu's per-build URLs change daily under a stable
+    basename (ubuntu-26.04-server-cloudimg-amd64.img) — a cache key derived
+    from just the basename let a stale build from a previous day collide
+    with today's cache lookup, skip re-downloading, and fail checksum
+    verification against today's expected hash. Gitea issue filed for this."""
+    mixin = ProxmoxVMProvisionMixin()
+    mixin._run_node_command = MagicMock(return_value="MISSING")
+
+    url_a = (
+        "https://cloud-images.ubuntu.com/releases/server/releases/resolute/"
+        "release-20260713/ubuntu-26.04-server-cloudimg-amd64.img"
+    )
+    url_b = (
+        "https://cloud-images.ubuntu.com/releases/server/releases/resolute/"
+        "release-20260714/ubuntu-26.04-server-cloudimg-amd64.img"
+    )
+
+    path_a = mixin._download_cloud_image(url_a, None, timeout=300)
+    path_b = mixin._download_cloud_image(url_b, None, timeout=300)
+
+    assert path_a != path_b
+    assert path_a.endswith("ubuntu-26.04-server-cloudimg-amd64.img")
+    assert path_b.endswith("ubuntu-26.04-server-cloudimg-amd64.img")
+
+
+def test_download_cloud_image_checksum_mismatch_retries_and_succeeds():
+    """The reported bug's actual scenario: a stale same-named cache entry
+    fails checksum, gets removed, and the retry re-downloads + verifies
+    successfully instead of failing the whole provisioning job outright."""
     mixin = ProxmoxVMProvisionMixin()
     mixin._run_node_command = MagicMock(
-        side_effect=["EXISTS", "wrong-checksum", ""]  # existence, checksum, rm
+        side_effect=[
+            "EXISTS",  # attempt 1: stale file already present
+            "wrong-checksum",  # attempt 1: checksum against stale content
+            "",  # rm -f
+            "MISSING",  # attempt 2: file gone, re-download
+            "",  # wget
+            "abc123",  # attempt 2: checksum against fresh content
+        ]
+    )
+
+    path = mixin._download_cloud_image(
+        "https://cloud-images.ubuntu.com/releases/server/releases/resolute/"
+        "release-20260713/ubuntu-26.04-server-cloudimg-amd64.img",
+        "sha256:abc123",
+        timeout=300,
+    )
+
+    assert path.endswith("ubuntu-26.04-server-cloudimg-amd64.img")
+    commands = [c[0][0] for c in mixin._run_node_command.call_args_list]
+    assert sum(1 for cmd in commands if "wget" in cmd) == 1
+    assert sum(1 for cmd in commands if cmd.startswith("rm -f")) == 1
+
+
+def test_download_cloud_image_checksum_mismatch_persists_raises_after_retry():
+    mixin = ProxmoxVMProvisionMixin()
+    mixin._run_node_command = MagicMock(
+        side_effect=[
+            "EXISTS",
+            "wrong-checksum",
+            "",  # rm -f (attempt 1)
+            "MISSING",
+            "",  # wget
+            "still-wrong",
+            "",  # rm -f (attempt 2)
+        ]
     )
 
     with pytest.raises(RuntimeError, match="Checksum mismatch"):
@@ -649,7 +755,7 @@ def test_download_cloud_image_checksum_mismatch_raises_and_removes_file():
         )
 
     commands = [c[0][0] for c in mixin._run_node_command.call_args_list]
-    assert any(cmd.startswith("rm -f") for cmd in commands)
+    assert sum(1 for cmd in commands if cmd.startswith("rm -f")) == 2
 
 
 # ---------------------------------------------------------------------------
@@ -722,6 +828,42 @@ def test_find_default_image_storage_excludes_storage_restricted_to_other_nodes()
     assert storage == "local-zfs"
 
 
+# ---------------------------------------------------------------------------
+# _get_storage_path
+# ---------------------------------------------------------------------------
+
+
+def test_get_storage_path_returns_path_from_cluster_config():
+    mixin = ProxmoxVMProvisionMixin()
+    mixin._api = MagicMock()
+    mixin._api.storage.return_value.get.return_value = {
+        "storage": "local",
+        "type": "dir",
+        "path": "/var/lib/vz",
+    }
+
+    path = mixin._get_storage_path("local")
+
+    assert path == "/var/lib/vz"
+    mixin._api.storage.assert_called_with("local")
+
+
+def test_get_storage_path_raises_when_storage_has_no_path():
+    """A storage type without a filesystem path (e.g. lvmthin, zfspool) can't
+    back content='snippets' at all — Proxmox itself only allows that content
+    type on dir/nfs/cifs/cephfs storages, so this should never actually be
+    reached for a real snippet storage, but must fail clearly if it is."""
+    mixin = ProxmoxVMProvisionMixin()
+    mixin._api = MagicMock()
+    mixin._api.storage.return_value.get.return_value = {
+        "storage": "local-lvm",
+        "type": "lvmthin",
+    }
+
+    with pytest.raises(ValueError, match="path"):
+        mixin._get_storage_path("local-lvm")
+
+
 def test_create_vm_from_cloud_init_with_real_world_storage_shape():
     """Regression: real Proxmox servers omit "enabled" for never-toggled storages
     (observed live: local-lvm/local-zfs/fast-zfs all had content=images but no
@@ -733,6 +875,7 @@ def test_create_vm_from_cloud_init_with_real_world_storage_shape():
 
     mock_api = MagicMock()
     mock_api.cluster.nextid.get.return_value = 104
+    mock_api.storage.return_value.get.return_value = {"path": "/var/lib/vz"}
 
     mock_node = MagicMock()
     mock_node.storage.get.return_value = [
@@ -743,7 +886,9 @@ def test_create_vm_from_cloud_init_with_real_world_storage_shape():
     ]
     mixin._api = mock_api
     mixin._node_api = MagicMock(return_value=mock_node)
-    mixin._download_cloud_image = MagicMock(return_value="/var/lib/vz/template/netork-images/debian-12.qcow2")
+    mixin._download_cloud_image = MagicMock(
+        return_value="/var/lib/vz/template/netork-images/debian-12.qcow2"
+    )
     mixin._run_node_command = MagicMock(return_value="")
 
     mock_vm = MagicMock()
@@ -760,10 +905,6 @@ def test_create_vm_from_cloud_init_with_real_world_storage_shape():
     mock_task.status.get.return_value = {"status": "stopped", "exitstatus": "OK"}
     mock_node.tasks.return_value = mock_task
 
-    mock_storage = MagicMock()
-    mock_storage.upload.post.return_value = {"filename": "snippets:snippets/104-user-data.yaml"}
-    mock_node.storage.return_value = mock_storage
-
     with patch("time.sleep"):
         result = mixin.create_vm_from_cloud_init(
             name="real-shape-vm",
@@ -786,6 +927,7 @@ def test_create_vm_from_cloud_init_explicit_storage_skips_auto_detect():
 
     mock_api = MagicMock()
     mock_api.cluster.nextid.get.return_value = 105
+    mock_api.storage.return_value.get.return_value = {"path": "/var/lib/vz"}
 
     mock_node = MagicMock()
     mock_node.storage.get.return_value = [
@@ -795,8 +937,12 @@ def test_create_vm_from_cloud_init_explicit_storage_skips_auto_detect():
     ]
     mixin._api = mock_api
     mixin._node_api = MagicMock(return_value=mock_node)
-    mixin._download_cloud_image = MagicMock(return_value="/var/lib/vz/template/netork-images/debian-12.qcow2")
+    mixin._download_cloud_image = MagicMock(
-    mixin._find_default_image_storage = MagicMock(side_effect=AssertionError("should not be called"))
+        return_value="/var/lib/vz/template/netork-images/debian-12.qcow2"
+    )
+    mixin._find_default_image_storage = MagicMock(
+        side_effect=AssertionError("should not be called")
+    )
     mixin._run_node_command = MagicMock(return_value="")
 
     mock_vm = MagicMock()
@@ -813,10 +959,6 @@ def test_create_vm_from_cloud_init_explicit_storage_skips_auto_detect():
     mock_task.status.get.return_value = {"status": "stopped", "exitstatus": "OK"}
     mock_node.tasks.return_value = mock_task
 
-    mock_storage = MagicMock()
-    mock_storage.upload.post.return_value = {"filename": "snippets:snippets/105-user-data.yaml"}
-    mock_node.storage.return_value = mock_storage
-
     with patch("time.sleep"):
         result = mixin.create_vm_from_cloud_init(
             name="explicit-storage-vm",
@@ -901,21 +1043,22 @@ def test_get_image_storages_excludes_storage_restricted_to_other_nodes():
 # ---------------------------------------------------------------------------
 
 
-def test_create_vm_uploads_snippet_as_file_object_with_correct_content():
+def test_create_vm_writes_snippet_via_ssh_with_correct_content():
-    """Regression: real Proxmox's /storage/{s}/upload endpoint expects the
+    """Regression: real Proxmox's /storage/{s}/upload endpoint rejects
-    "filename" parameter to be the file itself (multipart). proxmoxer only
+    content='snippets' outright ("value 'snippets' does not have a value in
-    builds a multipart request when the value is an io.IOBase instance —
+    the enumeration 'iso, vztmpl, import'") — that endpoint only handles
-    passing a plain string (with a separate, nonexistent "data" field, as the
+    ISOs, container templates, and imports. Snippets can only be written
-    old code did) makes proxmoxer send a normal form-urlencoded POST instead,
+    directly to the storage's filesystem path, so this must go through SSH
-    which real Proxmox responds to by closing the connection outright
+    (_run_node_command), not the upload API.
-    (observed live: requests.exceptions.ConnectionError /
+    (observed live: 400 Bad Request from Proxmox, right after the earlier
-    RemoteDisconnected('Remote end closed connection without response'),
+    multipart-upload fix had already gotten past a prior RemoteDisconnected
-    after the VM shell and disk import had already succeeded)."""
+    bug at the same step — two distinct real-world failures at this line)."""
     mixin = ProxmoxVMProvisionMixin()
     mixin._node_name = "pve1"
 
     mock_api = MagicMock()
     mock_api.cluster.nextid.get.return_value = 106
+    mock_api.storage.return_value.get.return_value = {"path": "/mnt/pve/snippet-nfs"}
 
     mock_node = MagicMock()
     mock_node.storage.get.return_value = [
@@ -924,7 +1067,9 @@ def test_create_vm_uploads_snippet_as_file_object_with_correct_content():
     ]
     mixin._api = mock_api
     mixin._node_api = MagicMock(return_value=mock_node)
-    mixin._download_cloud_image = MagicMock(return_value="/var/lib/vz/template/netork-images/debian-12.qcow2")
+    mixin._download_cloud_image = MagicMock(
+        return_value="/var/lib/vz/template/netork-images/debian-12.qcow2"
+    )
     mixin._run_node_command = MagicMock(return_value="")
 
     mock_vm = MagicMock()
@@ -941,28 +1086,191 @@ def test_create_vm_uploads_snippet_as_file_object_with_correct_content():
     mock_task.status.get.return_value = {"status": "stopped", "exitstatus": "OK"}
     mock_node.tasks.return_value = mock_task
 
-    mock_storage = MagicMock()
-    mock_storage.upload.post.return_value = {"filename": "local:snippets/106-user-data.yaml"}
-    mock_node.storage.return_value = mock_storage
-
     with patch("time.sleep"):
         mixin.create_vm_from_cloud_init(
-            name="upload-shape-vm",
+            name="write-shape-vm",
             image_url="https://cloud.debian.org/images/cloud/bookworm/latest/debian-12-genericcloud-amd64.qcow2",
             cpu=2,
             memory=2048,
             nics=[{"bridge": "vmbr0"}],
-            cloud_init_config={"hostname": "upload-shape-vm", "chpasswd": {"expire": False}},
+            cloud_init_config={"hostname": "write-shape-vm", "chpasswd": {"expire": False}},
         )
 
-    upload_call = mock_storage.upload.post.call_args
+    # No call to the upload API at all — snippets aren't a valid content
-    assert upload_call[1]["content"] == "snippets"
+    # type there.
-    assert "data" not in upload_call[1]
+    mock_node.storage.assert_not_called()
 
-    file_obj = upload_call[1]["filename"]
+    write_cmd = next(
-    assert isinstance(file_obj, io.IOBase)
+        c[0][0]
-    assert file_obj.name == "106-user-data.yaml"
+        for c in mixin._run_node_command.call_args_list
-    file_obj.seek(0)
+        if "snippets/106-user-data.yaml" in c[0][0]
-    content = file_obj.read().decode("utf-8")
+    )
+    assert "mkdir -p /mnt/pve/snippet-nfs/snippets" in write_cmd
+    assert "/mnt/pve/snippet-nfs/snippets/106-user-data.yaml" in write_cmd
+
+    encoded = write_cmd.split("echo ", 1)[1].split(" | base64 -d")[0]
+    content = base64.b64decode(encoded).decode("utf-8")
     assert content.startswith("#cloud-config\n")
-    assert "hostname: upload-shape-vm" in content
+    assert "hostname: write-shape-vm" in content
+
+
+# ---------------------------------------------------------------------------
+# Explicit NIC MAC address
+# ---------------------------------------------------------------------------
+
+
+def test_create_vm_from_cloud_init_with_explicit_mac():
+    """When nics[i]['mac'] is set, it's pinned via virtio=<mac>,bridge=...
+    (needed so a caller can create a matching DHCP reservation before the
+    VM even boots)."""
+    mixin = ProxmoxVMProvisionMixin()
+    mixin._node_name = "pve1"
+
+    mock_api = MagicMock()
+    mock_api.cluster.nextid.get.return_value = 107
+    mock_api.storage.return_value.get.return_value = {"path": "/var/lib/vz"}
+
+    mock_node = MagicMock()
+    mock_node.storage.get.return_value = [
+        {"storage": "local-lvm", "type": "lvmthin", "content": "images,rootdir", "enabled": 1},
+        {"storage": "snippets", "type": "dir", "content": "snippets", "enabled": 1},
+    ]
+    mixin._api = mock_api
+    mixin._node_api = MagicMock(return_value=mock_node)
+    mixin._download_cloud_image = MagicMock(
+        return_value="/var/lib/vz/template/netork-images/debian-12.qcow2"
+    )
+    mixin._run_node_command = MagicMock(return_value="")
+
+    mock_vm = MagicMock()
+    mock_node.qemu.return_value = mock_vm
+    mock_node.qemu.post.return_value = None
+    mock_vm.config.post.return_value = None
+    mock_vm.config.get.return_value = {
+        "unused0": "local-lvm:vm-107-disk-0",
+        "scsi0": "local-lvm:vm-107-disk-0",
+    }
+    mock_vm.status.start.post.return_value = "UPID:pve1:133:start"
+
+    mock_task = MagicMock()
+    mock_task.status.get.return_value = {"status": "stopped", "exitstatus": "OK"}
+    mock_node.tasks.return_value = mock_task
+
+    with patch("time.sleep"):
+        mixin.create_vm_from_cloud_init(
+            name="pinned-mac-vm",
+            image_url="https://cloud.debian.org/images/cloud/bookworm/latest/debian-12-genericcloud-amd64.qcow2",
+            cpu=2,
+            memory=2048,
+            nics=[{"bridge": "vmbr0", "vlan_tag": 10, "mac": "02:aa:bb:cc:dd:ee"}],
+            cloud_init_config={"hostname": "pinned-mac-vm"},
+        )
+
+    net_call_args = next(c for c in mock_vm.config.post.call_args_list if "net0" in c[1])
+    assert net_call_args[1]["net0"] == "virtio=02:aa:bb:cc:dd:ee,bridge=vmbr0,tag=10"
+
+
+def test_create_vm_from_cloud_init_without_mac_uses_bare_virtio():
+    """Regression: omitting nics[i]['mac'] must still produce the original
+    bare 'virtio,bridge=...' string (auto-generated MAC), not 'virtio=None,...'."""
+    mixin = ProxmoxVMProvisionMixin()
+    mixin._node_name = "pve1"
+
+    mock_api = MagicMock()
+    mock_api.cluster.nextid.get.return_value = 108
+    mock_api.storage.return_value.get.return_value = {"path": "/var/lib/vz"}
+
+    mock_node = MagicMock()
+    mock_node.storage.get.return_value = [
+        {"storage": "local-lvm", "type": "lvmthin", "content": "images,rootdir", "enabled": 1},
+        {"storage": "snippets", "type": "dir", "content": "snippets", "enabled": 1},
+    ]
+    mixin._api = mock_api
+    mixin._node_api = MagicMock(return_value=mock_node)
+    mixin._download_cloud_image = MagicMock(
+        return_value="/var/lib/vz/template/netork-images/debian-12.qcow2"
+    )
+    mixin._run_node_command = MagicMock(return_value="")
+
+    mock_vm = MagicMock()
+    mock_node.qemu.return_value = mock_vm
+    mock_node.qemu.post.return_value = None
+    mock_vm.config.post.return_value = None
+    mock_vm.config.get.return_value = {
+        "unused0": "local-lvm:vm-108-disk-0",
+        "scsi0": "local-lvm:vm-108-disk-0",
+    }
+    mock_vm.status.start.post.return_value = "UPID:pve1:134:start"
+
+    mock_task = MagicMock()
+    mock_task.status.get.return_value = {"status": "stopped", "exitstatus": "OK"}
+    mock_node.tasks.return_value = mock_task
+
+    with patch("time.sleep"):
+        mixin.create_vm_from_cloud_init(
+            name="auto-mac-vm",
+            image_url="https://cloud.debian.org/images/cloud/bookworm/latest/debian-12-genericcloud-amd64.qcow2",
+            cpu=2,
+            memory=2048,
+            nics=[{"bridge": "vmbr0"}],
+            cloud_init_config={"hostname": "auto-mac-vm"},
+        )
+
+    net_call_args = next(c for c in mock_vm.config.post.call_args_list if "net0" in c[1])
+    assert net_call_args[1]["net0"] == "virtio,bridge=vmbr0"
+
+
+def test_create_vm_cloudinit_drive_uses_image_storage_not_snippet_storage():
+    """Regression: real Proxmox fails at VM *start* time with "storage 'X'
+    does not support content-type 'images'" if ide2 (the cloud-init drive)
+    points at the snippets storage — a cloud-init drive is a disk image and
+    needs content='images', same as the root disk. Only cicustom (the
+    snippet file reference) should use the snippets storage. Found live: a
+    real deployment had 'local' (snippets-only) and 'local-zfs' (images) as
+    two distinct storages, and ide2 was wrongly set to the former."""
+    mixin = ProxmoxVMProvisionMixin()
+    mixin._node_name = "pve1"
+
+    mock_api = MagicMock()
+    mock_api.cluster.nextid.get.return_value = 109
+    mock_api.storage.return_value.get.return_value = {"path": "/var/lib/vz"}
+
+    mock_node = MagicMock()
+    mock_node.storage.get.return_value = [
+        {"storage": "local-zfs", "type": "zfspool", "content": "images,rootdir", "enabled": 1},
+        {"storage": "local", "type": "dir", "content": "iso,snippets,backup,vztmpl", "enabled": 1},
+    ]
+    mixin._api = mock_api
+    mixin._node_api = MagicMock(return_value=mock_node)
+    mixin._download_cloud_image = MagicMock(
+        return_value="/var/lib/vz/template/netork-images/debian-12.qcow2"
+    )
+    mixin._run_node_command = MagicMock(return_value="")
+
+    mock_vm = MagicMock()
+    mock_node.qemu.return_value = mock_vm
+    mock_node.qemu.post.return_value = None
+    mock_vm.config.post.return_value = None
+    mock_vm.config.get.return_value = {
+        "unused0": "local-zfs:vm-109-disk-0",
+        "scsi0": "local-zfs:vm-109-disk-0",
+    }
+    mock_vm.status.start.post.return_value = "UPID:pve1:135:start"
+
+    mock_task = MagicMock()
+    mock_task.status.get.return_value = {"status": "stopped", "exitstatus": "OK"}
+    mock_node.tasks.return_value = mock_task
+
+    with patch("time.sleep"):
+        mixin.create_vm_from_cloud_init(
+            name="cloudinit-storage-vm",
+            image_url="https://cloud.debian.org/images/cloud/bookworm/latest/debian-12-genericcloud-amd64.qcow2",
+            cpu=2,
+            memory=2048,
+            nics=[{"bridge": "vmbr0"}],
+            cloud_init_config={"hostname": "cloudinit-storage-vm"},
+        )
+
+    cloud_init_call = next(c for c in mock_vm.config.post.call_args_list if "ide2" in c[1])
+    assert cloud_init_call[1]["ide2"] == "local-zfs:cloudinit"
+    assert cloud_init_call[1]["cicustom"].startswith("user=local:snippets/")

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