Christian Manivong
ddd4e6fc03
vnet's SDN tag (VLAN ID) is now surfaced in get_network_targets() output instead of being silently discarded. Bridges never set this field.
514 lines
18 KiB
Python
514 lines
18 KiB
Python
"""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": "snippets", "type": "dir", "content": "snippets", "enabled": 1}
|
|
]
|
|
|
|
mock_node = MagicMock()
|
|
mixin._api = mock_api
|
|
mixin._node_api = MagicMock(return_value=mock_node)
|
|
|
|
# Mock VM operations
|
|
mock_vm = MagicMock()
|
|
mock_node.qemu.return_value = mock_vm
|
|
mock_vm.clone.post.return_value = "UPID:pve1:123:clone"
|
|
mock_vm.config.post.return_value = None
|
|
mock_vm.config.get.return_value = {"scsi0": "local:100/vm-101-disk-0.raw"}
|
|
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",
|
|
template="100",
|
|
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_vm.clone.post.called
|
|
|
|
# Verify NIC config was set correctly: net0 with tag=10, DHCP enabled
|
|
config_call_args = mock_vm.config.post.call_args_list[0] # First call (cores/memory/net0)
|
|
assert "net0" in config_call_args[1]
|
|
assert "tag=10" in config_call_args[1]["net0"]
|
|
assert "vmbr0" in config_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": "snippets", "type": "dir", "content": "snippets", "enabled": 1}
|
|
]
|
|
|
|
mock_node = MagicMock()
|
|
mixin._api = mock_api
|
|
mixin._node_api = MagicMock(return_value=mock_node)
|
|
|
|
# Mock VM operations
|
|
mock_vm = MagicMock()
|
|
mock_node.qemu.return_value = mock_vm
|
|
mock_vm.clone.post.return_value = "UPID:pve1:125:clone"
|
|
mock_vm.config.post.return_value = None
|
|
mock_vm.config.get.return_value = {"scsi0": "local:100/vm-102-disk-0.raw"}
|
|
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",
|
|
template="100",
|
|
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
|
|
config_call_args = mock_vm.config.post.call_args_list[0]
|
|
assert "net0" in config_call_args[1]
|
|
assert "net1" in config_call_args[1]
|
|
assert "tag=10" in config_call_args[1]["net0"]
|
|
assert "trunks=20;30" in config_call_args[1]["net1"]
|
|
assert "vmbr1" in config_call_args[1]["net1"]
|
|
|
|
# Verify DHCP config: ipconfig0 yes, ipconfig1 no
|
|
cloud_init_call_args = mock_vm.config.post.call_args_list[1] # Second call (ipconfig)
|
|
assert "ipconfig0" in cloud_init_call_args[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 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 for clone operation (so we get to the storage check)
|
|
mock_node = MagicMock()
|
|
mixin._node_api = MagicMock(return_value=mock_node)
|
|
mock_vm = MagicMock()
|
|
mock_node.qemu.return_value = mock_vm
|
|
mock_vm.clone.post.return_value = "UPID:pve1:123:clone"
|
|
mock_vm.config.post.return_value = None
|
|
|
|
# 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",
|
|
template="100",
|
|
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": "snippets", "type": "dir", "content": "snippets", "enabled": 1}
|
|
]
|
|
|
|
mock_node = MagicMock()
|
|
mixin._api = mock_api
|
|
mixin._node_api = MagicMock(return_value=mock_node)
|
|
|
|
# Mock VM operations
|
|
mock_vm = MagicMock()
|
|
mock_node.qemu.return_value = mock_vm
|
|
mock_vm.clone.post.return_value = "UPID:pve1:127:clone"
|
|
mock_vm.config.post.return_value = None
|
|
mock_vm.config.get.return_value = {"scsi0": "local:100/vm-103-disk-0.raw"}
|
|
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",
|
|
template="100",
|
|
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
|