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

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>

napalm_proxmox/driver.py

@@ -49,6 +49,7 @@ from napalm_proxmox.sdn_mixin import ProxmoxSDNMixin
 from napalm_proxmox.lldp_mixin import ProxmoxLLDPMixin
 from napalm_proxmox.config_mixin import ProxmoxConfigMixin
 from napalm_proxmox.vm_mixin import ProxmoxVMMixin
+from napalm_proxmox.vm_provision_mixin import ProxmoxVMProvisionMixin
 from napalm_proxmox.routing_mixin import ProxmoxRoutingMixin
 from napalm_proxmox.system_mixin import ProxmoxSystemMixin
 
@@ -68,6 +69,7 @@ class ProxmoxDriver(
     ProxmoxLLDPMixin,
     ProxmoxConfigMixin,
     ProxmoxVMMixin,
+    ProxmoxVMProvisionMixin,
     ProxmoxRoutingMixin,
     ProxmoxSystemMixin,
     HypervisorDriver,

napalm_proxmox/vm_provision_mixin.py

@@ -0,0 +1,389 @@
+"""VM provisioning mixin for Proxmox — creates, destroys, and monitors VMs via Cloud-Init."""
+
+from __future__ import annotations
+
+import logging
+import time
+import yaml
+from typing import Any, Dict, List
+from urllib.parse import quote
+
+from napalm_device_types.models import VMProvisionResultDict, VMStatusDict
+
+_logger = logging.getLogger(__name__)
+
+
+class ProxmoxVMProvisionMixin:
+    """Mixin to add VM provisioning to ProxmoxDriver."""
+
+    def _wait_for_task(self, upid: str, timeout: int = 120) -> None:
+        """
+        Poll a Proxmox task until completion.
+
+        Polls /nodes/{node}/tasks/{upid}/status until status == 'stopped'.
+        Raises RuntimeError if exitstatus != 'OK' or timeout exceeded.
+        """
+        start_time = time.time()
+        while True:
+            elapsed = time.time() - start_time
+            if elapsed > timeout:
+                raise RuntimeError(f"Task {upid} timed out after {timeout}s")
+
+            try:
+                task_status = self._node_api().tasks(upid).status.get()
+            except Exception as e:
+                _logger.debug(f"Error polling task {upid}: {e}")
+                time.sleep(2)
+                continue
+
+            if task_status.get("status") == "stopped":
+                exitstatus = task_status.get("exitstatus", "UNKNOWN")
+                if exitstatus != "OK":
+                    raise RuntimeError(
+                        f"Task {upid} failed with exitstatus='{exitstatus}': "
+                        f"{task_status.get('exitstatus_text', 'no error message')}"
+                    )
+                return
+
+            time.sleep(2)
+
+    def create_vm_from_cloud_init(
+        self,
+        name: str,
+        *,
+        template: str,
+        cpu: int,
+        memory: int,
+        nics: List[Dict[str, Any]],
+        cloud_init_config: Dict[str, Any],
+        ssh_public_keys: List[str] | None = None,
+        disk_resize_gb: int | None = None,
+        timeout: int = 180,
+    ) -> VMProvisionResultDict:
+        """
+        Create a new VM from a Cloud-Init template via Proxmox API.
+
+        Steps:
+          1. Get next available VMID from cluster
+          2. Clone template VM (full clone, new VMID)
+          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. 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
+            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:
+            {"vmid": str, "name": str, "node": str}
+
+        Raises:
+            RuntimeError: provisioning failure (clone, config, timeout, etc.)
+            ValueError: invalid storage or configuration
+        """
+        try:
+            _logger.info(f"Creating VM '{name}' from template {template}")
+
+            # Step 1: Get next VMID
+            next_vmid = self._api.cluster.nextid.get()
+            vmid = int(next_vmid)
+            _logger.info(f"Allocated VMID {vmid}")
+
+            # Step 2: Clone template
+            _logger.info(f"Cloning template {template} → VMID {vmid}")
+            clone_upid = self._node_api().qemu(template).clone.post(
+                newid=vmid,
+                full=1,
+                name=name,
+            )
+            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)")
+
+            config_args = {"cores": cpu, "memory": memory}
+
+            # 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 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...")
+            storages = self._api.storage.get()
+            snippet_storage = None
+            for storage in storages:
+                content = storage.get("content", "")
+                if "snippets" in content and storage.get("enabled"):
+                    snippet_storage = storage["storage"]
+                    break
+
+            if not snippet_storage:
+                raise ValueError(
+                    "No storage with content='snippets' found. "
+                    "Configure a snippet-capable storage (e.g. local, nfs dir) "
+                    "and enable it."
+                )
+            _logger.info(f"Using snippet storage: {snippet_storage}")
+
+            # Step 5: Render and upload Cloud-Init config
+            _logger.info(f"Rendering Cloud-Init config for VMID {vmid}")
+
+            user_data_yaml = "#cloud-config\n" + yaml.dump(
+                cloud_init_config, default_flow_style=False
+            )
+
+            filename = f"{vmid}-user-data.yaml"
+            _logger.debug(f"Uploading Cloud-Init snippet {filename} to {snippet_storage}")
+
+            # Upload to snippet storage
+            self._node_api().storage(snippet_storage).upload.post(
+                content="snippets",
+                filename=filename,
+                data=user_data_yaml,
+            )
+
+            # Step 6: Configure Cloud-Init references and SSH keys
+            _logger.info(f"Setting Cloud-Init config for VM {vmid}")
+
+            cloud_init_args = {
+                "ide2": f"{snippet_storage}:cloudinit",
+                "citype": "nocloud",
+                "cicustom": f"user={snippet_storage}:snippets/{filename}",
+            }
+
+            # 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)
+                cloud_init_args["sshkeys"] = quote(sshkeys)
+
+            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
+            _logger.info(f"Starting VM {vmid}")
+            start_upid = self._node_api().qemu(vmid).status.start.post()
+            self._wait_for_task(start_upid, timeout=timeout)
+
+            _logger.info(f"VM {vmid} ('{name}') provisioned successfully on {self._node_name}")
+            return {
+                "vmid": str(vmid),
+                "name": name,
+                "node": self._node_name,
+            }
+
+        except Exception as e:
+            _logger.exception(f"Failed to create VM '{name}': {e}")
+            raise
+
+    def destroy_vm(
+        self,
+        vmid: str,
+        *,
+        remove_disk: bool = True,
+        timeout: int = 60,
+    ) -> None:
+        """
+        Destroy a virtual machine and optionally remove its storage.
+
+        Steps:
+          1. Stop the VM if running
+          2. Delete VM configuration and optionally disks
+          3. Clean up Cloud-Init snippets
+
+        Args:
+            vmid: hypervisor VMID (string, e.g. "101")
+            remove_disk: if True, also delete disks and storage
+            timeout: max seconds for stop/delete operations
+
+        Raises:
+            RuntimeError: VM doesn't exist or destruction fails
+        """
+        try:
+            vmid_int = int(vmid)
+            _logger.info(f"Destroying VM {vmid}")
+
+            # Step 1: Stop the VM if running
+            try:
+                _logger.debug(f"Stopping VM {vmid}")
+                stop_upid = self._node_api().qemu(vmid_int).status.stop.post()
+                self._wait_for_task(stop_upid, timeout=timeout)
+            except Exception as e:
+                _logger.debug(f"VM {vmid} stop failed (may already be stopped): {e}")
+
+            # Step 2: Delete VM
+            _logger.debug(f"Deleting VM {vmid} configuration and disks")
+            self._node_api().qemu(vmid_int).delete(
+                purge=1,
+                destroy_unreferenced_disks=1 if remove_disk else 0,
+            )
+
+            # Step 3: Clean up Cloud-Init snippets
+            # (This is best-effort; snippet files may be unreachable if storage is unavailable)
+            try:
+                config = self._node_api().qemu(vmid_int).config.get()
+                cicustom = config.get("cicustom", "")
+                if "snippets/" in cicustom:
+                    parts = cicustom.split("=")
+                    if len(parts) >= 2:
+                        snippet_ref = parts[1]  # e.g. "snippets:snippets/101-user-data.yaml"
+                        storage, filepath = snippet_ref.split(":", 1)
+                        _logger.debug(f"Deleting snippet {filepath} from {storage}")
+                        try:
+                            self._node_api().storage(storage).content(filepath).delete()
+                        except Exception as e:
+                            _logger.warning(f"Failed to delete snippet {filepath}: {e}")
+            except Exception as e:
+                _logger.debug(f"Could not clean up snippets for VM {vmid}: {e}")
+
+            _logger.info(f"VM {vmid} destroyed successfully")
+
+        except Exception as e:
+            _logger.exception(f"Failed to destroy VM {vmid}: {e}")
+            raise
+
+    def get_vm_status(
+        self,
+        vmid: str,
+        *,
+        wait_for_ip: bool = False,
+        timeout: int = 300,
+        poll_interval: int = 5,
+    ) -> VMStatusDict:
+        """
+        Get the runtime status of a virtual machine.
+
+        Optionally waits for the guest-agent to report an IP address on the
+        management NIC (net0), useful after provisioning.
+
+        Args:
+            vmid: hypervisor VMID (string)
+            wait_for_ip: if True, poll until IP appears on net0
+            timeout: max seconds to wait for IP (if wait_for_ip=True)
+            poll_interval: seconds between status polls
+
+        Returns:
+            {"status": str, "ip_address": str, "hostname": str, "mac_address": str}
+            (ip_address, hostname, mac_address only if VM is running and has network info)
+
+        Raises:
+            RuntimeError: VM doesn't exist or wait_for_ip times out
+        """
+        try:
+            vmid_int = int(vmid)
+            _logger.debug(f"Getting status for VM {vmid}")
+
+            # Get VM config to infer net0 MAC (for matching guest-agent results)
+            try:
+                config = self._node_api().qemu(vmid_int).config.get()
+            except Exception:
+                # VM may not exist yet or config not readable
+                return {"status": "unknown"}
+
+            # Parse net0 MAC from config (if present)
+            net0_line = config.get("net0", "")
+            expected_mac = None
+            # Example: "virtio,bridge=vmbr0,tag=10" — no explicit MAC
+            # Proxmox auto-generates MACs in a deterministic pattern, but we'll
+            # match by looking for the first NIC's IP in guest-agent results
+
+            # Polling loop
+            start_time = time.time()
+            while True:
+                elapsed = time.time() - start_time
+                if wait_for_ip and elapsed > timeout:
+                    raise RuntimeError(
+                        f"VM {vmid} failed to acquire IP within {timeout}s"
+                    )
+
+                try:
+                    # Query guest-agent network interfaces
+                    agent_info = self._node_api().qemu(vmid_int).agent.network_get_interfaces.get()
+                    interfaces = agent_info.get("result", [])
+
+                    # Find net0 (first interface with IP)
+                    if interfaces:
+                        net0_iface = interfaces[0]  # Assumes net0 is first in list
+                        net0_mac = net0_iface.get("hardware-address", "")
+                        ip_addresses = net0_iface.get("ip-addresses", [])
+
+                        if ip_addresses:
+                            # Found IP
+                            ip_info = ip_addresses[0]
+                            ip_addr = ip_info.get("ip-address", "")
+                            if ip_addr:
+                                _logger.info(f"VM {vmid} acquired IP {ip_addr}")
+                                return {
+                                    "status": "running",
+                                    "ip_address": ip_addr,
+                                    "hostname": net0_iface.get("name", ""),
+                                    "mac_address": net0_mac,
+                                }
+
+                except Exception as e:
+                    _logger.debug(f"Error querying guest-agent for VM {vmid}: {e}")
+
+                if not wait_for_ip:
+                    # Return immediate status without IP
+                    return {"status": "running"}
+
+                # Wait before next poll
+                time.sleep(poll_interval)
+
+        except RuntimeError:
+            raise
+        except Exception as e:
+            _logger.exception(f"Error getting status for VM {vmid}: {e}")
+            raise RuntimeError(f"Failed to get VM {vmid} status: {e}")

tests/test_vm_provision_mixin.py

@@ -0,0 +1,396 @@
+"""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