napalm-proxmox/tests/test_vm_import_download.py

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