diff --git a/napalm_proxmox/vm_provision_mixin.py b/napalm_proxmox/vm_provision_mixin.py index e0b2950..de8351f 100644 --- a/napalm_proxmox/vm_provision_mixin.py +++ b/napalm_proxmox/vm_provision_mixin.py @@ -54,11 +54,13 @@ class ProxmoxVMProvisionMixin: template: str, cpu: int, memory: int, - nics: List[Dict[str, Any]], + mgmt_bridge: str, + mgmt_vlan_tag: int | None, + capture_bridge: str, + capture_vlan_tags: List[int], cloud_init_config: Dict[str, Any], ssh_public_keys: List[str] | None = None, - disk_resize_gb: int | None = None, - timeout: int = 180, + timeout: int = 120, ) -> VMProvisionResultDict: """ Create a new VM from a Cloud-Init template via Proxmox API. @@ -66,23 +68,24 @@ class ProxmoxVMProvisionMixin: Steps: 1. Get next available VMID from cluster 2. Clone template VM (full clone, new VMID) - 3. Configure CPU, memory, and network interfaces + 3. Configure CPU, memory, and dual NICs (mgmt + capture) 4. Verify snippet storage exists 5. Render cloud-init config to YAML and upload 6. Set Cloud-Init config references and SSH keys - 7. Optionally resize root disk - 8. Start the VM - 9. Return VMID, name, node + 7. Start the VM + 8. 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 - nics: list of NIC config dicts (bridge, vlan_tag/trunk_vlan_tags, dhcp flag) + 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) 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: @@ -109,30 +112,24 @@ class ProxmoxVMProvisionMixin: ) self._wait_for_task(clone_upid, timeout=timeout) - # Step 3: Configure CPU, memory, and NICs - _logger.info(f"Configuring VM {vmid}: {cpu} CPU, {memory}MB RAM, {len(nics)} NIC(s)") + # Step 3: Configure CPU, memory, and dual NICs + _logger.info(f"Configuring VM {vmid}: {cpu} CPU, {memory}MB RAM") - config_args = {"cores": cpu, "memory": memory} + # 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}" - # 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") + # 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 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) + self._node_api().qemu(vmid).config.post( + cores=cpu, + memory=memory, + net0=net0_config, + net1=net1_config, + ) # Step 4: Verify snippet storage exists _logger.info("Checking for snippet storage...") @@ -155,6 +152,12 @@ 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 ) @@ -176,14 +179,10 @@ 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) @@ -191,28 +190,7 @@ class ProxmoxVMProvisionMixin: self._node_api().qemu(vmid).config.post(**cloud_init_args) - # 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 + # Step 7: 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) diff --git a/tests/test_vm_provision_mixin.py b/tests/test_vm_provision_mixin.py index 7bdf4f6..65631c8 100644 --- a/tests/test_vm_provision_mixin.py +++ b/tests/test_vm_provision_mixin.py @@ -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_single_nic(): - """create_vm_from_cloud_init with single NIC, DHCP enabled (generic happy path).""" +def test_create_vm_from_cloud_init_happy_path(): + """create_vm_from_cloud_init clones template, configures NICs, injects Cloud-Init, starts VM.""" mixin = ProxmoxVMProvisionMixin() mixin._node_name = "pve1" @@ -87,7 +87,6 @@ def test_create_vm_from_cloud_init_single_nic(): 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 @@ -106,8 +105,11 @@ def test_create_vm_from_cloud_init_single_nic(): template="100", cpu=2, memory=2048, - nics=[{"bridge": "vmbr0", "vlan_tag": 10}], - cloud_init_config={"hostname": "test-vm"}, + mgmt_bridge="vmbr0", + mgmt_vlan_tag=10, + capture_bridge="vmbr1", + capture_vlan_tags=[20, 30], + cloud_init_config={"bootstrap_token": "xyz"}, timeout=120, ) @@ -116,78 +118,6 @@ def test_create_vm_from_cloud_init_single_nic(): 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.""" @@ -223,67 +153,14 @@ def test_create_vm_missing_snippet_storage(): template="100", cpu=2, memory=2048, - nics=[{"bridge": "vmbr0"}], + mgmt_bridge="vmbr0", + mgmt_vlan_tag=10, + capture_bridge="vmbr1", + capture_vlan_tags=[20], 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()