NAPALM/napalm-proxmox
1
0
Fork 0
Code Issues 1 Pull Requests 1 Actions Packages Projects Releases Wiki Activity
Files
12135735cb863d2968d3245fe2273e7f09a13c2e
napalm-proxmox/tests/test_vm_provision_mixin.py
T
Christian Manivong 12135735cb 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.
2026-07-07 12:12:44 +02:00

742 lines
27 KiB
Python
Raw Blame History

"""Tests for ProxmoxVMProvisionMixin — VM creation, destruction, status polling."""
from __future__ import annotations
import pytest
from unittest.mock import MagicMock, patch
from napalm_proxmox.driver import ProxmoxDriver
from napalm_proxmox.vm_provision_mixin import ProxmoxVMProvisionMixin
# ---------------------------------------------------------------------------
# Tests
# ---------------------------------------------------------------------------
def test_wait_for_task_success():
"""_wait_for_task succeeds when task reaches stopped status with exitstatus OK."""
mixin = ProxmoxVMProvisionMixin()
# Mock the _node_api() to return a mock that supports task polling
mock_node = MagicMock()
mock_task_status = MagicMock()
mock_task_status.status.get.side_effect = [
{"status": "running", "exitstatus": None},
{"status": "stopped", "exitstatus": "OK"},
]
mock_node.tasks.return_value = mock_task_status
mixin._node_api = MagicMock(return_value=mock_node)
# Should complete without raising
with patch("time.sleep"): # Speed up polling
result = mixin._wait_for_task("UPID:pve1:123:456:789:clone:100:root@pam:", timeout=30)
assert result is None
def test_wait_for_task_error():
"""_wait_for_task raises RuntimeError when task exits with non-OK status."""
mixin = ProxmoxVMProvisionMixin()
mock_node = MagicMock()
mock_task_status = MagicMock()
mock_task_status.status.get.return_value = {"status": "stopped", "exitstatus": "FAILED"}
mock_node.tasks.return_value = mock_task_status
mixin._node_api = MagicMock(return_value=mock_node)
with pytest.raises(RuntimeError, match="FAILED"):
with patch("time.sleep"):
mixin._wait_for_task("UPID:pve1:123:456:789:clone:100:root@pam:", timeout=30)
def test_wait_for_task_timeout():
"""_wait_for_task raises RuntimeError on timeout."""
mixin = ProxmoxVMProvisionMixin()
mock_node = MagicMock()
mock_task_status = MagicMock()
mock_task_status.status.get.return_value = {"status": "running"} # Always running
mock_node.tasks.return_value = mock_task_status
mixin._node_api = MagicMock(return_value=mock_node)
with pytest.raises(RuntimeError, match="timed out"):
with patch("napalm_proxmox.vm_provision_mixin.time.time") as mock_time:
# Simulate time passing: return incremented values to exceed timeout quickly
mock_time.side_effect = [0, 2, 4] # After 2 iterations, time > timeout=1
with patch("napalm_proxmox.vm_provision_mixin.time.sleep"):
mixin._wait_for_task("UPID:pve1:123:456:789:clone:100:root@pam:", timeout=1)
def test_create_vm_from_cloud_init_single_nic():
"""create_vm_from_cloud_init with single NIC, DHCP enabled (generic happy path)."""
mixin = ProxmoxVMProvisionMixin()
mixin._node_name = "pve1"
# Mock API hierarchy
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},
{"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_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",
"scsi0": "local-lvm:vm-101-disk-0",
}
mock_vm.status.start.post.return_value = "UPID:pve1:124:start"
# Mock task completion
mock_task = MagicMock()
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",
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}],
cloud_init_config={"hostname": "test-vm"},
timeout=120,
)
assert result["vmid"] == "101"
assert result["name"] == "test-vm"
assert result["node"] == "pve1"
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
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():
"""create_vm_from_cloud_init with dual NICs: net0 DHCP + net1 trunk (no DHCP)."""
mixin = ProxmoxVMProvisionMixin()
mixin._node_name = "pve1"
# Mock API hierarchy
mock_api = MagicMock()
mock_api.cluster.nextid.get.return_value = 102
mock_api.storage.get.return_value = [
{"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_node.qemu.post.return_value = None
mock_vm.config.post.return_value = None
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
mock_task = MagicMock()
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",
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
],
cloud_init_config={"hostname": "sat-1", "runcmd": ["custom cmd"]},
timeout=120,
)
assert result["vmid"] == "102"
assert result["name"] == "wireshark-sat-1"
# Verify both NICs configured
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 = 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
def test_create_vm_missing_snippet_storage():
"""create_vm_from_cloud_init raises ValueError if snippet storage unavailable."""
mixin = ProxmoxVMProvisionMixin()
mixin._node_name = "pve1"
# 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 = 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_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()
mock_task.status.get.return_value = {"status": "stopped", "exitstatus": "OK"}
mock_node.tasks.return_value = mock_task
with pytest.raises(ValueError, match="snippet"):
with patch("napalm_proxmox.vm_provision_mixin.time.sleep"):
mixin.create_vm_from_cloud_init(
name="test-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={},
)
def test_create_vm_with_disk_resize():
"""create_vm_from_cloud_init resizes disk when disk_resize_gb is set."""
mixin = ProxmoxVMProvisionMixin()
mixin._node_name = "pve1"
# Mock API hierarchy
mock_api = MagicMock()
mock_api.cluster.nextid.get.return_value = 103
mock_api.storage.get.return_value = [
{"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_node.qemu.post.return_value = None
mock_vm.config.post.return_value = None
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"
# Mock task completion
mock_task = MagicMock()
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",
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": "big-vm"},
disk_resize_gb=100,
timeout=120,
)
assert result["vmid"] == "103"
# Verify resize was called
mock_vm.resize.put.assert_called_once()
call_kwargs = mock_vm.resize.put.call_args[1]
assert call_kwargs["disk"] == "scsi0"
assert call_kwargs["size"] == "100G"
def test_get_vm_status_with_wait_for_ip():
"""get_vm_status(wait_for_ip=True) polls guest-agent until IP acquired."""
mixin = ProxmoxVMProvisionMixin()
mock_node = MagicMock()
mixin._node_api = MagicMock(return_value=mock_node)
# Mock VM config
mock_vm = MagicMock()
mock_node.qemu.return_value = mock_vm
mock_vm.config.get.return_value = {"net0": "virtio,bridge=vmbr0,tag=10"}
# Mock guest-agent: first no IP, then with IP
mock_agent = MagicMock()
mock_agent.network_get_interfaces.get.side_effect = [
{"result": [{"name": "eth0", "hardware-address": "aa:bb:cc:dd:ee:00"}]},
{
"result": [
{
"name": "eth0",
"hardware-address": "aa:bb:cc:dd:ee:00",
"ip-addresses": [{"ip-address": "10.0.0.100", "ip-address-type": "ipv4"}],
}
]
},
]
mock_vm.agent = mock_agent
with patch("time.sleep"):
result = mixin.get_vm_status("101", wait_for_ip=True, timeout=30, poll_interval=1)
assert result["status"] == "running"
assert result["ip_address"] == "10.0.0.100"
assert result["mac_address"] == "aa:bb:cc:dd:ee:00"
def test_get_vm_status_timeout_waiting_for_ip():
"""get_vm_status raises RuntimeError if IP acquisition times out."""
mixin = ProxmoxVMProvisionMixin()
mock_node = MagicMock()
mixin._node_api = MagicMock(return_value=mock_node)
# Mock VM config
mock_vm = MagicMock()
mock_node.qemu.return_value = mock_vm
mock_vm.config.get.return_value = {"net0": "virtio,bridge=vmbr0"}
# Mock guest-agent that never returns IP
mock_agent = MagicMock()
mock_agent.network_get_interfaces.get.return_value = {
"result": [{"name": "eth0", "hardware-address": "aa:bb:cc:dd:ee:00"}]
}
mock_vm.agent = mock_agent
with pytest.raises(RuntimeError, match="timeout|IP"):
with patch("time.sleep"):
mixin.get_vm_status("101", wait_for_ip=True, timeout=1, poll_interval=0.5)
def test_destroy_vm_success():
"""destroy_vm stops running VM, deletes it, cleans up snippets."""
mixin = ProxmoxVMProvisionMixin()
mock_node = MagicMock()
mixin._node_api = MagicMock(return_value=mock_node)
# Mock VM operations
mock_vm = MagicMock()
mock_node.qemu.return_value = mock_vm
mock_vm.config.get.return_value = {"cicustom": "user=snippets:snippets/101-user-data.yaml"}
mock_vm.status.stop.post.return_value = "UPID:pve1:125:stop"
mock_vm.delete.return_value = None
# Mock task completion
mock_task = MagicMock()
mock_task.status.get.return_value = {"status": "stopped", "exitstatus": "OK"}
mock_node.tasks.return_value = mock_task
# Mock snippet deletion (best-effort)
mock_storage = MagicMock()
mock_node.storage.return_value = mock_storage
mock_content = MagicMock()
mock_storage.content.return_value = mock_content
mock_content.delete.return_value = None
with patch("time.sleep"):
mixin.destroy_vm("101", remove_disk=True, timeout=60)
assert mock_vm.delete.called
def test_destroy_vm_already_stopped():
"""destroy_vm succeeds even if VM already stopped."""
mixin = ProxmoxVMProvisionMixin()
mock_node = MagicMock()
mixin._node_api = MagicMock(return_value=mock_node)
# Mock VM operations
mock_vm = MagicMock()
mock_node.qemu.return_value = mock_vm
mock_vm.config.get.return_value = {}
mock_vm.delete.return_value = None
with patch("time.sleep"):
mixin.destroy_vm("101", remove_disk=True, timeout=60)
# Verify delete was called (stop may fail or not be needed)
assert mock_vm.delete.called
# ---------------------------------------------------------------------------
# get_network_targets
# ---------------------------------------------------------------------------
def test_get_network_targets_filters_to_bridges_and_vnets():
"""get_network_targets excludes physical NICs/bonds; includes bridges + vnets."""
mixin = ProxmoxVMProvisionMixin()
mixin._get_node_network = MagicMock(
return_value=[
{"iface": "eth0", "type": "eth"},
{"iface": "bond0", "type": "bond"},
{"iface": "vmbr0", "type": "bridge"},
{"iface": "vmbr1", "type": "OVSBridge"},
]
)
mixin._get_sdn_vnets = MagicMock(
return_value=[{"vnet": "vnet0", "zone": "zone-vlan", "tag": 10}]
)
targets = mixin.get_network_targets()
names = {t["name"] for t in targets}
assert names == {"vmbr0", "vmbr1", "vnet0"}
assert "eth0" not in names
assert "bond0" not in names
def test_get_network_targets_linux_bridge_vlan_aware_flag():
"""A Linux bridge's vlan_aware reflects its bridge_vlan_aware config flag."""
mixin = ProxmoxVMProvisionMixin()
mixin._get_node_network = MagicMock(
return_value=[
{"iface": "vmbr0", "type": "bridge", "bridge_vlan_aware": 1},
{"iface": "vmbr2", "type": "bridge"}, # no flag -> not vlan aware
]
)
mixin._get_sdn_vnets = MagicMock(return_value=[])
targets = {t["name"]: t for t in mixin.get_network_targets()}
assert targets["vmbr0"]["kind"] == "bridge"
assert targets["vmbr0"]["vlan_aware"] is True
assert targets["vmbr2"]["vlan_aware"] is False
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(
return_value=[{"iface": "vmbr1", "type": "OVSBridge"}]
)
mixin._get_sdn_vnets = MagicMock(return_value=[])
targets = mixin.get_network_targets()
assert targets[0]["kind"] == "bridge"
assert targets[0]["vlan_aware"] is True
def test_get_network_targets_sdn_vnet_never_vlan_aware():
"""An SDN vnet is never vlan_aware — its VLAN is already fixed by zone/tag."""
mixin = ProxmoxVMProvisionMixin()
mixin._get_node_network = MagicMock(return_value=[])
mixin._get_sdn_vnets = MagicMock(
return_value=[
{"vnet": "vnet0", "zone": "zone-vlan", "tag": 10},
{"vnet": "vnet1", "zone": "zone-vlan", "tag": 20},
]
)
targets = mixin.get_network_targets()
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()
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"
Reference in New Issue View Git Blame Copy Permalink