Merge fix/cloudinit-drive-image-storage: ide2 needs images storage

Proxmox's cloud-init drive is a disk image and requires a storage with
content='images' — the same requirement as the root disk — not the
snippets storage. These are commonly different storages (e.g. 'local'
with content=snippets-only, 'local-zfs' with content=images), and real
Proxmox now creates the VM fine but fails at *start* time with "storage
'X' does not support content-type 'images'" once it tries to generate
the cloud-init ISO.

Found live: a real deployment created the VM successfully, and only
failed when the user started it manually on the Proxmox side.

napalm_proxmox/vm_provision_mixin.py

@@ -2,20 +2,174 @@
 
 from __future__ import annotations
 
+import base64
 import logging
 import time
 import yaml
 from typing import Any, Dict, List
 from urllib.parse import quote
 
-from napalm_device_types.models import NetworkTargetDict, VMProvisionResultDict, VMStatusDict
+from napalm_device_types.models import (
+    NetworkTargetDict,
+    StorageTargetDict,
+    VMProvisionResultDict,
+    VMStatusDict,
+)
 
 _logger = logging.getLogger(__name__)
 
+# Downloaded cloud images are cached here on the hypervisor node, keyed by
+# filename, so provisioning multiple VMs from the same image only pays the
+# download cost once.
+_IMAGE_CACHE_DIR = "/var/lib/vz/template/netork-images"
+
 
 class ProxmoxVMProvisionMixin:
     """Mixin to add VM provisioning to ProxmoxDriver."""
 
+    def _run_node_command(self, command: str, timeout: int) -> str:
+        """
+        Execute a shell command on the Proxmox node via SSH, raising on failure.
+
+        Unlike ``_exec_ssh_command`` (best-effort, fixed timeout, swallows
+        errors), this is for critical provisioning steps — image download,
+        disk import — where a non-zero exit or a caller-specific timeout must
+        surface as a hard failure rather than an empty string.
+        """
+        import paramiko
+
+        if self._ssh_client is None:
+            ssh_user = self._ssh_username or self.username
+            ssh_pass = self._ssh_password or self.password
+            ssh_pkey = None
+            if self._ssh_key and not ssh_pass:
+                from io import StringIO as _StringIO
+
+                ssh_pkey = paramiko.RSAKey.from_private_key(_StringIO(self._ssh_key))
+            self._ssh_client = paramiko.SSHClient()
+            self._ssh_client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
+            connect_kwargs: Dict[str, Any] = {
+                "hostname": self.hostname,
+                "port": 22,
+                "username": ssh_user,
+                "timeout": self.timeout,
+            }
+            if ssh_pkey:
+                connect_kwargs["pkey"] = ssh_pkey
+            else:
+                connect_kwargs["password"] = ssh_pass
+            self._ssh_client.connect(**connect_kwargs)
+
+        _, stdout, stderr = self._ssh_client.exec_command(command, timeout=timeout)
+        exit_status = stdout.channel.recv_exit_status()
+        out = stdout.read().decode().strip()
+        err = stderr.read().decode().strip()
+        if exit_status != 0:
+            raise RuntimeError(f"Command failed (exit {exit_status}): {command}\n{err or out}")
+        return out
+
+    def _download_cloud_image(
+        self, image_url: str, image_checksum: str | None, timeout: int
+    ) -> str:
+        """
+        Download image_url to the node's image cache dir if not already present.
+
+        Returns the local path on the hypervisor node. Verifies image_checksum
+        (format "<algo>:<hex>", e.g. "sha256:abc123...") if given, re-downloading
+        is left to the caller's next attempt if verification fails.
+        """
+        filename = image_url.rstrip("/").rsplit("/", 1)[-1]
+        local_path = f"{_IMAGE_CACHE_DIR}/{filename}"
+
+        exists = self._run_node_command(
+            f"mkdir -p {_IMAGE_CACHE_DIR} && test -f {local_path} && echo EXISTS || echo MISSING",
+            timeout=30,
+        )
+        if "EXISTS" not in exists:
+            _logger.info(f"Downloading cloud image {image_url} -> {local_path}")
+            self._run_node_command(
+                f"wget -q -O {local_path}.tmp '{image_url}' && mv {local_path}.tmp {local_path}",
+                timeout=timeout,
+            )
+
+        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():
+                # Remove the bad file so a retry re-downloads instead of reusing it.
+                self._run_node_command(f"rm -f {local_path}", timeout=30)
+                raise RuntimeError(
+                    f"Checksum mismatch for {image_url}: expected {expected}, got {actual}"
+                )
+
+        return local_path
+
+    def _find_default_image_storage(self) -> str:
+        """Find a storage suitable for VM root disks (content includes 'images').
+
+        Proxmox's /storage API omits the "enabled" field entirely for storages
+        that were never explicitly toggled — it is not present-and-falsy, it is
+        just absent, defaulting to enabled. Only an explicit 0 means disabled.
+
+        Queries the node-scoped /nodes/{node}/storage endpoint, not the
+        cluster-wide /storage one: a storage can be configured with a "nodes"
+        restriction limiting it to other cluster members, and the cluster-wide
+        list doesn't reflect that — it would happily return a storage this
+        node can't actually see, and "qm importdisk" would fail with
+        "storage 'X' is not available on node 'Y'" after the VM shell was
+        already created.
+        """
+        for storage in self._node_api().storage.get():
+            content = storage.get("content", "")
+            if "images" in content and storage.get("enabled", 1) != 0:
+                return storage["storage"]
+        raise ValueError(
+            "No storage with content='images' found. Configure a storage for VM disks."
+        )
+
+    def _get_storage_path(self, storage: str) -> str:
+        """Resolve a storage's filesystem path on the node.
+
+        Needed to write Cloud-Init snippets directly: Proxmox's
+        /storage/{s}/upload API only accepts content in {iso, vztmpl,
+        import} — "snippets" is rejected outright, so snippets must be
+        written straight to the filesystem instead. Only dir-backed storages
+        (dir, nfs, cifs, cephfs) expose "path"; those are also the only
+        storage types Proxmox itself allows content='snippets' on.
+        """
+        config = self._api.storage(storage).get()
+        path = config.get("path")
+        if not path:
+            raise ValueError(
+                f"Storage '{storage}' has no filesystem path (content='snippets' "
+                "requires a dir/nfs/cifs/cephfs-backed storage)"
+            )
+        return path
+
+    def get_image_storages(self) -> List[StorageTargetDict]:
+        """List node-available storage pools suitable for a new VM's root disk."""
+        targets: List[StorageTargetDict] = []
+        for storage in self._node_api().storage.get():
+            content = storage.get("content", "")
+            if "images" not in content or storage.get("enabled", 1) == 0:
+                continue
+            if storage.get("active", 1) == 0:
+                continue
+            total = storage.get("total") or 0
+            avail = storage.get("avail") or 0
+            targets.append(
+                {
+                    "name": storage["storage"],
+                    "type": storage.get("type", ""),
+                    "total_gb": round(total / (1024**3), 1),
+                    "available_gb": round(avail / (1024**3), 1),
+                }
+            )
+        return targets
+
     def _wait_for_task(self, upid: str, timeout: int = 120) -> None:
         """
         Poll a Proxmox task until completion.
@@ -51,68 +205,107 @@ class ProxmoxVMProvisionMixin:
         self,
         name: str,
         *,
-        template: str,
+        image_url: str,
         cpu: int,
         memory: int,
         nics: List[Dict[str, Any]],
         cloud_init_config: Dict[str, Any],
+        image_checksum: str | None = None,
         ssh_public_keys: List[str] | None = None,
         disk_resize_gb: int | None = None,
+        storage: str | None = None,
+        download_timeout: int = 300,
         timeout: int = 180,
     ) -> VMProvisionResultDict:
         """
