fix(vm_provision_mixin): storage 'enabled' absent means enabled, not disabled

Proxmox's /storage API omits the "enabled" key entirely for storages that
were never explicitly toggled, rather than defaulting it to 1 — it isn't
present-and-falsy, it's just absent. Both _find_default_image_storage and
the snippet-storage discovery treated storage.get("enabled") as truthy-check,
so every storage without an explicit "enabled": 1 was silently excluded.

Confirmed live against a real Proxmox test server: local-lvm, local-zfs, and
fast-zfs all had content=images with no "enabled" key at all, causing
create_vm_from_cloud_init to always fail with "No storage with
content='images' found" despite multiple valid storages existing. All prior
tests used "enabled": 1 explicitly in their fixtures, masking the bug.

Fix: storage.get("enabled", 1) != 0 — absent or truthy means enabled, only
an explicit 0 excludes it. 4 new regression tests, 28 total pass.

napalm_proxmox/vm_provision_mixin.py

@@ -12,10 +12,110 @@ from napalm_device_types.models import NetworkTargetDict, VMProvisionResultDict,
 
 _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.
+        """
+        for storage in self._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 _wait_for_task(self, upid: str, timeout: int = 120) -> None:
         """
         Poll a Proxmox task until completion.
@@ -51,68 +151,104 @@ 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,
+        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)
-          3. Configure CPU, memory, and network interfaces
-          4. Verify snippet storage exists
-          5. Render cloud-init config to YAML and upload
-          6. Set Cloud-Init config references and SSH keys
-          7. Optionally resize root disk
-          8. Start the VM
-          9. Return VMID, name, node
+          2. Create an empty VM shell (no clone — no pre-existing template needed)
+          3. Download the cloud image on the node (cached by filename) and
+             import it as the VM's root disk
+          4. Configure CPU, memory, and network interfaces
+          5. Verify snippet storage exists
+          6. Render cloud-init config to YAML and upload
+          7. Set Cloud-Init config references and SSH keys
+          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
+            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
-            _logger.info(f"Cloning template {template} → VMID {vmid}")
-            clone_upid = self._node_api().qemu(template).clone.post(
-                newid=vmid,
-                full=1,
+            # Step 2: Create empty VM shell (no disks yet)
+            _logger.info(f"Creating VM shell {vmid}")
+            self._node_api().qemu.post(
+                vmid=vmid,
                 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
-            _logger.info(f"Configuring VM {vmid}: {cpu} CPU, {memory}MB RAM, {len(nics)} NIC(s)")
+            # Step 3: Download cloud image (cached) and import as root disk
+            local_path = self._download_cloud_image(
+                image_url, image_checksum, timeout=download_timeout
+            )
+            image_storage = 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):
@@ -134,13 +270,14 @@ 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 on Proxmox — see _find_default_image_storage)
             _logger.info("Checking for snippet storage...")
             storages = self._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 +289,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(
@@ -169,7 +306,7 @@ class ProxmoxVMProvisionMixin:
                 data=user_data_yaml,
             )
 
-            # 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 = {
@@ -191,7 +328,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 +349,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)

tests/test_vm_provision_mixin.py

@@ -75,19 +75,25 @@ def test_create_vm_from_cloud_init_single_nic():
     mock_api = MagicMock()
     mock_api.cluster.nextid.get.return_value = 101
     mock_api.storage.get.return_value = [
-        {"storage": "snippets", "type": "dir", "content": "snippets", "enabled": 1}
+        {"storage": "local-lvm", "type": "lvmthin", "content": "images,rootdir", "enabled": 1},
+        {"storage": "snippets", "type": "dir", "content": "snippets", "enabled": 1},
     ]
 
     mock_node = MagicMock()
     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
@@ -103,7 +109,7 @@ def test_create_vm_from_cloud_init_single_nic():
     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,15 @@ 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)
