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