"""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() # Regression: without agent="1" Proxmox never attaches the virtio-serial # channel the QEMU guest agent needs, so get_vm_status(wait_for_ip=True) # can never succeed no matter what's installed inside the guest. assert mock_node.qemu.post.call_args[1]["agent"] == "1" # 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"} # The real Proxmox REST path is "network-get-interfaces" (hyphens), # reached via agent(...) as a callable resource, not dotted attribute # access — mock that exact call shape. mock_agent_call = MagicMock() mock_vm.agent.return_value = mock_agent_call mock_agent_call.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"}], } ] }, ] with patch("time.sleep"): result = mixin.get_vm_status("101", wait_for_ip=True, timeout=30, poll_interval=1) mock_vm.agent.assert_called_with("network-get-interfaces") 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_skips_loopback_interface(): """The guest agent commonly reports "lo" before the real NIC — it must be skipped rather than mistaken for the VM's address.""" mixin = ProxmoxVMProvisionMixin() mock_node = MagicMock() mixin._node_api = MagicMock(return_value=mock_node) mock_vm = MagicMock() mock_node.qemu.return_value = mock_vm mock_vm.config.get.return_value = {"net0": "virtio,bridge=vmbr0"} mock_agent_call = MagicMock() mock_vm.agent.return_value = mock_agent_call mock_agent_call.get.return_value = { "result": [ { "name": "lo", "hardware-address": "00:00:00:00:00:00", "ip-addresses": [{"ip-address": "127.0.0.1", "ip-address-type": "ipv4"}], }, { "name": "eth0", "hardware-address": "aa:bb:cc:dd:ee:00", "ip-addresses": [{"ip-address": "10.0.0.101", "ip-address-type": "ipv4"}], }, ] } with patch("time.sleep"): result = mixin.get_vm_status("101", wait_for_ip=True, timeout=30, poll_interval=1) assert result["ip_address"] == "10.0.0.101" 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_call = MagicMock() mock_vm.agent.return_value = mock_agent_call mock_agent_call.get.return_value = { "result": [{"name": "eth0", "hardware-address": "aa:bb:cc:dd:ee:00"}] } 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=,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" def test_create_vm_cloudinit_drive_uses_image_storage_not_snippet_storage(): """Regression: real Proxmox fails at VM *start* time with "storage 'X' does not support content-type 'images'" if ide2 (the cloud-init drive) points at the snippets storage — a cloud-init drive is a disk image and needs content='images', same as the root disk. Only cicustom (the snippet file reference) should use the snippets storage. Found live: a real deployment had 'local' (snippets-only) and 'local-zfs' (images) as two distinct storages, and ide2 was wrongly set to the former.""" mixin = ProxmoxVMProvisionMixin() mixin._node_name = "pve1" mock_api = MagicMock() mock_api.cluster.nextid.get.return_value = 109 mock_api.storage.return_value.get.return_value = {"path": "/var/lib/vz"} mock_node = MagicMock() mock_node.storage.get.return_value = [ {"storage": "local-zfs", "type": "zfspool", "content": "images,rootdir", "enabled": 1}, {"storage": "local", "type": "dir", "content": "iso,snippets,backup,vztmpl", "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-109-disk-0", "scsi0": "local-zfs:vm-109-disk-0", } mock_vm.status.start.post.return_value = "UPID:pve1:135: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="cloudinit-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": "cloudinit-storage-vm"}, ) cloud_init_call = next(c for c in mock_vm.config.post.call_args_list if "ide2" in c[1]) assert cloud_init_call[1]["ide2"] == "local-zfs:cloudinit" assert cloud_init_call[1]["cicustom"].startswith("user=local:snippets/")