Merge pull request 'feat(vm-provision): let Proxmox download cloud images into an import storage' (#4) from feat/pve-import-download into master

Provisioning downloaded every cloud image over SSH into
/var/lib/vz/template/netork-images, on the node's root filesystem. On a
small root that fills up and takes Proxmox down with it (netOrk #480).

When the node has an active storage with content type "import" (Proxmox
8.2+), Proxmox now does it itself: download-url with checksum
verification into that storage, then import-from as the root disk. The
file is named after a hash of the full URL and reused when present.
Proxmox takes the format from the extension and has no ".img", so
Ubuntu's qcow2 .img is stored as .qcow2 -- a wrong guess fails at import
instead of attaching a qcow2 container as a raw disk.

Without an import storage, or for an image type Proxmox cannot import,
the SSH download is used as before.

CHANGELOG.md

@@ -8,6 +8,14 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 ## [Unreleased]
 
 ### Added
+- Cloud images for `create_vm_from_cloud_init` are downloaded by Proxmox itself
+  when the node has an active storage with content type `import` (Proxmox
+  8.2+): `download-url` with checksum verification, then `import-from` as the
+  root disk. The image no longer passes through the node's root filesystem.
+  Files are named `netork-<url hash>-<name>.<qcow2|raw|vmdk>` and reused;
+  Ubuntu's `.img` is stored as `.qcow2`. Without such a storage, or for an
+  image type Proxmox cannot import, the SSH download into
+  `/var/lib/vz/template/netork-images` is used as before.
 - `HypervisorDriver` contract methods `start_vm`, `stop_vm`, `reboot_vm`,
   `suspend_vm` and `get_vm_config`. They accept a VM's name or vmid, raise
   `ValueError`/`RuntimeError` instead of returning a result dict, and wait

napalm_proxmox/driver.py

@@ -187,30 +187,44 @@ class ProxmoxDriver(
             ) from exc
 
     def _resolve_node(self) -> str:
-        """Resolve the node name from the hostname or optional_args."""
+        """Resolve the PVE node this connection talks to.
+
+        In a cluster, ``GET /nodes`` lists every member, so its first entry is
+        just some node, not necessarily the one at ``self.hostname``. That used
+        to be taken blindly, and a device then reported another node's name and
+        VMs. ``GET /cluster/status`` marks the node the session landed on with
+        ``local: 1``; failing that, the node is matched by IP or name.
+
+        API errors propagate so that ``open()`` fails with the real cause (e.g.
+        a TLS verification error) rather than succeeding on a guessed name.
+        """
         if self._node:
             self._node_name = self._node
             return self._node_name
 
-        # Try the node's hostname via API cluster/resources
-        try:
-            nodes = self._api.nodes.get()
-            # Match by hostname or IP
-            for n in nodes:
-                n_node = n.get("node", "")
-                if n_node:
-                    # First match: the node exists in the cluster
-                    self._node_name = n_node
-                    return self._node_name
-        except Exception:
-            pass
+        members = [
+            s for s in (self._api.cluster.status.get() or []) if s.get("type") == "node"
+        ]
+        short_host = self.hostname.split(".")[0]
+        for pick in (
+            lambda s: s.get("local"),
+            lambda s: s.get("ip") == self.hostname,
+            lambda s: s.get("name") in (self.hostname, short_host),
+        ):
+            match = next((s for s in members if pick(s)), None)
+            if match:
+                self._node_name = match["name"]
+                return self._node_name
 
-        # Fallback: use the configured hostname as node name
-        # (may not match the PVE node name — SSH-based methods will fail,
-        # but API methods that target a specific node name require correct
-        # resolution)
-        self._node_name = self.hostname
-        return self._node_name
+        names = [n.get("node") for n in (self._api.nodes.get() or []) if n.get("node")]
+        if len(names) == 1:
+            self._node_name = names[0]
+            return self._node_name
+
+        raise ConnectionException(
+            f"Cannot tell which PVE node {self.hostname} is among {names}; "
+            f"set the 'node' driver argument"
+        )
 
     def close(self) -> None:
         """Close the connection."""

napalm_proxmox/vm_provision_mixin.py

@@ -5,6 +5,7 @@ from __future__ import annotations
 import base64
 import hashlib
 import logging
+import re
 import time
 import yaml
 from typing import Any, Dict, List
@@ -24,6 +25,43 @@ _logger = logging.getLogger(__name__)
 # download cost once.
 _IMAGE_CACHE_DIR = "/var/lib/vz/template/netork-images"
 
+# Proxmox reads an import volume's format off its extension and accepts only
+# these. Ubuntu ships its qcow2 cloud images as ".img", so that is stored as
+# qcow2: if an .img is really raw, the import fails loudly, whereas guessing
+# raw for a qcow2 would attach the qcow2 container as a raw disk silently.
+_IMPORT_EXTENSIONS = {"qcow2": "qcow2", "raw": "raw", "vmdk": "vmdk", "img": "qcow2"}
+
+# Characters Proxmox keeps in a content file name (PVE::Storage's
+# SAFE_CHAR_CLASS_RE); anything else it would rewrite behind our back.
+_UNSAFE_FILENAME_CHARS = re.compile(r"[^A-Za-z0-9\-.+=_]")
+
+
+def _url_key(image_url: str) -> str:
+    """Short hash of the full URL — Ubuntu and others publish a new build
+    under the same basename every day, so the basename alone is no cache key."""
+    return hashlib.sha256(image_url.encode()).hexdigest()[:12]
+
+
+def _split_checksum(image_checksum: str) -> tuple[str, str]:
+    """``"<algo>:<hex>"`` as ``(algo, hex)``; the algorithm defaults to sha256."""
+    algo, _, expected = image_checksum.partition(":")
+    return (algo or "sha256").lower(), expected
+
+
+def _import_volume_name(image_url: str) -> str | None:
+    """The file name *image_url* gets in an import storage.
+
+    None when Proxmox cannot import the image's type at all (compressed,
+    ISO, no extension) — provisioning then downloads it over SSH as before.
+    """
+    basename = image_url.rstrip("/").rsplit("/", 1)[-1]
+    stem, dot, ext = basename.rpartition(".")
+    extension = _IMPORT_EXTENSIONS.get(ext.lower()) if dot and stem else None
+    if extension is None:
+        return None
+    safe_stem = _UNSAFE_FILENAME_CHARS.sub("_", stem)
+    return f"netork-{_url_key(image_url)}-{safe_stem}.{extension}"
+
 
 class ProxmoxVMProvisionMixin:
     """Mixin to add VM provisioning to ProxmoxDriver."""
@@ -89,11 +127,9 @@ class ProxmoxVMProvisionMixin:
         before raising.
         """
         filename = image_url.rstrip("/").rsplit("/", 1)[-1]
-        url_hash = hashlib.sha256(image_url.encode()).hexdigest()[:12]
-        local_path = f"{_IMAGE_CACHE_DIR}/{url_hash}-{filename}"
+        local_path = f"{_IMAGE_CACHE_DIR}/{_url_key(image_url)}-{filename}"
 
-        algo, _, expected = (image_checksum or "").partition(":")
-        algo = (algo or "sha256").lower()
+        algo, expected = _split_checksum(image_checksum or "")
 
         max_attempts = 2
         for attempt in range(1, max_attempts + 1):
@@ -133,6 +169,125 @@ class ProxmoxVMProvisionMixin:
 
         raise AssertionError("unreachable")  # loop always returns or raises above
 
+    def _find_import_storage(self) -> str | None:
+        """The first storage on this node that accepts content "import".
+
+        Such a storage (Proxmox 8.2+) can take a cloud image straight from its
+        URL. Inactive storages (typically a share that is not mounted) are
+        skipped rather than failed on: the SSH path still works without them.
+        Node-scoped for the same reason as _find_default_image_storage.
+        """
+        for storage in self._node_api().storage.get():
+            content = storage.get("content", "").split(",")
+            if "import" not in content:
+                continue
+            if storage.get("enabled", 1) == 0 or storage.get("active", 1) == 0:
+                continue
+            return storage["storage"]
+        return None
+
+    def _import_cloud_image(
+        self,
+        storage: str,
+        filename: str,
+        image_url: str,
+        image_checksum: str | None,
+        timeout: int,
+    ) -> str:
+        """Have Proxmox download *image_url* into *storage*; returns the volume id.
+
+        Proxmox runs the download as a task and verifies the checksum itself.
+        A file already in the storage is reused — the name carries a hash of
+        the full URL, and Proxmox only creates it once the download (and its
+        checksum check) has succeeded, so an existing file is a complete one.
+        """
+        volid = f"{storage}:import/{filename}"
+        present = self._node_api().storage(storage).content.get(content="import")
+        if any(item.get("volid") == volid for item in present):
+            _logger.info(f"Cloud image already in {storage}: {volid}")
+            return volid
+
+        params: dict[str, Any] = {"url": image_url, "content": "import", "filename": filename}
+        if image_checksum:
+            algo, expected = _split_checksum(image_checksum)
+            params["checksum"] = expected
+            params["checksum-algorithm"] = algo
+        _logger.info(f"Downloading cloud image {image_url} into {volid}")
+        upid = self._node_api().storage(storage)("download-url").post(**params)
+        self._wait_for_task(upid, timeout=timeout)
+        return volid
+
+    def _import_over_ssh(
+        self,
+        vmid: int,
+        image_storage: str,
+        image_url: str,
+        image_checksum: str | None,
+        download_timeout: int,
+        timeout: int,
+    ) -> None:
+        """Download the image on the node and import it as the root disk.
+
+        The path for nodes without an import storage, or for an image type
+        Proxmox cannot import itself.
+        """
+        local_path = self._download_cloud_image(image_url, image_checksum, timeout=download_timeout)
+        _logger.info(f"Importing {local_path} into VM {vmid} on storage {image_storage}")
+        self._run_node_command(
+            f"qm importdisk {vmid} {local_path} {image_storage} --format qcow2",
+            timeout=timeout,
+        )
+
+        # Proxmox leaves the imported disk as an "unusedN" reference — find
+        # it and attach it as the boot disk.
+        imported_config = self._node_api().qemu(vmid).config.get()
+        unused_value = next((v for k, v in imported_config.items() if k.startswith("unused")), None)
+        if not unused_value:
+            raise RuntimeError(
+                f"Disk import for VM {vmid} did not produce an unused disk reference"
+            )
+        self._node_api().qemu(vmid).config.post(
+            scsi0=f"{unused_value},discard=on",
+            boot="order=scsi0",
+        )
+
+    def _attach_root_disk(
+        self,
+        vmid: int,
+        image_storage: str,
+        image_url: str,
+        image_checksum: str | None,
+        download_timeout: int,
+        timeout: int,
+    ) -> None:
+        """Get the cloud image onto the node and make it the VM's root disk.
+
+        Through an import storage when the node has one — Proxmox downloads,
+        verifies and copies the image itself — and over SSH otherwise.
+        """
+        import_storage = self._find_import_storage()
+        filename = _import_volume_name(image_url) if import_storage else None
+        if not import_storage or not filename:
+            self._import_over_ssh(
+                vmid, image_storage, image_url, image_checksum, download_timeout, timeout
+            )
+            return
+
+        volid = self._import_cloud_image(
+            import_storage, filename, image_url, image_checksum, timeout=download_timeout
+        )
+        _logger.info(f"Importing {volid} into VM {vmid} on storage {image_storage}")
+        upid = (
+            self._node_api()
+            .qemu(vmid)
+            .config.post(
+                scsi0=f"{image_storage}:0,import-from={volid},discard=on",
+                boot="order=scsi0",
+            )
+        )
+        if upid:
+            self._wait_for_task(upid, timeout=timeout)
+
     def _find_default_image_storage(self) -> str:
         """Find a storage suitable for VM root disks (content includes 'images').
 
@@ -306,30 +461,9 @@ class ProxmoxVMProvisionMixin:
             )
 
             # Step 3: Download cloud image (cached) and import as root disk
-            local_path = self._download_cloud_image(
-                image_url, image_checksum, timeout=download_timeout
-            )
             image_storage = storage or self._find_default_image_storage()
-
-            _logger.info(f"Importing {local_path} into VM {vmid} on storage {image_storage}")
-            self._run_node_command(
-                f"qm importdisk {vmid} {local_path} {image_storage} --format qcow2",
-                timeout=timeout,
-            )
-
-            # Proxmox leaves the imported disk as an "unusedN" reference — find
-            # it and attach it as the boot disk.
-            imported_config = self._node_api().qemu(vmid).config.get()
-            unused_value = next(
-                (v for k, v in imported_config.items() if k.startswith("unused")), None
-            )
-            if not unused_value:
-                raise RuntimeError(
-                    f"Disk import for VM {vmid} did not produce an unused disk reference"
-                )
-            self._node_api().qemu(vmid).config.post(
-                scsi0=f"{unused_value},discard=on",
-                boot="order=scsi0",
+            self._attach_root_disk(
+                vmid, image_storage, image_url, image_checksum, download_timeout, timeout
             )
 
             # Step 4: Configure network interfaces (CPU/memory already set at shell creation)

tests/conftest.py

@@ -138,6 +138,7 @@ def _build_mock_api(
     pve_users=None,
     exec_return="",
     sensors=None,
+    cluster_status=None,
 ):
     """Build a MagicMock ProxmoxAPI with pre-configured return values."""
     api = MagicMock()
@@ -165,6 +166,7 @@ def _build_mock_api(
     # SDN
     cluster = MagicMock()
     api.cluster = cluster
+    cluster.status.get.return_value = cluster_status or []
     cluster.sdn.zones.get.return_value = sdn_zones or SDN_ZONES
     cluster.sdn.vnets.get.return_value = sdn_vnets or SDN_VNETS
 

tests/test_connection.py

@@ -76,3 +76,78 @@ class TestIsAlive:
     def test_not_alive_when_api_fails(self, driver):
         driver._api.version.get.side_effect = Exception("timeout")
         assert driver.is_alive() == {"is_alive": False}
+
+
+# A four-node cluster as GET /cluster/status reports it. The node the API
+# session landed on carries local=1 — here the third one, so taking the first
+# entry of /nodes (the old behaviour) picks the wrong host.
+CLUSTER_NODES = [
+    {"type": "node", "name": "pve-01", "ip": "172.22.8.101", "local": 0},
+    {"type": "node", "name": "pve-02", "ip": "172.22.8.102", "local": 0},
+    {"type": "node", "name": "pve-dual", "ip": "172.22.8.120", "local": 1},
+    {"type": "node", "name": "pve-garden", "ip": "172.22.8.5", "local": 0},
+]
+CLUSTER_STATUS = [{"type": "cluster", "name": "home", "nodes": 4}, *CLUSTER_NODES]
+CLUSTER_NODES_LIST = [{"node": n["name"], "status": "online"} for n in CLUSTER_NODES]
+
+
+def _open_with(hostname, **api_kwargs):
+    drv = ProxmoxDriver(hostname, "root", "secret")
+    mock_api = _build_mock_api(**api_kwargs)
+    with patch("napalm_proxmox.driver.ProxmoxAPI", return_value=mock_api):
+        drv.open()
+    return drv
+
+
+class TestClusterNodeResolution:
+    def test_local_node_wins_over_first_listed(self):
+        drv = _open_with(
+            "172.22.8.120", nodes=CLUSTER_NODES_LIST, cluster_status=CLUSTER_STATUS
+        )
+        assert drv._node_name == "pve-dual"
+
+    def test_matches_by_ip_without_local_flag(self):
+        status = [{**n, "local": 0} for n in CLUSTER_STATUS if n["type"] == "node"]
+        drv = _open_with("172.22.8.102", nodes=CLUSTER_NODES_LIST, cluster_status=status)
+        assert drv._node_name == "pve-02"
+
+    def test_matches_by_name_without_local_flag(self):
+        status = [{**n, "local": 0} for n in CLUSTER_STATUS if n["type"] == "node"]
+        drv = _open_with("pve-garden", nodes=CLUSTER_NODES_LIST, cluster_status=status)
+        assert drv._node_name == "pve-garden"
+
+    def test_matches_short_name_of_fqdn(self):
+        status = [{**n, "local": 0} for n in CLUSTER_STATUS if n["type"] == "node"]
+        drv = _open_with(
+            "pve-01.mgmt.example.com", nodes=CLUSTER_NODES_LIST, cluster_status=status
+        )
+        assert drv._node_name == "pve-01"
+
+    def test_single_node_without_cluster_status(self):
+        drv = _open_with("10.0.0.9", nodes=[{"node": "solo", "status": "online"}])
+        assert drv._node_name == "solo"
+
+    def test_ambiguous_cluster_refuses_to_guess(self):
+        status = [{**n, "local": 0} for n in CLUSTER_STATUS if n["type"] == "node"]
+        with pytest.raises(ConnectionException, match="node"):
+            _open_with("10.9.9.9", nodes=CLUSTER_NODES_LIST, cluster_status=status)
+
+    def test_api_error_fails_open_instead_of_guessing(self):
+        # A TLS failure used to be swallowed here: the IP became the node
+        # name, open() succeeded and every later call failed quietly.
+        drv = ProxmoxDriver("172.22.8.120", "root", "secret")
+        mock_api = _build_mock_api()
+        mock_api.cluster.status.get.side_effect = Exception("CERTIFICATE_VERIFY_FAILED")
+        mock_api.nodes.get.side_effect = Exception("CERTIFICATE_VERIFY_FAILED")
+        with patch("napalm_proxmox.driver.ProxmoxAPI", return_value=mock_api):
+            with pytest.raises(ConnectionException, match="CERTIFICATE_VERIFY_FAILED"):
+                drv.open()
+
+    def test_explicit_node_skips_lookup(self):
+        drv = ProxmoxDriver("172.22.8.120", "root", "secret", optional_args={"node": "pve-dual"})
+        mock_api = _build_mock_api()
+        with patch("napalm_proxmox.driver.ProxmoxAPI", return_value=mock_api):
+            drv.open()
+        assert drv._node_name == "pve-dual"
+        mock_api.cluster.status.get.assert_not_called()
+        mock_api.nodes.get.assert_not_called()

tests/test_vm_import_download.py

@@ -0,0 +1,241 @@
+"""Tests for provisioning through a storage with content type "import".
+
+Proxmox (8.2+) can download a disk image into such a storage itself
+(``download-url``, with checksum verification) and attach it to a VM with
+``import-from``. When the node has one, provisioning uses it instead of
+downloading over SSH into the node's root filesystem.
+"""
+
+from __future__ import annotations
+
+from unittest.mock import MagicMock, patch
+
+from napalm_proxmox.vm_provision_mixin import (
+    ProxmoxVMProvisionMixin,
+    _import_volume_name,
+)
+
+_UBUNTU = (
+    "https://cloud-images.ubuntu.com/releases/26.04/release/ubuntu-26.04-server-cloudimg-amd64.img"
+)
+_DEBIAN = "https://cloud.debian.org/images/cloud/bookworm/latest/debian-12-genericcloud-amd64.qcow2"
+
+
+def _mixin_with_storages(storages: list[dict]) -> tuple[ProxmoxVMProvisionMixin, MagicMock]:
+    mixin = ProxmoxVMProvisionMixin()
+    mixin._node_name = "pve1"
+    node = MagicMock()
+    node.storage.get.return_value = storages
+    mixin._node_api = MagicMock(return_value=node)
+    return mixin, node
+
+
+# ── which storage ─────────────────────────────────────────────────────────────
+
+
+def test_find_import_storage_picks_a_storage_that_accepts_import():
+    mixin, _ = _mixin_with_storages(
+        [
+            {"storage": "local", "content": "iso,backup,vztmpl,snippets"},
+            {"storage": "software", "content": "import,iso", "active": 1},
+        ]
+    )
+    assert mixin._find_import_storage() == "software"
+
+
+def test_find_import_storage_does_not_mistake_images_for_import():
+    mixin, _ = _mixin_with_storages([{"storage": "local-zfs", "content": "images,rootdir"}])
+    assert mixin._find_import_storage() is None
+
+
+def test_find_import_storage_skips_disabled_and_inactive_storages():
+    """An inactive storage is typically an unmounted share; Proxmox cannot
+    download into it, and the SSH path still works without it."""
+    mixin, _ = _mixin_with_storages(
+        [
+            {"storage": "off", "content": "import", "enabled": 0},
+            {"storage": "unmounted", "content": "import", "active": 0},
+        ]
+    )
+    assert mixin._find_import_storage() is None
+
+
+# ── what the file is called ───────────────────────────────────────────────────
+
+
+def test_import_volume_name_keeps_a_qcow2_extension():
+    name = _import_volume_name(_DEBIAN)
+    assert name is not None
+    assert name.endswith("-debian-12-genericcloud-amd64.qcow2")
+
+
+def test_import_volume_name_stores_an_img_as_qcow2():
+    """Proxmox reads the format off the extension and does not accept .img.
+    Ubuntu's .img is qcow2; guessing qcow2 for a raw .img fails loudly at
+    import, while the opposite guess would import a qcow2 container as a raw
+    disk without complaint."""
+    name = _import_volume_name(_UBUNTU)
+    assert name is not None
+    assert name.endswith("-ubuntu-26.04-server-cloudimg-amd64.qcow2")
+
+
+def test_import_volume_name_is_none_for_types_proxmox_cannot_import():
+    assert _import_volume_name("https://example.org/image.qcow2.xz") is None
+    assert _import_volume_name("https://example.org/installer.iso") is None
+    assert _import_volume_name("https://example.org/noextension") is None
+
+
+def test_import_volume_name_differs_for_the_same_basename_under_another_url():
+    """Ubuntu publishes daily builds under one basename; a cache keyed on the
+    basename alone would serve yesterday's build."""
+    a = _import_volume_name("https://x/release-20260927/ubuntu-26.04-server-cloudimg-amd64.img")
+    b = _import_volume_name("https://x/release-20260928/ubuntu-26.04-server-cloudimg-amd64.img")
+    assert a != b
+
+
+def test_import_volume_name_uses_only_characters_proxmox_keeps():
+    name = _import_volume_name("https://x/My Image (beta)+1.qcow2")
+    assert name is not None
+    assert all(c.isalnum() or c in "-.+=_" for c in name)
+
+
+# ── the download ──────────────────────────────────────────────────────────────
+
+
+def test_import_cloud_image_reuses_a_file_already_in_the_storage():
+    mixin, node = _mixin_with_storages([])
+    name = _import_volume_name(_DEBIAN)
+    node.storage.return_value.content.get.return_value = [
+        {"volid": f"software:import/{name}", "content": "import"}
+    ]
+
+    volid = mixin._import_cloud_image("software", name, _DEBIAN, None, timeout=300)
+
+    assert volid == f"software:import/{name}"
+    node.storage.return_value.return_value.post.assert_not_called()
+
+
+def test_import_cloud_image_has_proxmox_download_and_verify_it():
+    mixin, node = _mixin_with_storages([])
+    name = _import_volume_name(_UBUNTU)
+    node.storage.return_value.content.get.return_value = []
+    download = node.storage.return_value.return_value
+    download.post.return_value = "UPID:pve1:download"
+    mixin._wait_for_task = MagicMock()
+
+    volid = mixin._import_cloud_image("software", name, _UBUNTU, "sha256:abc123", timeout=300)
+
+    assert volid == f"software:import/{name}"
+    node.storage.assert_any_call("software")
+    node.storage.return_value.assert_called_with("download-url")
+    download.post.assert_called_once_with(
+        url=_UBUNTU,
+        content="import",
+        filename=name,
+        checksum="abc123",
+        **{"checksum-algorithm": "sha256"},
+    )
+    mixin._wait_for_task.assert_called_once_with("UPID:pve1:download", timeout=300)
+
+
+def test_import_cloud_image_without_checksum_sends_none():
+    mixin, node = _mixin_with_storages([])
+    name = _import_volume_name(_DEBIAN)
+    node.storage.return_value.content.get.return_value = []
+    mixin._wait_for_task = MagicMock()
+
+    mixin._import_cloud_image("software", name, _DEBIAN, None, timeout=300)
+
+    kwargs = node.storage.return_value.return_value.post.call_args.kwargs
+    assert "checksum" not in kwargs
+    assert "checksum-algorithm" not in kwargs
+
+
+# ── provisioning end to end ───────────────────────────────────────────────────
+
+
+def _provisioning_mixin(storages: list[dict]) -> tuple[ProxmoxVMProvisionMixin, MagicMock]:
+    mixin, node = _mixin_with_storages(storages)
+    api = MagicMock()
+    api.cluster.nextid.get.return_value = 101
+    api.storage.return_value.get.return_value = {"path": "/var/lib/vz"}
+    mixin._api = api
+    mixin._run_node_command = MagicMock(return_value="")
+    mixin._download_cloud_image = MagicMock(return_value="/var/lib/vz/template/x.qcow2")
+    vm = MagicMock()
+    node.qemu.return_value = vm
+    vm.config.get.return_value = {"unused0": "local-zfs:vm-101-disk-0"}
+    vm.config.post.return_value = None
+    node.tasks.return_value.status.get.return_value = {"status": "stopped", "exitstatus": "OK"}
+    node.storage.return_value.content.get.return_value = []
+    return mixin, node
+
+
+def _provision(mixin: ProxmoxVMProvisionMixin, image_url: str) -> None:
+    with patch("time.sleep"):
+        mixin.create_vm_from_cloud_init(
+            name="vm",
+            image_url=image_url,
+            cpu=1,
+            memory=1024,
+            nics=[{"bridge": "vmbr0"}],
+            cloud_init_config={"hostname": "vm"},
+            storage="local-zfs",
+            timeout=60,
+        )
+
+
+_WITH_IMPORT = [
+    {"storage": "local", "content": "iso,backup,vztmpl,snippets"},
+    {"storage": "software", "content": "import,iso"},
+    {"storage": "local-zfs", "content": "images,rootdir"},
+]
+
+
+def test_provisioning_downloads_through_the_import_storage_when_there_is_one():
+    mixin, node = _provisioning_mixin(_WITH_IMPORT)
+    name = _import_volume_name(_UBUNTU)
+
+    _provision(mixin, _UBUNTU)
+
+    mixin._download_cloud_image.assert_not_called()
+    assert not any("importdisk" in c.args[0] for c in mixin._run_node_command.call_args_list), (
+        "no SSH download/import when Proxmox can do it"
+    )
+    disk = next(
+        c.kwargs for c in node.qemu.return_value.config.post.call_args_list if "scsi0" in c.kwargs
+    )
+    assert disk["scsi0"] == f"local-zfs:0,import-from=software:import/{name},discard=on"
+    assert disk["boot"] == "order=scsi0"
+
+
+def test_provisioning_waits_for_the_import_task():
+    """Attaching with import-from copies the image — a task, which has to
+    finish before the cloud-init drive and the resize touch the VM."""
+    mixin, node = _provisioning_mixin(_WITH_IMPORT)
+    node.qemu.return_value.config.post.side_effect = lambda **kw: (
+        "UPID:pve1:import" if "scsi0" in kw else None
+    )
+    mixin._wait_for_task = MagicMock()
+
+    _provision(mixin, _DEBIAN)
+
+    waited = [c.args[0] for c in mixin._wait_for_task.call_args_list]
+    assert "UPID:pve1:import" in waited
+
+
+def test_provisioning_falls_back_to_ssh_without_an_import_storage():
+    mixin, _ = _provisioning_mixin([s for s in _WITH_IMPORT if s["storage"] != "software"])
+
+    _provision(mixin, _UBUNTU)
+
+    mixin._download_cloud_image.assert_called_once()
+    assert any("qm importdisk 101" in c.args[0] for c in mixin._run_node_command.call_args_list)
+
+
+def test_provisioning_falls_back_to_ssh_for_an_image_type_proxmox_cannot_import():
+    mixin, _ = _provisioning_mixin(_WITH_IMPORT)
+
+    _provision(mixin, "https://example.org/image.qcow2.xz")
+
+    mixin._download_cloud_image.assert_called_once()