-        Create a new VM from a Cloud-Init template via Proxmox API.
+        Create a new VM from a downloaded cloud image via Proxmox API.
 
         Steps:
           1. Get next available VMID from cluster
-          2. Clone template VM (full clone, new VMID)
+          2. Create an empty VM shell (no clone — no pre-existing template needed)
-          3. Configure CPU, memory, and network interfaces
+          3. Download the cloud image on the node (cached by filename) and
-          4. Verify snippet storage exists
+             import it as the VM's root disk
-          5. Render cloud-init config to YAML and upload
+          4. Configure CPU, memory, and network interfaces
-          6. Set Cloud-Init config references and SSH keys
+          5. Verify snippet storage exists
-          7. Optionally resize root disk
+          6. Render cloud-init config to YAML and upload
-          8. Start the VM
+          7. Set Cloud-Init config references and SSH keys
-          9. Return VMID, name, node
+          8. Optionally resize root disk
+          9. Start the VM
+          10. Return VMID, name, node
 
         Args:
             name: new VM display name
-            template: template VMID/name to clone from
+            image_url: URL of the cloud image to download and use as root disk
             cpu: number of vCPUs
             memory: RAM in MB
             nics: list of NIC config dicts (bridge, vlan_tag/trunk_vlan_tags, dhcp flag)
             cloud_init_config: user-data dict (will be YAML-rendered)
+            image_checksum: expected "<algo>:<hex>" checksum of the image, verified
+                after download (None = no verification)
             ssh_public_keys: SSH public keys to inject
             disk_resize_gb: resize root disk to this size (None = no resize)
-            timeout: max seconds for provisioning
+            storage: storage pool for the root disk (None = auto-detect first
+                enabled, node-available storage with content='images')
+            download_timeout: max seconds for the image download (skipped if cached)
+            timeout: max seconds for the remaining provisioning steps
 
         Returns:
             {"vmid": str, "name": str, "node": str}
 
         Raises:
