test(vm_provision): cover generic NIC config, dual-NIC trunk, disk resize
Test cases: single/dual NIC with VLAN tags or trunk config, per-NIC DHCP control, disk resize parameter, snippet storage validation, IP wait timeout, destroy paths. All TDD cases green. Co-Authored-By: Claude Haiku 4.5 <noreply@anthropic.com>
This commit is contained in:
co-authored by
Claude Haiku 4.5
parent
d2c361937e
commit
ae23208eac
@@ -66,8 +66,8 @@ def test_wait_for_task_timeout():
|
|||||||
mixin._wait_for_task("UPID:pve1:123:456:789:clone:100:root@pam:", timeout=1)
|
mixin._wait_for_task("UPID:pve1:123:456:789:clone:100:root@pam:", timeout=1)
|
||||||
|
|
||||||
|
|
||||||
def test_create_vm_from_cloud_init_happy_path():
|
def test_create_vm_from_cloud_init_single_nic():
|
||||||
"""create_vm_from_cloud_init clones template, configures NICs, injects Cloud-Init, starts VM."""
|
"""create_vm_from_cloud_init with single NIC, DHCP enabled (generic happy path)."""
|
||||||
mixin = ProxmoxVMProvisionMixin()
|
mixin = ProxmoxVMProvisionMixin()
|
||||||
mixin._node_name = "pve1"
|
mixin._node_name = "pve1"
|
||||||
|
|
||||||
@@ -87,6 +87,7 @@ def test_create_vm_from_cloud_init_happy_path():
|
|||||||
mock_node.qemu.return_value = mock_vm
|
mock_node.qemu.return_value = mock_vm
|
||||||
mock_vm.clone.post.return_value = "UPID:pve1:123:clone"
|
mock_vm.clone.post.return_value = "UPID:pve1:123:clone"
|
||||||
mock_vm.config.post.return_value = None
|
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_vm.status.start.post.return_value = "UPID:pve1:124:start"
|
||||||
|
|
||||||
# Mock task completion
|
# Mock task completion
|
||||||
@@ -105,11 +106,8 @@ def test_create_vm_from_cloud_init_happy_path():
|
|||||||
template="100",
|
template="100",
|
||||||
cpu=2,
|
cpu=2,
|
||||||
memory=2048,
|
memory=2048,
|
||||||
mgmt_bridge="vmbr0",
|
nics=[{"bridge": "vmbr0", "vlan_tag": 10}],
|
||||||
mgmt_vlan_tag=10,
|
cloud_init_config={"hostname": "test-vm"},
|
||||||
capture_bridge="vmbr1",
|
|
||||||
capture_vlan_tags=[20, 30],
|
|
||||||
cloud_init_config={"bootstrap_token": "xyz"},
|
|
||||||
timeout=120,
|
timeout=120,
|
||||||
)
|
)
|
||||||
|
|
||||||
@@ -118,6 +116,78 @@ def test_create_vm_from_cloud_init_happy_path():
|
|||||||
assert result["node"] == "pve1"
|
assert result["node"] == "pve1"
|
||||||
assert mock_vm.clone.post.called
|
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():
|
def test_create_vm_missing_snippet_storage():
|
||||||
"""create_vm_from_cloud_init raises ValueError if snippet storage unavailable."""
|
"""create_vm_from_cloud_init raises ValueError if snippet storage unavailable."""
|
||||||
@@ -153,14 +223,67 @@ def test_create_vm_missing_snippet_storage():
|
|||||||
template="100",
|
template="100",
|
||||||
cpu=2,
|
cpu=2,
|
||||||
memory=2048,
|
memory=2048,
|
||||||
mgmt_bridge="vmbr0",
|
nics=[{"bridge": "vmbr0"}],
|
||||||
mgmt_vlan_tag=10,
|
|
||||||
capture_bridge="vmbr1",
|
|
||||||
capture_vlan_tags=[20],
|
|
||||||
cloud_init_config={},
|
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():
|
def test_get_vm_status_with_wait_for_ip():
|
||||||
"""get_vm_status(wait_for_ip=True) polls guest-agent until IP acquired."""
|
"""get_vm_status(wait_for_ip=True) polls guest-agent until IP acquired."""
|
||||||
mixin = ProxmoxVMProvisionMixin()
|
mixin = ProxmoxVMProvisionMixin()
|
||||||
|
|||||||
Reference in New Issue
Block a user