Christian Manivong
1d6aabb5a1
nics[i]['mac'] is set via virtio=<mac>,bridge=... instead of the bare virtio,bridge=... form, so a caller-supplied MAC actually takes effect (needed for DHCP reservations created before the VM exists).
1092 lines
41 KiB
Python
1092 lines
41 KiB
Python
"""Tests for ProxmoxVMProvisionMixin — VM creation, destruction, status polling."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import base64
|
|
|
|
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.return_value.get.return_value = {"path": "/var/lib/vz"}
|
|
|
|
mock_node = MagicMock()
|
|
mock_node.storage.get.return_value = [
|
|
{"storage": "local-lvm", "type": "lvmthin", "content": "images,rootdir", "enabled": 1},
|
|
{"storage": "snippets", "type": "dir", "content": "snippets", "enabled": 1},
|
|
]
|
|
mixin._api = mock_api
|
|
mixin._node_api = MagicMock(return_value=mock_node)
|
|
mixin._download_cloud_image = MagicMock(return_value="/var/lib/vz/template/netork-images/debian-12.qcow2")
|
|
mixin._run_node_command = MagicMock(return_value="")
|
|
|
|
# Mock VM 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
|
|
|
|
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"]
|
|
|
|
# Regression: the snippet must be written directly to the filesystem via
|
|
# SSH, not uploaded via the /storage/upload API — real Proxmox rejects
|
|
# content='snippets' on that endpoint outright (see
|
|
# test_create_vm_writes_snippet_via_ssh_with_correct_content for the
|
|
# dedicated check).
|
|
write_cmd = next(
|
|
c[0][0] for c in mixin._run_node_command.call_args_list if "snippets/101-user-data.yaml" in c[0][0]
|
|
)
|
|
assert "/var/lib/vz/snippets" in write_cmd
|
|
|
|
|
|
def test_create_vm_from_cloud_init_dual_nic_trunk():
|
|
"""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.return_value.get.return_value = {"path": "/var/lib/vz"}
|
|
|
|
mock_node = MagicMock()
|
|
mock_node.storage.get.return_value = [
|
|
{"storage": "local-lvm", "type": "lvmthin", "content": "images,rootdir", "enabled": 1},
|
|
{"storage": "snippets", "type": "dir", "content": "snippets", "enabled": 1},
|
|
]
|
|
mixin._api = mock_api
|
|
mixin._node_api = MagicMock(return_value=mock_node)
|
|
mixin._download_cloud_image = MagicMock(return_value="/var/lib/vz/template/netork-images/debian-12.qcow2")
|
|
mixin._run_node_command = MagicMock(return_value="")
|
|
|
|
# Mock VM 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
|
|
|
|
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
|
|
|
|
mixin._api = mock_api
|
|
|
|
mock_node = MagicMock()
|
|
mock_node.storage.get.return_value = [
|
|
{"storage": "local-lvm", "type": "lvmthin", "content": "images,rootdir", "enabled": 1}
|
|
]
|
|
mixin._node_api = MagicMock(return_value=mock_node)
|
|
mixin._download_cloud_image = MagicMock(return_value="/var/lib/vz/template/netork-images/debian-12.qcow2")
|
|
mixin._run_node_command = MagicMock(return_value="")
|
|
mock_vm = MagicMock()
|
|
mock_node.qemu.return_value = mock_vm
|
|
mock_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.return_value.get.return_value = {"path": "/var/lib/vz"}
|
|
|
|
mock_node = MagicMock()
|
|
mock_node.storage.get.return_value = [
|
|
{"storage": "local-lvm", "type": "lvmthin", "content": "images,rootdir", "enabled": 1},
|
|
{"storage": "snippets", "type": "dir", "content": "snippets", "enabled": 1},
|
|
]
|
|
mixin._api = mock_api
|
|
mixin._node_api = MagicMock(return_value=mock_node)
|
|
mixin._download_cloud_image = MagicMock(return_value="/var/lib/vz/template/netork-images/debian-12.qcow2")
|
|
mixin._run_node_command = MagicMock(return_value="")
|
|
|
|
# Mock VM 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
|
|
|
|
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()
|
|
mock_node = MagicMock()
|
|
mock_node.storage.get.return_value = [
|
|
{"storage": "local-lvm", "type": "lvmthin", "content": "images,rootdir"},
|
|
]
|
|
mixin._node_api = MagicMock(return_value=mock_node)
|
|
|
|
storage = mixin._find_default_image_storage()
|
|
|
|
assert storage == "local-lvm"
|
|
|
|
|
|
def test_find_default_image_storage_excludes_explicitly_disabled():
|
|
mixin = ProxmoxVMProvisionMixin()
|
|
mock_node = MagicMock()
|
|
mock_node.storage.get.return_value = [
|
|
{"storage": "old-storage", "type": "dir", "content": "images", "enabled": 0},
|
|
{"storage": "local-lvm", "type": "lvmthin", "content": "images,rootdir", "enabled": 1},
|
|
]
|
|
mixin._node_api = MagicMock(return_value=mock_node)
|
|
|
|
storage = mixin._find_default_image_storage()
|
|
|
|
assert storage == "local-lvm"
|
|
|
|
|
|
def test_find_default_image_storage_raises_when_none_found():
|
|
mixin = ProxmoxVMProvisionMixin()
|
|
mock_node = MagicMock()
|
|
mock_node.storage.get.return_value = [
|
|
{"storage": "local", "type": "dir", "content": "backup,iso,vztmpl"},
|
|
]
|
|
mixin._node_api = MagicMock(return_value=mock_node)
|
|
|
|
with pytest.raises(ValueError, match="images"):
|
|
mixin._find_default_image_storage()
|
|
|
|
|
|
def test_find_default_image_storage_excludes_storage_restricted_to_other_nodes():
|
|
"""Regression: a storage configured cluster-wide but restricted via 'nodes'
|
|
to other cluster members must not be picked — /nodes/{node}/storage (unlike
|
|
the cluster-wide /storage endpoint) only lists what's actually available on
|
|
this node, so querying it is what makes this exclusion happen naturally.
|
|
This was the real-world bug: local-lvm (nodes=pve-garden) was picked for a
|
|
VM on pve-02, and 'qm importdisk' failed with 'storage not available on
|
|
node', leaving a VM shell with no disk attached."""
|
|
mixin = ProxmoxVMProvisionMixin()
|
|
mock_node = MagicMock()
|
|
# /nodes/pve-02/storage never even lists local-lvm — Proxmox itself filters
|
|
# node-restricted storages out of this endpoint.
|
|
mock_node.storage.get.return_value = [
|
|
{"storage": "local-zfs", "type": "zfspool", "content": "images,rootdir"},
|
|
]
|
|
mixin._node_api = MagicMock(return_value=mock_node)
|
|
|
|
storage = mixin._find_default_image_storage()
|
|
|
|
assert storage == "local-zfs"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# _get_storage_path
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_get_storage_path_returns_path_from_cluster_config():
|
|
mixin = ProxmoxVMProvisionMixin()
|
|
mixin._api = MagicMock()
|
|
mixin._api.storage.return_value.get.return_value = {
|
|
"storage": "local",
|
|
"type": "dir",
|
|
"path": "/var/lib/vz",
|
|
}
|
|
|
|
path = mixin._get_storage_path("local")
|
|
|
|
assert path == "/var/lib/vz"
|
|
mixin._api.storage.assert_called_with("local")
|
|
|
|
|
|
def test_get_storage_path_raises_when_storage_has_no_path():
|
|
"""A storage type without a filesystem path (e.g. lvmthin, zfspool) can't
|
|
back content='snippets' at all — Proxmox itself only allows that content
|
|
type on dir/nfs/cifs/cephfs storages, so this should never actually be
|
|
reached for a real snippet storage, but must fail clearly if it is."""
|
|
mixin = ProxmoxVMProvisionMixin()
|
|
mixin._api = MagicMock()
|
|
mixin._api.storage.return_value.get.return_value = {
|
|
"storage": "local-lvm",
|
|
"type": "lvmthin",
|
|
}
|
|
|
|
with pytest.raises(ValueError, match="path"):
|
|
mixin._get_storage_path("local-lvm")
|
|
|
|
|
|
def test_create_vm_from_cloud_init_with_real_world_storage_shape():
|
|
"""Regression: real Proxmox servers omit "enabled" for never-toggled storages
|
|
(observed live: local-lvm/local-zfs/fast-zfs all had content=images but no
|
|
"enabled" key at all) — this used to make create_vm_from_cloud_init fail
|
|
with "No storage with content='images' found" even though usable storage
|
|
existed."""
|
|
mixin = ProxmoxVMProvisionMixin()
|
|
mixin._node_name = "pve1"
|
|
|
|
mock_api = MagicMock()
|
|
mock_api.cluster.nextid.get.return_value = 104
|
|
mock_api.storage.return_value.get.return_value = {"path": "/var/lib/vz"}
|
|
|
|
mock_node = MagicMock()
|
|
mock_node.storage.get.return_value = [
|
|
{"storage": "local-lvm", "type": "lvmthin", "content": "images,rootdir"},
|
|
{"storage": "local-zfs", "type": "zfspool", "content": "rootdir,images"},
|
|
{"storage": "local", "type": "dir", "content": "backup,iso,vztmpl"},
|
|
{"storage": "snippets", "type": "dir", "content": "snippets"},
|
|
]
|
|
mixin._api = mock_api
|
|
mixin._node_api = MagicMock(return_value=mock_node)
|
|
mixin._download_cloud_image = MagicMock(return_value="/var/lib/vz/template/netork-images/debian-12.qcow2")
|
|
mixin._run_node_command = MagicMock(return_value="")
|
|
|
|
mock_vm = MagicMock()
|
|
mock_node.qemu.return_value = mock_vm
|
|
mock_node.qemu.post.return_value = None
|
|
mock_vm.config.post.return_value = None
|
|
mock_vm.config.get.return_value = {
|
|
"unused0": "local-lvm:vm-104-disk-0",
|
|
"scsi0": "local-lvm:vm-104-disk-0",
|
|
}
|
|
mock_vm.status.start.post.return_value = "UPID:pve1:130:start"
|
|
|
|
mock_task = MagicMock()
|
|
mock_task.status.get.return_value = {"status": "stopped", "exitstatus": "OK"}
|
|
mock_node.tasks.return_value = mock_task
|
|
|
|
with patch("time.sleep"):
|
|
result = mixin.create_vm_from_cloud_init(
|
|
name="real-shape-vm",
|
|
image_url="https://cloud.debian.org/images/cloud/bookworm/latest/debian-12-genericcloud-amd64.qcow2",
|
|
cpu=2,
|
|
memory=2048,
|
|
nics=[{"bridge": "vmbr0"}],
|
|
cloud_init_config={"hostname": "real-shape-vm"},
|
|
)
|
|
|
|
assert result["vmid"] == "104"
|
|
|
|
|
|
def test_create_vm_from_cloud_init_explicit_storage_skips_auto_detect():
|
|
"""When storage= is given explicitly, it's used directly and
|
|
_find_default_image_storage is never consulted (so an explicit choice
|
|
always wins, even if auto-detect would have picked something else)."""
|
|
mixin = ProxmoxVMProvisionMixin()
|
|
mixin._node_name = "pve1"
|
|
|
|
mock_api = MagicMock()
|
|
mock_api.cluster.nextid.get.return_value = 105
|
|
mock_api.storage.return_value.get.return_value = {"path": "/var/lib/vz"}
|
|
|
|
mock_node = MagicMock()
|
|
mock_node.storage.get.return_value = [
|
|
{"storage": "local-lvm", "type": "lvmthin", "content": "images,rootdir", "enabled": 1},
|
|
{"storage": "fast-zfs", "type": "zfspool", "content": "images,rootdir", "enabled": 1},
|
|
{"storage": "snippets", "type": "dir", "content": "snippets", "enabled": 1},
|
|
]
|
|
mixin._api = mock_api
|
|
mixin._node_api = MagicMock(return_value=mock_node)
|
|
mixin._download_cloud_image = MagicMock(return_value="/var/lib/vz/template/netork-images/debian-12.qcow2")
|
|
mixin._find_default_image_storage = MagicMock(side_effect=AssertionError("should not be called"))
|
|
mixin._run_node_command = MagicMock(return_value="")
|
|
|
|
mock_vm = MagicMock()
|
|
mock_node.qemu.return_value = mock_vm
|
|
mock_node.qemu.post.return_value = None
|
|
mock_vm.config.post.return_value = None
|
|
mock_vm.config.get.return_value = {
|
|
"unused0": "fast-zfs:vm-105-disk-0",
|
|
"scsi0": "fast-zfs:vm-105-disk-0",
|
|
}
|
|
mock_vm.status.start.post.return_value = "UPID:pve1:131:start"
|
|
|
|
mock_task = MagicMock()
|
|
mock_task.status.get.return_value = {"status": "stopped", "exitstatus": "OK"}
|
|
mock_node.tasks.return_value = mock_task
|
|
|
|
with patch("time.sleep"):
|
|
result = mixin.create_vm_from_cloud_init(
|
|
name="explicit-storage-vm",
|
|
image_url="https://cloud.debian.org/images/cloud/bookworm/latest/debian-12-genericcloud-amd64.qcow2",
|
|
cpu=2,
|
|
memory=2048,
|
|
nics=[{"bridge": "vmbr0"}],
|
|
cloud_init_config={"hostname": "explicit-storage-vm"},
|
|
storage="fast-zfs",
|
|
)
|
|
|
|
assert result["vmid"] == "105"
|
|
mixin._find_default_image_storage.assert_not_called()
|
|
import_cmd = next(
|
|
c[0][0] for c in mixin._run_node_command.call_args_list if "qm importdisk" in c[0][0]
|
|
)
|
|
assert "fast-zfs" in import_cmd
|
|
assert "local-lvm" not in import_cmd
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# get_image_storages
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_get_image_storages_filters_to_images_content():
|
|
mixin = ProxmoxVMProvisionMixin()
|
|
mock_node = MagicMock()
|
|
mock_node.storage.get.return_value = [
|
|
{
|
|
"storage": "local-zfs",
|
|
"type": "zfspool",
|
|
"content": "images,rootdir",
|
|
"total": 100 * 1024**3,
|
|
"avail": 40 * 1024**3,
|
|
},
|
|
{"storage": "local", "type": "dir", "content": "backup,iso,vztmpl,snippets"},
|
|
]
|
|
mixin._node_api = MagicMock(return_value=mock_node)
|
|
|
|
targets = mixin.get_image_storages()
|
|
|
|
assert len(targets) == 1
|
|
assert targets[0]["name"] == "local-zfs"
|
|
assert targets[0]["type"] == "zfspool"
|
|
assert targets[0]["total_gb"] == 100.0
|
|
assert targets[0]["available_gb"] == 40.0
|
|
|
|
|
|
def test_get_image_storages_excludes_disabled_and_inactive():
|
|
mixin = ProxmoxVMProvisionMixin()
|
|
mock_node = MagicMock()
|
|
mock_node.storage.get.return_value = [
|
|
{"storage": "disabled-store", "type": "dir", "content": "images", "enabled": 0},
|
|
{"storage": "inactive-store", "type": "nfs", "content": "images", "active": 0},
|
|
{"storage": "ok-store", "type": "dir", "content": "images"},
|
|
]
|
|
mixin._node_api = MagicMock(return_value=mock_node)
|
|
|
|
targets = mixin.get_image_storages()
|
|
|
|
assert [t["name"] for t in targets] == ["ok-store"]
|
|
|
|
|
|
def test_get_image_storages_excludes_storage_restricted_to_other_nodes():
|
|
"""Node-scoped query naturally excludes storages Proxmox itself doesn't
|
|
list for this node (e.g. restricted via 'nodes' to other cluster members)."""
|
|
mixin = ProxmoxVMProvisionMixin()
|
|
mock_node = MagicMock()
|
|
mock_node.storage.get.return_value = [
|
|
{"storage": "local-zfs", "type": "zfspool", "content": "images,rootdir"},
|
|
]
|
|
mixin._node_api = MagicMock(return_value=mock_node)
|
|
|
|
targets = mixin.get_image_storages()
|
|
|
|
assert [t["name"] for t in targets] == ["local-zfs"]
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Snippet upload — must be a real multipart file, not a filename+data string pair
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_create_vm_writes_snippet_via_ssh_with_correct_content():
|
|
"""Regression: real Proxmox's /storage/{s}/upload endpoint rejects
|
|
content='snippets' outright ("value 'snippets' does not have a value in
|
|
the enumeration 'iso, vztmpl, import'") — that endpoint only handles
|
|
ISOs, container templates, and imports. Snippets can only be written
|
|
directly to the storage's filesystem path, so this must go through SSH
|
|
(_run_node_command), not the upload API.
|
|
(observed live: 400 Bad Request from Proxmox, right after the earlier
|
|
multipart-upload fix had already gotten past a prior RemoteDisconnected
|
|
bug at the same step — two distinct real-world failures at this line)."""
|
|
mixin = ProxmoxVMProvisionMixin()
|
|
mixin._node_name = "pve1"
|
|
|
|
mock_api = MagicMock()
|
|
mock_api.cluster.nextid.get.return_value = 106
|
|
mock_api.storage.return_value.get.return_value = {"path": "/mnt/pve/snippet-nfs"}
|
|
|
|
mock_node = MagicMock()
|
|
mock_node.storage.get.return_value = [
|
|
{"storage": "local-zfs", "type": "zfspool", "content": "images,rootdir", "enabled": 1},
|
|
{"storage": "local", "type": "dir", "content": "snippets", "enabled": 1},
|
|
]
|
|
mixin._api = mock_api
|
|
mixin._node_api = MagicMock(return_value=mock_node)
|
|
mixin._download_cloud_image = MagicMock(return_value="/var/lib/vz/template/netork-images/debian-12.qcow2")
|
|
mixin._run_node_command = MagicMock(return_value="")
|
|
|
|
mock_vm = MagicMock()
|
|
mock_node.qemu.return_value = mock_vm
|
|
mock_node.qemu.post.return_value = None
|
|
mock_vm.config.post.return_value = None
|
|
mock_vm.config.get.return_value = {
|
|
"unused0": "local-zfs:vm-106-disk-0",
|
|
"scsi0": "local-zfs:vm-106-disk-0",
|
|
}
|
|
mock_vm.status.start.post.return_value = "UPID:pve1:132:start"
|
|
|
|
mock_task = MagicMock()
|
|
mock_task.status.get.return_value = {"status": "stopped", "exitstatus": "OK"}
|
|
mock_node.tasks.return_value = mock_task
|
|
|
|
with patch("time.sleep"):
|
|
mixin.create_vm_from_cloud_init(
|
|
name="write-shape-vm",
|
|
image_url="https://cloud.debian.org/images/cloud/bookworm/latest/debian-12-genericcloud-amd64.qcow2",
|
|
cpu=2,
|
|
memory=2048,
|
|
nics=[{"bridge": "vmbr0"}],
|
|
cloud_init_config={"hostname": "write-shape-vm", "chpasswd": {"expire": False}},
|
|
)
|
|
|
|
# No call to the upload API at all — snippets aren't a valid content
|
|
# type there.
|
|
mock_node.storage.assert_not_called()
|
|
|
|
write_cmd = next(
|
|
c[0][0]
|
|
for c in mixin._run_node_command.call_args_list
|
|
if "snippets/106-user-data.yaml" in c[0][0]
|
|
)
|
|
assert "mkdir -p /mnt/pve/snippet-nfs/snippets" in write_cmd
|
|
assert "/mnt/pve/snippet-nfs/snippets/106-user-data.yaml" in write_cmd
|
|
|
|
encoded = write_cmd.split("echo ", 1)[1].split(" | base64 -d")[0]
|
|
content = base64.b64decode(encoded).decode("utf-8")
|
|
assert content.startswith("#cloud-config\n")
|
|
assert "hostname: write-shape-vm" in content
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Explicit NIC MAC address
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_create_vm_from_cloud_init_with_explicit_mac():
|
|
"""When nics[i]['mac'] is set, it's pinned via virtio=<mac>,bridge=...
|
|
(needed so a caller can create a matching DHCP reservation before the
|
|
VM even boots)."""
|
|
mixin = ProxmoxVMProvisionMixin()
|
|
mixin._node_name = "pve1"
|
|
|
|
mock_api = MagicMock()
|
|
mock_api.cluster.nextid.get.return_value = 107
|
|
mock_api.storage.return_value.get.return_value = {"path": "/var/lib/vz"}
|
|
|
|
mock_node = MagicMock()
|
|
mock_node.storage.get.return_value = [
|
|
{"storage": "local-lvm", "type": "lvmthin", "content": "images,rootdir", "enabled": 1},
|
|
{"storage": "snippets", "type": "dir", "content": "snippets", "enabled": 1},
|
|
]
|
|
mixin._api = mock_api
|
|
mixin._node_api = MagicMock(return_value=mock_node)
|
|
mixin._download_cloud_image = MagicMock(return_value="/var/lib/vz/template/netork-images/debian-12.qcow2")
|
|
mixin._run_node_command = MagicMock(return_value="")
|
|
|
|
mock_vm = MagicMock()
|
|
mock_node.qemu.return_value = mock_vm
|
|
mock_node.qemu.post.return_value = None
|
|
mock_vm.config.post.return_value = None
|
|
mock_vm.config.get.return_value = {
|
|
"unused0": "local-lvm:vm-107-disk-0",
|
|
"scsi0": "local-lvm:vm-107-disk-0",
|
|
}
|
|
mock_vm.status.start.post.return_value = "UPID:pve1:133:start"
|
|
|
|
mock_task = MagicMock()
|
|
mock_task.status.get.return_value = {"status": "stopped", "exitstatus": "OK"}
|
|
mock_node.tasks.return_value = mock_task
|
|
|
|
with patch("time.sleep"):
|
|
mixin.create_vm_from_cloud_init(
|
|
name="pinned-mac-vm",
|
|
image_url="https://cloud.debian.org/images/cloud/bookworm/latest/debian-12-genericcloud-amd64.qcow2",
|
|
cpu=2,
|
|
memory=2048,
|
|
nics=[{"bridge": "vmbr0", "vlan_tag": 10, "mac": "02:aa:bb:cc:dd:ee"}],
|
|
cloud_init_config={"hostname": "pinned-mac-vm"},
|
|
)
|
|
|
|
net_call_args = next(c for c in mock_vm.config.post.call_args_list if "net0" in c[1])
|
|
assert net_call_args[1]["net0"] == "virtio=02:aa:bb:cc:dd:ee,bridge=vmbr0,tag=10"
|
|
|
|
|
|
def test_create_vm_from_cloud_init_without_mac_uses_bare_virtio():
|
|
"""Regression: omitting nics[i]['mac'] must still produce the original
|
|
bare 'virtio,bridge=...' string (auto-generated MAC), not 'virtio=None,...'."""
|
|
mixin = ProxmoxVMProvisionMixin()
|
|
mixin._node_name = "pve1"
|
|
|
|
mock_api = MagicMock()
|
|
mock_api.cluster.nextid.get.return_value = 108
|
|
mock_api.storage.return_value.get.return_value = {"path": "/var/lib/vz"}
|
|
|
|
mock_node = MagicMock()
|
|
mock_node.storage.get.return_value = [
|
|
{"storage": "local-lvm", "type": "lvmthin", "content": "images,rootdir", "enabled": 1},
|
|
{"storage": "snippets", "type": "dir", "content": "snippets", "enabled": 1},
|
|
]
|
|
mixin._api = mock_api
|
|
mixin._node_api = MagicMock(return_value=mock_node)
|
|
mixin._download_cloud_image = MagicMock(return_value="/var/lib/vz/template/netork-images/debian-12.qcow2")
|
|
mixin._run_node_command = MagicMock(return_value="")
|
|
|
|
mock_vm = MagicMock()
|
|
mock_node.qemu.return_value = mock_vm
|
|
mock_node.qemu.post.return_value = None
|
|
mock_vm.config.post.return_value = None
|
|
mock_vm.config.get.return_value = {
|
|
"unused0": "local-lvm:vm-108-disk-0",
|
|
"scsi0": "local-lvm:vm-108-disk-0",
|
|
}
|
|
mock_vm.status.start.post.return_value = "UPID:pve1:134:start"
|
|
|
|
mock_task = MagicMock()
|
|
mock_task.status.get.return_value = {"status": "stopped", "exitstatus": "OK"}
|
|
mock_node.tasks.return_value = mock_task
|
|
|
|
with patch("time.sleep"):
|
|
mixin.create_vm_from_cloud_init(
|
|
name="auto-mac-vm",
|
|
image_url="https://cloud.debian.org/images/cloud/bookworm/latest/debian-12-genericcloud-amd64.qcow2",
|
|
cpu=2,
|
|
memory=2048,
|
|
nics=[{"bridge": "vmbr0"}],
|
|
cloud_init_config={"hostname": "auto-mac-vm"},
|
|
)
|
|
|
|
net_call_args = next(c for c in mock_vm.config.post.call_args_list if "net0" in c[1])
|
|
assert net_call_args[1]["net0"] == "virtio,bridge=vmbr0"
|