NAPALM/napalm-proxmox
1
0
Fork 0
Code Issues 1 Pull Requests Actions Packages Projects Releases Wiki Activity

Reapply "Merge feature/generic-vm-provisioning: generalize vm_provision_mixin for arbitrary NIC configs"

Browse Source 
This reverts commit 6ede48d244.
This commit is contained in:
Christian Manivong
2026-07-07 08:21:00 +02:00
parent 6ede48d244
commit a5a5b634b0
2 changed files with 192 additions and 47 deletions
Download Patch File Download Diff File Expand all files Collapse all files
napalm_proxmox/vm_provision_mixin.py
+58 -36
View File
@@ -54,13 +54,11 @@ class ProxmoxVMProvisionMixin:
template: str,
cpu: int,
memory: int,
mgmt_bridge: str,
mgmt_vlan_tag: int | None,
capture_bridge: str,
capture_vlan_tags: List[int],
nics: List[Dict[str, Any]],
cloud_init_config: Dict[str, Any],
ssh_public_keys: List[str] | None = None,
timeout: int = 120,
disk_resize_gb: int | None = None,
timeout: int = 180,
) -> VMProvisionResultDict:
"""
Create a new VM from a Cloud-Init template via Proxmox API.
@@ -68,24 +66,23 @@ class ProxmoxVMProvisionMixin:
Steps:
1. Get next available VMID from cluster
2. Clone template VM (full clone, new VMID)
3. Configure CPU, memory, and dual NICs (mgmt + capture)
3. Configure CPU, memory, and network interfaces
4. Verify snippet storage exists
5. Render cloud-init config to YAML and upload
6. Set Cloud-Init config references and SSH keys
7. Start the VM
8. Return VMID, name, node
7. Optionally resize root disk
8. Start the VM
9. Return VMID, name, node
Args:
name: new VM display name
template: template VMID/name to clone from
cpu: number of vCPUs
memory: RAM in MB
mgmt_bridge: management bridge name
mgmt_vlan_tag: VLAN tag for mgmt NIC (None = untagged)
capture_bridge: packet capture bridge name (must be VLAN-aware)
capture_vlan_tags: list of VLAN IDs for capture NIC (trunk)
nics: list of NIC config dicts (bridge, vlan_tag/trunk_vlan_tags, dhcp flag)
cloud_init_config: user-data dict (will be YAML-rendered)
ssh_public_keys: SSH public keys to inject
disk_resize_gb: resize root disk to this size (None = no resize)
timeout: max seconds for provisioning
Returns:
@@ -112,24 +109,30 @@ class ProxmoxVMProvisionMixin:
)
self._wait_for_task(clone_upid, timeout=timeout)
# Step 3: Configure CPU, memory, and dual NICs
_logger.info(f"Configuring VM {vmid}: {cpu} CPU, {memory}MB RAM")
# Step 3: Configure CPU, memory, and NICs
_logger.info(f"Configuring VM {vmid}: {cpu} CPU, {memory}MB RAM, {len(nics)} NIC(s)")
# Build net0 (mgmt) config
net0_config = f"virtio,bridge={mgmt_bridge}"
if mgmt_vlan_tag is not None:
net0_config += f",tag={mgmt_vlan_tag}"
config_args = {"cores": cpu, "memory": memory}
# Build net1 (capture trunk) config
vlan_list = ";".join(str(v) for v in capture_vlan_tags)
net1_config = f"virtio,bridge={capture_bridge},trunks={vlan_list}"
# Build NIC config strings generically
for i, nic in enumerate(nics):
bridge = nic.get("bridge")
if not bridge:
raise ValueError(f"NIC {i}: bridge is required")
self._node_api().qemu(vmid).config.post(
cores=cpu,
memory=memory,
net0=net0_config,
net1=net1_config,
)
# Build base config: model + bridge
net_config = f"virtio,bridge={bridge}"
# Add VLAN configuration (access vs trunk)
if "trunk_vlan_tags" in nic and nic["trunk_vlan_tags"]:
vlan_list = ";".join(str(v) for v in nic["trunk_vlan_tags"])
net_config += f",trunks={vlan_list}"
elif "vlan_tag" in nic and nic["vlan_tag"] is not None:
net_config += f",tag={nic['vlan_tag']}"
config_args[f"net{i}"] = net_config
self._node_api().qemu(vmid).config.post(**config_args)
# Step 4: Verify snippet storage exists
_logger.info("Checking for snippet storage...")
@@ -152,12 +155,6 @@ class ProxmoxVMProvisionMixin:
# Step 5: Render and upload Cloud-Init config
_logger.info(f"Rendering Cloud-Init config for VMID {vmid}")
# Ensure cloud_init_config includes runcmd to bring up capture NIC
if "runcmd" not in cloud_init_config:
cloud_init_config["runcmd"] = []
if not any("eth1" in cmd if isinstance(cmd, str) else False for cmd in cloud_init_config.get("runcmd", [])):
cloud_init_config["runcmd"].insert(0, "ip link set eth1 up")
user_data_yaml = "#cloud-config\n" + yaml.dump(
cloud_init_config, default_flow_style=False
)
@@ -179,10 +176,14 @@ class ProxmoxVMProvisionMixin:
"ide2": f"{snippet_storage}:cloudinit",
"citype": "nocloud",
"cicustom": f"user={snippet_storage}:snippets/{filename}",
"ipconfig0": "ip=dhcp", # net0 gets DHCP (mgmt)
# NO ipconfig1 for net1 (capture NIC stays unnummeriert)
}
# Configure DHCP for NICs where enabled (default True for index 0, False otherwise)
for i, nic in enumerate(nics):
dhcp_enabled = nic.get("dhcp", i == 0) # Default DHCP for first NIC only
if dhcp_enabled:
cloud_init_args[f"ipconfig{i}"] = "ip=dhcp"
if ssh_public_keys:
# URL-encode SSH keys for Proxmox API
sshkeys = ";".join(ssh_public_keys)
@@ -190,7 +191,28 @@ class ProxmoxVMProvisionMixin:
self._node_api().qemu(vmid).config.post(**cloud_init_args)
# Step 7: Start the VM
# Step 7: Optionally resize root disk
if disk_resize_gb is not None:
_logger.info(f"Resizing root disk to {disk_resize_gb}GB")
# Find root disk (scsi0, virtio0, ide0, sata0 — whichever is first)
try:
config = self._node_api().qemu(vmid).config.get()
root_disk = None
for prefix in ("scsi", "virtio", "ide", "sata"):
if f"{prefix}0" in config:
root_disk = f"{prefix}0"
break
if root_disk:
self._node_api().qemu(vmid).resize.put(
disk=root_disk,
size=f"{disk_resize_gb}G",
)
else:
_logger.warning(f"Could not find root disk for VM {vmid}, skipping resize")
except Exception as e:
_logger.warning(f"Failed to resize disk: {e}, continuing anyway")
# Step 8: Start the VM
_logger.info(f"Starting VM {vmid}")
start_upid = self._node_api().qemu(vmid).status.start.post()
self._wait_for_task(start_upid, timeout=timeout)
tests/test_vm_provision_mixin.py
+134 -11
View File
@@ -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)
def test_create_vm_from_cloud_init_happy_path():
"""create_vm_from_cloud_init clones template, configures NICs, injects Cloud-Init, starts VM."""
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"
@@ -87,6 +87,7 @@ def test_create_vm_from_cloud_init_happy_path():
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
@@ -105,11 +106,8 @@ def test_create_vm_from_cloud_init_happy_path():
template="100",
cpu=2,
memory=2048,
mgmt_bridge="vmbr0",
mgmt_vlan_tag=10,
capture_bridge="vmbr1",
capture_vlan_tags=[20, 30],
cloud_init_config={"bootstrap_token": "xyz"},
nics=[{"bridge": "vmbr0", "vlan_tag": 10}],
cloud_init_config={"hostname": "test-vm"},
timeout=120,
)
@@ -118,6 +116,78 @@ def test_create_vm_from_cloud_init_happy_path():
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."""
@@ -153,14 +223,67 @@ def test_create_vm_missing_snippet_storage():
template="100",
cpu=2,
memory=2048,
mgmt_bridge="vmbr0",
mgmt_vlan_tag=10,
capture_bridge="vmbr1",
capture_vlan_tags=[20],
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()