-    assert "net0" in config_call_args[1]
-    assert "tag=10" in config_call_args[1]["net0"]
-    assert "vmbr0" in config_call_args[1]["net0"]
+    net_call_args = next(
+        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"]
 
 
 def test_create_vm_from_cloud_init_dual_nic_trunk():
@@ -132,19 +140,25 @@ def test_create_vm_from_cloud_init_dual_nic_trunk():
     mock_api = MagicMock()
     mock_api.cluster.nextid.get.return_value = 102
     mock_api.storage.get.return_value = [
-        {"storage": "snippets", "type": "dir", "content": "snippets", "enabled": 1}
+        {"storage": "local-lvm", "type": "lvmthin", "content": "images,rootdir", "enabled": 1},
+        {"storage": "snippets", "type": "dir", "content": "snippets", "enabled": 1},
     ]
 
     mock_node = MagicMock()
     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
@@ -160,7 +174,7 @@ def test_create_vm_from_cloud_init_dual_nic_trunk():
     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=[
@@ -175,16 +189,18 @@ 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]
-    assert "net0" in config_call_args[1]
-    assert "net1" in config_call_args[1]
-    assert "tag=10" in config_call_args[1]["net0"]
-    assert "trunks=20;30" in config_call_args[1]["net1"]
-    assert "vmbr1" in config_call_args[1]["net1"]
+    net_call_args = next(
+        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"]
+    assert "vmbr1" in net_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)
-    assert "ipconfig0" in cloud_init_call_args[1]
+    cloud_init_call_args = next(
+        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,7 +210,7 @@ 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 = [
@@ -203,13 +219,15 @@ def test_create_vm_missing_snippet_storage():
 
     mixin._api = mock_api
 
-    # Mock node for clone operation (so we get to the storage check)
     mock_node = MagicMock()
     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 +238,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"}],
@@ -237,19 +255,25 @@ def test_create_vm_with_disk_resize():
     mock_api = MagicMock()
     mock_api.cluster.nextid.get.return_value = 103
     mock_api.storage.get.return_value = [
-        {"storage": "snippets", "type": "dir", "content": "snippets", "enabled": 1}
+        {"storage": "local-lvm", "type": "lvmthin", "content": "images,rootdir", "enabled": 1},
+        {"storage": "snippets", "type": "dir", "content": "snippets", "enabled": 1},
     ]
 
     mock_node = MagicMock()
     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"
 
@@ -266,7 +290,7 @@ def test_create_vm_with_disk_resize():
     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"}],
@@ -511,3 +535,207 @@ def test_get_network_targets_bridge_has_no_fixed_vlan_tag():
     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()
+    mixin._api = MagicMock()
+    mixin._api.storage.get.return_value = [
+        {"storage": "local-lvm", "type": "lvmthin", "content": "images,rootdir"},
+    ]
+
+    storage = mixin._find_default_image_storage()
+
+    assert storage == "local-lvm"
+
+
+def test_find_default_image_storage_excludes_explicitly_disabled():
+    mixin = ProxmoxVMProvisionMixin()
+    mixin._api = MagicMock()
+    mixin._api.storage.get.return_value = [
+        {"storage": "old-storage", "type": "dir", "content": "images", "enabled": 0},
+        {"storage": "local-lvm", "type": "lvmthin", "content": "images,rootdir", "enabled": 1},
+    ]
+
+    storage = mixin._find_default_image_storage()
+
+    assert storage == "local-lvm"
+
+
+def test_find_default_image_storage_raises_when_none_found():
+    mixin = ProxmoxVMProvisionMixin()
+    mixin._api = MagicMock()
+    mixin._api.storage.get.return_value = [
+        {"storage": "local", "type": "dir", "content": "backup,iso,vztmpl"},
+    ]
+
+    with pytest.raises(ValueError, match="images"):
+        mixin._find_default_image_storage()
+
+
+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.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"},
+    ]
+
+    mock_node = MagicMock()
+    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
+
+    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",
+            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"