-            RuntimeError: provisioning failure (clone, config, timeout, etc.)
+            RuntimeError: provisioning failure (download, import, config, timeout, etc.)
             ValueError: invalid storage or configuration
         """
         try:
-            _logger.info(f"Creating VM '{name}' from template {template}")
+            _logger.info(f"Creating VM '{name}' from image {image_url}")
 
             # Step 1: Get next VMID
             next_vmid = self._api.cluster.nextid.get()
             vmid = int(next_vmid)
             _logger.info(f"Allocated VMID {vmid}")
 
-            # Step 2: Clone template
+            # Step 2: Create empty VM shell (no disks yet)
-            _logger.info(f"Cloning template {template} → VMID {vmid}")
+            _logger.info(f"Creating VM shell {vmid}")
-            clone_upid = self._node_api().qemu(template).clone.post(
+            self._node_api().qemu.post(
-                newid=vmid,
+                vmid=vmid,
-                full=1,
                 name=name,
+                memory=memory,
+                cores=cpu,
+                ostype="l26",
+                scsihw="virtio-scsi-pci",
             )
-            self._wait_for_task(clone_upid, timeout=timeout)
 
-            # Step 3: Configure CPU, memory, and NICs
+            # Step 3: Download cloud image (cached) and import as root disk
-            _logger.info(f"Configuring VM {vmid}: {cpu} CPU, {memory}MB RAM, {len(nics)} NIC(s)")
+            local_path = self._download_cloud_image(
+                image_url, image_checksum, timeout=download_timeout
+            )
+            image_storage = storage or self._find_default_image_storage()
 
-            config_args = {"cores": cpu, "memory": memory}
+            _logger.info(f"Importing {local_path} into VM {vmid} on storage {image_storage}")
+            self._run_node_command(
+                f"qm importdisk {vmid} {local_path} {image_storage} --format qcow2",
+                timeout=timeout,
+            )
+
+            # Proxmox leaves the imported disk as an "unusedN" reference — find
+            # it and attach it as the boot disk.
+            imported_config = self._node_api().qemu(vmid).config.get()
+            unused_value = next(
+                (v for k, v in imported_config.items() if k.startswith("unused")), None
+            )
+            if not unused_value:
+                raise RuntimeError(
+                    f"Disk import for VM {vmid} did not produce an unused disk reference"
+                )
+            self._node_api().qemu(vmid).config.post(
+                scsi0=f"{unused_value},discard=on",
+                boot="order=scsi0",
+            )
+
+            # Step 4: Configure network interfaces (CPU/memory already set at shell creation)
+            _logger.info(f"Configuring {len(nics)} NIC(s) for VM {vmid}")
+
+            config_args: Dict[str, Any] = {}
 
             # Build NIC config strings generically
             for i, nic in enumerate(nics):
@@ -120,8 +313,9 @@ class ProxmoxVMProvisionMixin:
                 if not bridge:
                     raise ValueError(f"NIC {i}: bridge is required")
 
-                # Build base config: model + bridge
+                # Build base config: model[=mac] + bridge
-                net_config = f"virtio,bridge={bridge}"
+                mac = nic.get("mac")
+                net_config = f"virtio={mac},bridge={bridge}" if mac else f"virtio,bridge={bridge}"
 
                 # Add VLAN configuration (access vs trunk)
                 if "trunk_vlan_tags" in nic and nic["trunk_vlan_tags"]:
@@ -134,13 +328,15 @@ class ProxmoxVMProvisionMixin:
 
             self._node_api().qemu(vmid).config.post(**config_args)
 
-            # Step 4: Verify snippet storage exists
+            # Step 5: Verify snippet storage exists
+            # (enabled is absent-not-falsy, and node-scoping matters — see
+            # _find_default_image_storage)
             _logger.info("Checking for snippet storage...")
-            storages = self._api.storage.get()
+            storages = self._node_api().storage.get()
             snippet_storage = None
             for storage in storages:
                 content = storage.get("content", "")
-                if "snippets" in content and storage.get("enabled"):
+                if "snippets" in content and storage.get("enabled", 1) != 0:
                     snippet_storage = storage["storage"]
                     break
 
@@ -152,7 +348,7 @@ class ProxmoxVMProvisionMixin:
                 )
             _logger.info(f"Using snippet storage: {snippet_storage}")
 
-            # Step 5: Render and upload Cloud-Init config
+            # Step 6: Render and upload Cloud-Init config
             _logger.info(f"Rendering Cloud-Init config for VMID {vmid}")
 
             user_data_yaml = "#cloud-config\n" + yaml.dump(
@@ -160,20 +356,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 /storage/{s}/upload API only accepts content in
-            self._node_api().storage(snippet_storage).upload.post(
+            # {iso, vztmpl, import} — "snippets" is rejected outright
-                content="snippets",
+            # ("does not have a value in the enumeration"). Snippets can only
-                filename=filename,
+            # be written directly to the filesystem, so resolve the storage's
-                data=user_data_yaml,
+            # backing path and write the file over SSH instead.
+            storage_path = self._get_storage_path(snippet_storage)
+            encoded = base64.b64encode(user_data_yaml.encode("utf-8")).decode("ascii")
+            self._run_node_command(
+                f"mkdir -p {storage_path}/snippets && "
+                f"echo {encoded} | base64 -d > {storage_path}/snippets/{filename}",
+                timeout=30,
             )
 
-            # Step 6: Configure Cloud-Init references and SSH keys
+            # Step 7: Configure Cloud-Init references and SSH keys
             _logger.info(f"Setting Cloud-Init config for VM {vmid}")
 
             cloud_init_args = {
-                "ide2": f"{snippet_storage}:cloudinit",
+                # The cloud-init drive is a disk image — it needs a storage
+                # with content='images' (same requirement as the root disk),
+                # NOT the snippet storage (content='snippets'). These are
+                # often different storages; Proxmox fails at VM start with
+                # "storage 'X' does not support content-type 'images'" if
+                # this points at a snippets-only storage.
+                "ide2": f"{image_storage}:cloudinit",
                 "citype": "nocloud",
                 "cicustom": f"user={snippet_storage}:snippets/{filename}",
             }
@@ -191,7 +399,7 @@ class ProxmoxVMProvisionMixin:
 
             self._node_api().qemu(vmid).config.post(**cloud_init_args)
 
-            # Step 7: Optionally resize root disk
+            # Step 8: Optionally resize root disk
             if disk_resize_gb is not None:
                 _logger.info(f"Resizing root disk to {disk_resize_gb}GB")
                 # Find root disk (scsi0, virtio0, ide0, sata0 — whichever is first)
@@ -212,7 +420,7 @@ class ProxmoxVMProvisionMixin:
                 except Exception as e:
                     _logger.warning(f"Failed to resize disk: {e}, continuing anyway")
 
-            # Step 8: Start the VM
+            # Step 9: Start the VM
             _logger.info(f"Starting VM {vmid}")
             start_upid = self._node_api().qemu(vmid).status.start.post()
             self._wait_for_task(start_upid, timeout=timeout)
@@ -344,9 +552,7 @@ class ProxmoxVMProvisionMixin:
             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
@@ -415,6 +621,15 @@ class ProxmoxVMProvisionMixin:
             if not name:
                 continue
             # A vnet's VLAN is already fixed by its zone/tag — no separate vlan_tag applies.
-            targets.append({"name": name, "kind": "vnet", "vlan_aware": False})
+            tag = vnet.get("tag")
+            fixed_vlan_tag = int(tag) if tag is not None else None
+            targets.append(
+                {
+                    "name": name,
+                    "kind": "vnet",
+                    "vlan_aware": False,
+                    "fixed_vlan_tag": fixed_vlan_tag,
+                }
+            )
 
         return targets

tests/test_vm_provision_mixin.py

@@ -2,6 +2,8 @@
 
 from __future__ import annotations
 
+import base64
+
 import pytest
 from unittest.mock import MagicMock, patch
 from napalm_proxmox.driver import ProxmoxDriver
@@ -74,20 +76,29 @@ 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.get.return_value = [
+    mock_api.storage.return_value.get.return_value = {"path": "/var/lib/vz"}
-        {"storage": "snippets", "type": "dir", "content": "snippets", "enabled": 1}
-    ]
 
     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 operations
     mock_vm = MagicMock()
     mock_node.qemu.return_value = mock_vm
-    mock_vm.clone.post.return_value = "UPID:pve1:123:clone"
+    mock_node.qemu.post.return_value = None
     mock_vm.config.post.return_value = None
-    mock_vm.config.get.return_value = {"scsi0": "local:100/vm-101-disk-0.raw"}
+    mock_vm.config.get.return_value = {
+        "unused0": "local-lvm:vm-101-disk-0",
+        "scsi0": "local-lvm:vm-101-disk-0",
+    }
     mock_vm.status.start.post.return_value = "UPID:pve1:124:start"
 
     # Mock task completion
@@ -95,15 +106,10 @@ 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",
-            template="100",
+            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}],
@@ -114,13 +120,25 @@ def test_create_vm_from_cloud_init_single_nic():
     assert result["vmid"] == "101"
     assert result["name"] == "test-vm"
     assert result["node"] == "pve1"
-    assert mock_vm.clone.post.called
+    assert mock_node.qemu.post.called
+    mixin._download_cloud_image.assert_called_once()
 
     # Verify NIC config was set correctly: net0 with tag=10, DHCP enabled
-    config_call_args = mock_vm.config.post.call_args_list[0]  # First call (cores/memory/net0)
+    net_call_args = next(c for c in mock_vm.config.post.call_args_list if "net0" in c[1])
-    assert "net0" in config_call_args[1]
+    assert "tag=10" in net_call_args[1]["net0"]
-    assert "tag=10" in config_call_args[1]["net0"]
+    assert "vmbr0" in net_call_args[1]["net0"]
-    assert "vmbr0" in config_call_args[1]["net0"]
+
+    # Regression: the snippet must be written directly to the filesystem via
+    # SSH, not uploaded via the /storage/upload API — real Proxmox rejects
+    # content='snippets' on that endpoint outright (see
+    # test_create_vm_writes_snippet_via_ssh_with_correct_content for the
+    # dedicated check).
+    write_cmd = next(
+        c[0][0]
+        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():
@@ -131,20 +149,29 @@ 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.get.return_value = [
+    mock_api.storage.return_value.get.return_value = {"path": "/var/lib/vz"}
-        {"storage": "snippets", "type": "dir", "content": "snippets", "enabled": 1}
-    ]
 
     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 operations
     mock_vm = MagicMock()
     mock_node.qemu.return_value = mock_vm
-    mock_vm.clone.post.return_value = "UPID:pve1:125:clone"
+    mock_node.qemu.post.return_value = None
     mock_vm.config.post.return_value = None
-    mock_vm.config.get.return_value = {"scsi0": "local:100/vm-102-disk-0.raw"}
+    mock_vm.config.get.return_value = {
+        "unused0": "local-lvm:vm-102-disk-0",
+        "scsi0": "local-lvm:vm-102-disk-0",
+    }
     mock_vm.status.start.post.return_value = "UPID:pve1:126:start"
 
     # Mock task completion
@@ -152,20 +179,19 @@ 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",
-            template="100",
+            image_url="https://cloud.debian.org/images/cloud/bookworm/latest/debian-12-genericcloud-amd64.qcow2",
             cpu=4,
             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,
@@ -175,16 +201,16 @@ def test_create_vm_from_cloud_init_dual_nic_trunk():
     assert result["name"] == "wireshark-sat-1"
 
     # Verify both NICs configured
-    config_call_args = mock_vm.config.post.call_args_list[0]
+    net_call_args = next(c for c in mock_vm.config.post.call_args_list if "net0" in c[1])
-    assert "net0" in config_call_args[1]
+    assert "net1" in net_call_args[1]
-    assert "net1" in config_call_args[1]
+    assert "tag=10" in net_call_args[1]["net0"]
-    assert "tag=10" in config_call_args[1]["net0"]
+    assert "trunks=20;30" in net_call_args[1]["net1"]
-    assert "trunks=20;30" in config_call_args[1]["net1"]
+    assert "vmbr1" in net_call_args[1]["net1"]
-    assert "vmbr1" in config_call_args[1]["net1"]
 
     # Verify DHCP config: ipconfig0 yes, ipconfig1 no
-    cloud_init_call_args = mock_vm.config.post.call_args_list[1]  # Second call (ipconfig)
+    cloud_init_call_args = next(
-    assert "ipconfig0" in cloud_init_call_args[1]
+        c for c in mock_vm.config.post.call_args_list if "ipconfig0" in c[1]
+    )
     assert cloud_init_call_args[1]["ipconfig0"] == "ip=dhcp"
     assert "ipconfig1" not in cloud_init_call_args[1]  # net1 has no DHCP
 
@@ -194,22 +220,26 @@ def test_create_vm_missing_snippet_storage():
     mixin = ProxmoxVMProvisionMixin()
     mixin._node_name = "pve1"
 
-    # Mock API with no snippet storage
+    # Mock API with images storage but no snippet storage
     mock_api = MagicMock()
     mock_api.cluster.nextid.get.return_value = 101
-    mock_api.storage.get.return_value = [
-        {"storage": "local-lvm", "type": "lvmthin", "content": "images,rootdir", "enabled": 1}
-    ]
 
     mixin._api = mock_api
 
-    # Mock node for clone operation (so we get to the storage check)
     mock_node = MagicMock()
+    mock_node.storage.get.return_value = [
+        {"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._run_node_command = MagicMock(return_value="")
     mock_vm = MagicMock()
     mock_node.qemu.return_value = mock_vm
-    mock_vm.clone.post.return_value = "UPID:pve1:123:clone"
+    mock_node.qemu.post.return_value = None
     mock_vm.config.post.return_value = None
+    mock_vm.config.get.return_value = {"unused0": "local-lvm:vm-101-disk-0"}
 
     # Mock task to allow clone to complete
     mock_task = MagicMock()
@@ -220,7 +250,7 @@ def test_create_vm_missing_snippet_storage():
         with patch("napalm_proxmox.vm_provision_mixin.time.sleep"):
             mixin.create_vm_from_cloud_init(
                 name="test-vm",
-                template="100",
+                image_url="https://cloud.debian.org/images/cloud/bookworm/latest/debian-12-genericcloud-amd64.qcow2",
                 cpu=2,
                 memory=2048,
                 nics=[{"bridge": "vmbr0"}],
@@ -236,20 +266,29 @@ def test_create_vm_with_disk_resize():
     # Mock API hierarchy
     mock_api = MagicMock()
     mock_api.cluster.nextid.get.return_value = 103
-    mock_api.storage.get.return_value = [
+    mock_api.storage.return_value.get.return_value = {"path": "/var/lib/vz"}
-        {"storage": "snippets", "type": "dir", "content": "snippets", "enabled": 1}
-    ]
 
     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 operations
     mock_vm = MagicMock()
     mock_node.qemu.return_value = mock_vm
-    mock_vm.clone.post.return_value = "UPID:pve1:127:clone"
+    mock_node.qemu.post.return_value = None
     mock_vm.config.post.return_value = None
-    mock_vm.config.get.return_value = {"scsi0": "local:100/vm-103-disk-0.raw"}
+    mock_vm.config.get.return_value = {
+        "unused0": "local-lvm:vm-103-disk-0",
+        "scsi0": "local-lvm:vm-103-disk-0",
+    }
     mock_vm.resize.put.return_value = None
     mock_vm.status.start.post.return_value = "UPID:pve1:128:start"
 
@@ -258,15 +297,10 @@ 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",
-            template="100",
+            image_url="https://cloud.debian.org/images/cloud/bookworm/latest/debian-12-genericcloud-amd64.qcow2",
             cpu=2,
             memory=2048,
             nics=[{"bridge": "vmbr0"}],
@@ -445,9 +479,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()
@@ -472,3 +504,662 @@ def test_get_network_targets_sdn_vnet_never_vlan_aware():
     assert len(targets) == 2
     assert all(t["kind"] == "vnet" for t in targets)
     assert all(t["vlan_aware"] is False for t in targets)
+
+
+def test_get_network_targets_vnet_exposes_fixed_vlan_tag():
+    """A vnet's fixed_vlan_tag surfaces its SDN tag for display purposes."""
+    mixin = ProxmoxVMProvisionMixin()
+    mixin._get_node_network = MagicMock(return_value=[])
+    mixin._get_sdn_vnets = MagicMock(
+        return_value=[{"vnet": "vnet0", "zone": "zone-vlan", "tag": 10}]
+    )
+
+    targets = mixin.get_network_targets()
+
+    assert targets[0]["fixed_vlan_tag"] == 10
+
+
+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(return_value=[{"vnet": "vnet2", "zone": "zone-vxlan"}])
+
+    targets = mixin.get_network_targets()
+
+    assert targets[0]["fixed_vlan_tag"] is None
+
+
+def test_get_network_targets_bridge_has_no_fixed_vlan_tag():
+    """Bridges never carry a single fixed VLAN tag — only vnets do."""
+    mixin = ProxmoxVMProvisionMixin()
+    mixin._get_node_network = MagicMock(
+        return_value=[{"iface": "vmbr0", "type": "bridge", "bridge_vlan_aware": 1}]
+    )
+    mixin._get_sdn_vnets = MagicMock(return_value=[])
+
+    targets = mixin.get_network_targets()
+
+    assert "fixed_vlan_tag" not in targets[0] or targets[0]["fixed_vlan_tag"] is None
+
+
+# ---------------------------------------------------------------------------
+# _run_node_command
+# ---------------------------------------------------------------------------
+
+
+def _mock_ssh_exec(exit_status: int, stdout_text: str = "", stderr_text: str = ""):
+    """Build a mock (stdin, stdout, stderr) tuple as returned by exec_command."""
+    mock_stdin = MagicMock()
+    mock_stdout = MagicMock()
+    mock_stderr = MagicMock()
+    mock_stdout.channel.recv_exit_status.return_value = exit_status
+    mock_stdout.read.return_value = stdout_text.encode()
+    mock_stderr.read.return_value = stderr_text.encode()
+    return mock_stdin, mock_stdout, mock_stderr
+
+
+def test_run_node_command_success_returns_stdout():
+    mixin = ProxmoxVMProvisionMixin()
+    mixin._ssh_client = MagicMock()
+    mixin._ssh_client.exec_command.return_value = _mock_ssh_exec(0, stdout_text="hello\n")
+
+    result = mixin._run_node_command("echo hello", timeout=10)
+
+    assert result == "hello"
+
+
+def test_run_node_command_nonzero_exit_raises():
+    mixin = ProxmoxVMProvisionMixin()
+    mixin._ssh_client = MagicMock()
+    mixin._ssh_client.exec_command.return_value = _mock_ssh_exec(
+        1, stderr_text="No such file or directory"
+    )
+
+    with pytest.raises(RuntimeError, match="No such file or directory"):
+        mixin._run_node_command("cat /nonexistent", timeout=10)
+
+
+# ---------------------------------------------------------------------------
+# _download_cloud_image
+# ---------------------------------------------------------------------------
+
+
+def test_download_cloud_image_skips_download_when_cached():
+    mixin = ProxmoxVMProvisionMixin()
+    mixin._run_node_command = MagicMock(return_value="EXISTS")
+
+    path = mixin._download_cloud_image(
+        "https://cloud.debian.org/images/cloud/bookworm/latest/debian-12-genericcloud-amd64.qcow2",
+        None,
+        timeout=300,
+    )
+
+    assert path.endswith("debian-12-genericcloud-amd64.qcow2")
+    # Only the existence check ran — no wget call
+    assert mixin._run_node_command.call_count == 1
+    assert "wget" not in mixin._run_node_command.call_args_list[0][0][0]
+
+
+def test_download_cloud_image_downloads_when_missing():
+    mixin = ProxmoxVMProvisionMixin()
+    mixin._run_node_command = MagicMock(return_value="MISSING")
+
+    mixin._download_cloud_image(
+        "https://cloud.debian.org/images/cloud/bookworm/latest/debian-12-genericcloud-amd64.qcow2",
+        None,
+        timeout=300,
+    )
+
+    commands = [c[0][0] for c in mixin._run_node_command.call_args_list]
+    assert any("wget" in cmd for cmd in commands)
+
+
+def test_download_cloud_image_verifies_matching_checksum():
+    mixin = ProxmoxVMProvisionMixin()
+    mixin._run_node_command = MagicMock(
+        side_effect=["EXISTS", "abc123"]  # existence check, then checksum
+    )
+
+    path = mixin._download_cloud_image(
+        "https://cloud.debian.org/images/cloud/bookworm/latest/debian-12-genericcloud-amd64.qcow2",
+        "sha256:abc123",
+        timeout=300,
+    )
+
+    assert path.endswith("debian-12-genericcloud-amd64.qcow2")
+
+
+def test_download_cloud_image_checksum_mismatch_raises_and_removes_file():
+    mixin = ProxmoxVMProvisionMixin()
+    mixin._run_node_command = MagicMock(
+        side_effect=["EXISTS", "wrong-checksum", ""]  # existence, checksum, rm
+    )
+
+    with pytest.raises(RuntimeError, match="Checksum mismatch"):
+        mixin._download_cloud_image(
+            "https://cloud.debian.org/images/cloud/bookworm/latest/debian-12-genericcloud-amd64.qcow2",
+            "sha256:abc123",
+            timeout=300,
+        )
+
+    commands = [c[0][0] for c in mixin._run_node_command.call_args_list]
+    assert any(cmd.startswith("rm -f") for cmd in commands)
+
+
+# ---------------------------------------------------------------------------
+# _find_default_image_storage
+# ---------------------------------------------------------------------------
+
+
+def test_find_default_image_storage_treats_absent_enabled_as_enabled():
+    """Proxmox omits "enabled" entirely for storages never explicitly toggled —
+    absent must mean enabled, not disabled. Regression: this was previously
+    treated as falsy, causing every real Proxmox server to report no usable
+    image storage even when several existed."""
+    mixin = ProxmoxVMProvisionMixin()
+    mock_node = MagicMock()
+    mock_node.storage.get.return_value = [
+        {"storage": "local-lvm", "type": "lvmthin", "content": "images,rootdir"},
+    ]
+    mixin._node_api = MagicMock(return_value=mock_node)
+
+    storage = mixin._find_default_image_storage()
+
+    assert storage == "local-lvm"
+
+
+def test_find_default_image_storage_excludes_explicitly_disabled():
+    mixin = ProxmoxVMProvisionMixin()
+    mock_node = MagicMock()
+    mock_node.storage.get.return_value = [
+        {"storage": "old-storage", "type": "dir", "content": "images", "enabled": 0},
+        {"storage": "local-lvm", "type": "lvmthin", "content": "images,rootdir", "enabled": 1},
+    ]
+    mixin._node_api = MagicMock(return_value=mock_node)
+
+    storage = mixin._find_default_image_storage()
+
+    assert storage == "local-lvm"
+
+
+def test_find_default_image_storage_raises_when_none_found():
+    mixin = ProxmoxVMProvisionMixin()
+    mock_node = MagicMock()
+    mock_node.storage.get.return_value = [
+        {"storage": "local", "type": "dir", "content": "backup,iso,vztmpl"},
+    ]
+    mixin._node_api = MagicMock(return_value=mock_node)
+
+    with pytest.raises(ValueError, match="images"):
+        mixin._find_default_image_storage()
+
+
+def test_find_default_image_storage_excludes_storage_restricted_to_other_nodes():
+    """Regression: a storage configured cluster-wide but restricted via 'nodes'
+    to other cluster members must not be picked — /nodes/{node}/storage (unlike
+    the cluster-wide /storage endpoint) only lists what's actually available on
+    this node, so querying it is what makes this exclusion happen naturally.
+    This was the real-world bug: local-lvm (nodes=pve-garden) was picked for a
+    VM on pve-02, and 'qm importdisk' failed with 'storage not available on
+    node', leaving a VM shell with no disk attached."""
+    mixin = ProxmoxVMProvisionMixin()
+    mock_node = MagicMock()
+    # /nodes/pve-02/storage never even lists local-lvm — Proxmox itself filters
+    # node-restricted storages out of this endpoint.
+    mock_node.storage.get.return_value = [
+        {"storage": "local-zfs", "type": "zfspool", "content": "images,rootdir"},
+    ]
+    mixin._node_api = MagicMock(return_value=mock_node)
+
+    storage = mixin._find_default_image_storage()
+
+    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
+    "enabled" key at all) — this used to make create_vm_from_cloud_init fail
+    with "No storage with content='images' found" even though usable storage
+    existed."""
+    mixin = ProxmoxVMProvisionMixin()
+    mixin._node_name = "pve1"
+
+    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 = [
+        {"storage": "local-lvm", "type": "lvmthin", "content": "images,rootdir"},
+        {"storage": "local-zfs", "type": "zfspool", "content": "rootdir,images"},
+        {"storage": "local", "type": "dir", "content": "backup,iso,vztmpl"},
+        {"storage": "snippets", "type": "dir", "content": "snippets"},
+    ]
+    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-104-disk-0",
+        "scsi0": "local-lvm:vm-104-disk-0",
+    }
+    mock_vm.status.start.post.return_value = "UPID:pve1:130: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"):
+        result = mixin.create_vm_from_cloud_init(
+            name="real-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": "real-shape-vm"},
+        )
+
+    assert result["vmid"] == "104"
+
+
+def test_create_vm_from_cloud_init_explicit_storage_skips_auto_detect():
+    """When storage= is given explicitly, it's used directly and
+    _find_default_image_storage is never consulted (so an explicit choice
+    always wins, even if auto-detect would have picked something else)."""
+    mixin = ProxmoxVMProvisionMixin()
+    mixin._node_name = "pve1"
+
+    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 = [
+        {"storage": "local-lvm", "type": "lvmthin", "content": "images,rootdir", "enabled": 1},
+        {"storage": "fast-zfs", "type": "zfspool", "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._find_default_image_storage = MagicMock(
+        side_effect=AssertionError("should not be called")
+    )
+    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": "fast-zfs:vm-105-disk-0",
+        "scsi0": "fast-zfs:vm-105-disk-0",
+    }
+    mock_vm.status.start.post.return_value = "UPID:pve1:131: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"):
+        result = mixin.create_vm_from_cloud_init(
+            name="explicit-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": "explicit-storage-vm"},
+            storage="fast-zfs",
+        )
+
+    assert result["vmid"] == "105"
+    mixin._find_default_image_storage.assert_not_called()
+    import_cmd = next(
+        c[0][0] for c in mixin._run_node_command.call_args_list if "qm importdisk" in c[0][0]
+    )
+    assert "fast-zfs" in import_cmd
+    assert "local-lvm" not in import_cmd
+
+
+# ---------------------------------------------------------------------------
+# get_image_storages
+# ---------------------------------------------------------------------------
+
+
+def test_get_image_storages_filters_to_images_content():
+    mixin = ProxmoxVMProvisionMixin()
+    mock_node = MagicMock()
+    mock_node.storage.get.return_value = [
+        {
+            "storage": "local-zfs",
+            "type": "zfspool",
+            "content": "images,rootdir",
+            "total": 100 * 1024**3,
+            "avail": 40 * 1024**3,
+        },
+        {"storage": "local", "type": "dir", "content": "backup,iso,vztmpl,snippets"},
+    ]
+    mixin._node_api = MagicMock(return_value=mock_node)
+
+    targets = mixin.get_image_storages()
+
+    assert len(targets) == 1
+    assert targets[0]["name"] == "local-zfs"
+    assert targets[0]["type"] == "zfspool"
+    assert targets[0]["total_gb"] == 100.0
+    assert targets[0]["available_gb"] == 40.0
+
+
+def test_get_image_storages_excludes_disabled_and_inactive():
+    mixin = ProxmoxVMProvisionMixin()
+    mock_node = MagicMock()
+    mock_node.storage.get.return_value = [
+        {"storage": "disabled-store", "type": "dir", "content": "images", "enabled": 0},
+        {"storage": "inactive-store", "type": "nfs", "content": "images", "active": 0},
+        {"storage": "ok-store", "type": "dir", "content": "images"},
+    ]
+    mixin._node_api = MagicMock(return_value=mock_node)
+
+    targets = mixin.get_image_storages()
+
+    assert [t["name"] for t in targets] == ["ok-store"]
+
+
+def test_get_image_storages_excludes_storage_restricted_to_other_nodes():
+    """Node-scoped query naturally excludes storages Proxmox itself doesn't
+    list for this node (e.g. restricted via 'nodes' to other cluster members)."""
+    mixin = ProxmoxVMProvisionMixin()
+    mock_node = MagicMock()
+    mock_node.storage.get.return_value = [
+        {"storage": "local-zfs", "type": "zfspool", "content": "images,rootdir"},
+    ]
+    mixin._node_api = MagicMock(return_value=mock_node)
+
+    targets = mixin.get_image_storages()
+
+    assert [t["name"] for t in targets] == ["local-zfs"]
+
+
+# ---------------------------------------------------------------------------
+# Snippet upload — must be a real multipart file, not a filename+data string pair
+# ---------------------------------------------------------------------------
+
+
+def test_create_vm_writes_snippet_via_ssh_with_correct_content():
+    """Regression: real Proxmox's /storage/{s}/upload endpoint rejects
+    content='snippets' outright ("value 'snippets' does not have a value in
+    the enumeration 'iso, vztmpl, import'") — that endpoint only handles
+    ISOs, container templates, and imports. Snippets can only be written
+    directly to the storage's filesystem path, so this must go through SSH
+    (_run_node_command), not the upload API.
+    (observed live: 400 Bad Request from Proxmox, right after the earlier
+    multipart-upload fix had already gotten past a prior RemoteDisconnected
+    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 = [
+        {"storage": "local-zfs", "type": "zfspool", "content": "images,rootdir", "enabled": 1},
+        {"storage": "local", "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-zfs:vm-106-disk-0",
+        "scsi0": "local-zfs:vm-106-disk-0",
+    }
+    mock_vm.status.start.post.return_value = "UPID:pve1:132: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="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": "write-shape-vm", "chpasswd": {"expire": False}},
+        )
+
+    # No call to the upload API at all — snippets aren't a valid content
+    # type there.
+    mock_node.storage.assert_not_called()
+
+    write_cmd = next(
+        c[0][0]
+        for c in mixin._run_node_command.call_args_list
+        if "snippets/106-user-data.yaml" in c[0][0]
+    )
+    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: 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/")