CHANGELOG.md

@@ -8,6 +8,23 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 ## [Unreleased]
 
 ### Added
+- FreeBSD and OpenBSD guests: `create_vm_from_cloud_init(guest_os=...)` with
+  `GUEST_AGENTS` from napalm-device-types 3.0. A BSD guest gets OS type
+  `other` and a network-config v2 snippet of its own (`cicustom network=`),
+  because FreeBSD's nuageinit skips runcmd on the v1 Proxmox generates.
+  OpenBSD's agent runs over ISA serial behind `serial0`, as the image keeps
+  its console on com0. An unknown `guest_os` is refused before anything is
+  created.
+- Packed cloud images (FreeBSD's `.qcow2.xz`) are unpacked on the node after
+  the packed file's checksum is verified; only the unpacked file is cached.
+  Proxmox's `download-url` unpacks ISOs only, so they always go over SSH.
+
+### Fixed
+- `get_vm_status` no longer reports a loopback address as the VM's IP when
+  the agent lists loopback under another name than `lo` (`lo0` on the BSDs,
+  listed first by OpenBSD).
+- `destroy_vm` reads the VM's cloud-init snippets before deleting the VM, and
+  removes all of them; it read them afterwards, when the config was gone (#15).
 - Cloud images for `create_vm_from_cloud_init` are downloaded by Proxmox itself
   when the node has an active storage with content type `import` (Proxmox
   8.2+): `download-url` with checksum verification, then `import-from` as the

napalm_proxmox/vm_provision_mixin.py

@@ -4,19 +4,26 @@ from __future__ import annotations
 
 import base64
 import hashlib
+import ipaddress
 import logging
 import re
 import time
-import yaml
 from typing import TYPE_CHECKING, Any, Dict, List
 from urllib.parse import quote
 
+import yaml
 from napalm_device_types.models import (
     NetworkTargetDict,
     StorageTargetDict,
     VMProvisionResultDict,
     VMStatusDict,
 )
+from napalm_device_types.provisioning import (
+    QEMU_GUEST_AGENTS,
+    GuestAgents,
+    network_config,
+    split_compression,
+)
 
 if TYPE_CHECKING:
     # Type-only: VMCpuTypeDict is newer than the napalm_device_types floor in
@@ -76,6 +83,18 @@ _IMAGE_CACHE_DIR = "/var/lib/vz/template/netork-images"
 # raw for a qcow2 would attach the qcow2 container as a raw disk silently.
 _IMPORT_EXTENSIONS = {"qcow2": "qcow2", "raw": "raw", "vmdk": "vmdk", "img": "qcow2"}
 
+# The VM shell each guest needs. OpenBSD's qemu-ga cannot use virtio-serial,
+# and over ISA serial it answers only as the second port (cua01): the OpenBSD
+# cloud image keeps its console on com0, which serial0 takes (netork#793).
+_GUEST_VM_SETTINGS: dict[str, dict[str, str]] = {
+    "linux": {"ostype": "l26", "agent": "1"},
+    "freebsd": {"ostype": "other", "agent": "1"},
+    "openbsd": {"ostype": "other", "agent": "1,type=isa", "serial0": "socket"},
+}
+
+# A NIC as Proxmox reports it: "virtio=BC:24:11:AA:BB:02,bridge=vmbr0".
+_NIC_MAC = re.compile(r"^[a-z0-9]+=([0-9A-Fa-f:]{17})")
+
 # Characters Proxmox keeps in a content file name (PVE::Storage's
 # SAFE_CHAR_CLASS_RE); anything else it would rewrite behind our back.
 _UNSAFE_FILENAME_CHARS = re.compile(r"[^A-Za-z0-9\-.+=_]")
@@ -111,6 +130,10 @@ def _import_volume_name(image_url: str) -> str | None:
 class ProxmoxVMProvisionMixin:
     """Mixin to add VM provisioning to ProxmoxDriver."""
 
+    #: Every guest QEMU's agent runs on; see _GUEST_VM_SETTINGS for the VM
+    #: hardware each one needs.
+    GUEST_AGENTS: GuestAgents = QEMU_GUEST_AGENTS
+
     def _run_node_command(self, command: str, timeout: int) -> str:
         """
         Execute a shell command on the Proxmox node via SSH, raising on failure.
@@ -214,6 +237,61 @@ class ProxmoxVMProvisionMixin:
 
         raise AssertionError("unreachable")  # loop always returns or raises above
 
+    def _cloud_image_on_node(self, image_url: str, image_checksum: str | None, timeout: int) -> str:
+        """The node path of the image, downloaded, verified and unpacked.
+
+        A packed image (FreeBSD ships .qcow2.xz) is unpacked on the node after
+        its checksum, which covers the packed file, has been verified; only
+        the unpacked file is kept. Proxmox's download-url unpacks ISOs only,
+        so a packed image always comes this way.
+        """
+        filename = image_url.rstrip("/").rsplit("/", 1)[-1]
+        unpacked_name, unpack = split_compression(filename)
+        if unpack is None:
+            return self._download_cloud_image(image_url, image_checksum, timeout=timeout)
+
+        unpacked = f"{_IMAGE_CACHE_DIR}/{_url_key(image_url)}-{unpacked_name}"
+        cached = self._run_node_command(
+            f"mkdir -p {_IMAGE_CACHE_DIR} && test -f {unpacked} && echo EXISTS || echo MISSING",
+            timeout=30,
+        )
+        if "EXISTS" in cached:
+            return unpacked
+        packed = self._download_cloud_image(image_url, image_checksum, timeout=timeout)
+        _logger.info(f"Unpacking {packed} -> {unpacked}")
+        self._run_node_command(
+            f"{unpack} {packed} > {unpacked}.tmp && mv {unpacked}.tmp {unpacked} && rm -f {packed}",
+            timeout=timeout,
+        )
+        return unpacked
+
+    def _write_snippet(self, storage_path: str, filename: str, content: str) -> None:
+        """Write a cloud-init snippet into *storage_path*/snippets on the node.
+
+        Proxmox's /storage/{s}/upload API only accepts content in
+        {iso, vztmpl, import} — "snippets" is rejected outright ("does not
+        have a value in the enumeration"). Snippets can only be written
+        directly to the filesystem, so this goes over SSH.
+        """
+        _logger.debug(f"Writing Cloud-Init snippet {filename} to {storage_path}/snippets")
+        encoded = base64.b64encode(content.encode("utf-8")).decode("ascii")
+        self._run_node_command(
+            f"mkdir -p {storage_path}/snippets && "
+            f"echo {encoded} | base64 -d > {storage_path}/snippets/{filename}",
+            timeout=30,
+        )
+
+    def _nic_macs(self, vmid: int, nics: List[Dict[str, Any]]) -> list[tuple[str, bool]]:
+        """``(mac, dhcp)`` per NIC, with the MAC Proxmox assigned where none was given."""
+        config = self._node_api().qemu(vmid).config.get()
+        macs = []
+        for i, nic in enumerate(nics):
+            match = _NIC_MAC.match(config.get(f"net{i}", ""))
+            if not match:
+                raise RuntimeError(f"VM {vmid} has no MAC address on net{i}")
+            macs.append((match.group(1), nic.get("dhcp", i == 0)))
+        return macs
+
     def _find_import_storage(self) -> str | None:
         """The first storage on this node that accepts content "import".
 
@@ -276,7 +354,7 @@ class ProxmoxVMProvisionMixin:
         The path for nodes without an import storage, or for an image type
         Proxmox cannot import itself.
         """
-        local_path = self._download_cloud_image(image_url, image_checksum, timeout=download_timeout)
+        local_path = self._cloud_image_on_node(image_url, image_checksum, download_timeout)
         _logger.info(f"Importing {local_path} into VM {vmid} on storage {image_storage}")
         self._run_node_command(
             f"qm importdisk {vmid} {local_path} {image_storage} --format qcow2",
@@ -491,6 +569,7 @@ class ProxmoxVMProvisionMixin:
         disk_resize_gb: int | None = None,
         storage: str | None = None,
         cpu_type: str | None = None,
+        guest_os: str = "linux",
         download_timeout: int = 300,
         timeout: int = 180,
     ) -> VMProvisionResultDict:
@@ -524,6 +603,10 @@ class ProxmoxVMProvisionMixin:
             storage: storage pool for the root disk (None = auto-detect first
                 enabled, node-available storage with content='images')
             cpu_type: CPU model from get_vm_cpu_types() (None = x86-64-v2-AES)
+            guest_os: the OS in the image, a key of GUEST_AGENTS. It sets the
+                OS type and how the agent is attached; a guest other than Linux
+                also gets a network-config v2 snippet of its own, because
+                FreeBSD's nuageinit skips runcmd on the v1 Proxmox writes.
             download_timeout: max seconds for the image download (skipped if cached)
             timeout: max seconds for the remaining provisioning steps
 
@@ -536,8 +619,13 @@ class ProxmoxVMProvisionMixin:
                 unknown or that this node's CPU cannot run (raised before
                 anything is created)
         """
+        if guest_os not in self.GUEST_AGENTS:
+            raise ValueError(
+                f"Cannot provision a {guest_os!r} guest; this driver provisions "
+                f"{', '.join(sorted(self.GUEST_AGENTS))}"
+            )
         try:
-            _logger.info(f"Creating VM '{name}' from image {image_url}")
+            _logger.info(f"Creating {guest_os} VM '{name}' from image {image_url}")
             cpu_model = self._resolve_cpu_type(cpu_type)
 
             # Step 1: Get next VMID
@@ -553,12 +641,11 @@ class ProxmoxVMProvisionMixin:
                 memory=memory,
                 cores=cpu,
                 cpu=cpu_model,
-                ostype="l26",
                 scsihw="virtio-scsi-pci",
-                # Without this, Proxmox never attaches the virtio-serial
-                # channel the QEMU guest agent needs — get_vm_status's
-                # agent queries (below) would have nothing to talk to.
-                agent="1",
+                # Includes agent: without it Proxmox never attaches the
+                # channel the QEMU guest agent needs — get_vm_status's agent
+                # queries (below) would have nothing to talk to.
+                **_GUEST_VM_SETTINGS[guest_os],
             )
 
             # Step 3: Download cloud image (cached) and import as root disk
@@ -621,20 +708,20 @@ class ProxmoxVMProvisionMixin:
             )
 
             filename = f"{vmid}-user-data.yaml"
-            _logger.debug(f"Writing Cloud-Init snippet {filename} to {snippet_storage}")
-
-            # Proxmox's /storage/{s}/upload API only accepts content in
-            # {iso, vztmpl, import} — "snippets" is rejected outright
-            # ("does not have a value in the enumeration"). Snippets can only
-            # be written directly to the filesystem, so resolve the storage's
-            # backing path and write the file over SSH instead.
             storage_path = self._get_storage_path(snippet_storage)
-            encoded = base64.b64encode(user_data_yaml.encode("utf-8")).decode("ascii")
-            self._run_node_command(
-                f"mkdir -p {storage_path}/snippets && "
-                f"echo {encoded} | base64 -d > {storage_path}/snippets/{filename}",
-                timeout=30,
-            )
+            self._write_snippet(storage_path, filename, user_data_yaml)
+            cicustom = f"user={snippet_storage}:snippets/{filename}"
+
+            # Proxmox writes network-config v1, and FreeBSD's nuageinit fails
+            # on it and then skips runcmd, which starts the guest agent. A
+            # guest other than Linux gets the v2 every cloud-init reads.
+            network = network_config(self._nic_macs(vmid, nics)) if guest_os != "linux" else None
+            if network:
+                network_filename = f"{vmid}-network-config.yaml"
+                self._write_snippet(
+                    storage_path, network_filename, yaml.safe_dump(network, sort_keys=False)
+                )
+                cicustom += f",network={snippet_storage}:snippets/{network_filename}"
 
             # Step 7: Configure Cloud-Init references and SSH keys
             _logger.info(f"Setting Cloud-Init config for VM {vmid}")
@@ -648,7 +735,7 @@ class ProxmoxVMProvisionMixin:
                 # this points at a snippets-only storage.
                 "ide2": f"{image_storage}:cloudinit",
                 "citype": "nocloud",
-                "cicustom": f"user={snippet_storage}:snippets/{filename}",
+                "cicustom": cicustom,
             }
 
             # Configure DHCP for NICs where enabled (default True for index 0, False otherwise)
@@ -736,6 +823,10 @@ class ProxmoxVMProvisionMixin:
             except Exception as e:
                 _logger.debug(f"VM {vmid} stop failed (may already be stopped): {e}")
 
+            # The snippets have to be known before the delete: afterwards the
+            # VM's config is gone (napalm-proxmox#15).
+            snippets = self._cicustom_snippets(vmid_int)
+
             # Step 2: Delete VM
             # Proxmox's API parameter is hyphenated (destroy-unreferenced-disks),
             # not a valid Python identifier — proxmoxer forwards kwargs to the
@@ -750,21 +841,12 @@ class ProxmoxVMProvisionMixin:
 
             # 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}")
+            for storage, filepath in snippets:
+                _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}")
 
             _logger.info(f"VM {vmid} destroyed successfully")
 
@@ -772,6 +854,24 @@ class ProxmoxVMProvisionMixin:
             _logger.exception(f"Failed to destroy VM {vmid}: {e}")
             raise
 
+    def _cicustom_snippets(self, vmid: int) -> list[tuple[str, str]]:
+        """``(storage, path)`` of every snippet the VM's cicustom names.
+
+        cicustom reads "user=local:snippets/101-user-data.yaml,network=...".
+        Best-effort: a VM whose config cannot be read has none to clean up.
+        """
+        try:
+            cicustom = self._node_api().qemu(vmid).config.get().get("cicustom", "")
+        except Exception as e:
+            _logger.debug(f"Could not read the snippets of VM {vmid}: {e}")
+            return []
+        snippets = []
+        for entry in cicustom.split(","):
+            storage, _, path = entry.partition("=")[2].partition(":")
+            if path.startswith("snippets/"):
+                snippets.append((storage, path))
+        return snippets
+
     def get_vm_status(
         self,
         vmid: str,
@@ -829,17 +929,18 @@ class ProxmoxVMProvisionMixin:
                     interfaces = (agent_info or {}).get("result", [])
 
                     # The guest agent does not report interfaces in a fixed order —
-                    # "lo" commonly comes first. Skip it and take the first real
-                    # NIC that has an IPv4 address.
+                    # loopback commonly comes first, named "lo" on Linux and
+                    # "lo0" on the BSDs. Skip loopback addresses, whatever the
+                    # interface is called, and take the first IPv4 address.
                     for iface in interfaces:
                         name = iface.get("name", "")
-                        if not name or name == "lo":
+                        if not name:
                             continue
                         for addr in iface.get("ip-addresses", []):
                             if addr.get("ip-address-type") != "ipv4":
                                 continue
                             ip_addr = addr.get("ip-address", "")
-                            if ip_addr:
+                            if ip_addr and not ipaddress.ip_address(ip_addr).is_loopback:
                                 _logger.info(f"VM {vmid} acquired IP {ip_addr}")
                                 return {
                                     "status": "running",

pyproject.toml

@@ -25,7 +25,7 @@ classifiers = [
 requires-python = ">=3.9"
 dependencies = [
     "napalm>=5.0.0",
-    "napalm_device_types>=2.4.0",
+    "napalm_device_types>=3.0.0",
     "paramiko>=5.0.0",  # CVE-2026-44405; imported directly for SSH fallback (driver.py)
     "proxmoxer>=2.0.0",
     "netaddr>=0.9.0",

tests/test_vm_guest_os.py

@@ -0,0 +1,276 @@
+"""Provisioning guests other than Linux: FreeBSD and OpenBSD (NetOrk/netork#794).
+
+What each guest needs was found on FreeBSD 15.1 and OpenBSD 7.9 cloud images
+(NetOrk/netork#793).
+"""
+
+from __future__ import annotations
+
+import base64
+from unittest.mock import MagicMock, patch
+
+import pytest
+import yaml
+from napalm_device_types.provisioning import QEMU_GUEST_AGENTS
+
+from napalm_proxmox.driver import ProxmoxDriver
+from napalm_proxmox.vm_provision_mixin import ProxmoxVMProvisionMixin
+
+_FREEBSD_URL = (
+    "https://download.freebsd.org/releases/VM-IMAGES/15.1-RELEASE/amd64/Latest/"
+    "FreeBSD-15.1-RELEASE-amd64-BASIC-CLOUDINIT-ufs.qcow2.xz"
+)
+_DEBIAN_URL = (
+    "https://cloud.debian.org/images/cloud/trixie/latest/debian-13-genericcloud-amd64.qcow2"
+)
+
+
+def _mixin(net0: str = "virtio=BC:24:11:AA:BB:02,bridge=vmbr0") -> tuple:
+    """A mixin whose node answers like Proxmox; the VM config carries *net0*."""
+    mixin = ProxmoxVMProvisionMixin()
+    mixin._node_name = "pve1"
+    api = MagicMock()
+    api.cluster.nextid.get.return_value = 101
+    api.storage.return_value.get.return_value = {"path": "/var/lib/vz"}
+    node = MagicMock()
+    node.storage.get.return_value = [
+        {"storage": "local-lvm", "type": "lvmthin", "content": "images,rootdir", "enabled": 1},
+        {"storage": "local", "type": "dir", "content": "snippets,iso", "enabled": 1},
+    ]
+    vm = MagicMock()
+    node.qemu.return_value = vm
+    vm.config.get.return_value = {"unused0": "local-lvm:vm-101-disk-0", "net0": net0}
+    vm.status.start.post.return_value = "UPID:pve1:start"
+    node.tasks.return_value.status.get.return_value = {"status": "stopped", "exitstatus": "OK"}
+    mixin._api = api
+    mixin._node_api = MagicMock(return_value=node)
+    mixin._download_cloud_image = MagicMock(
+        return_value="/var/lib/vz/template/netork-images/x.qcow2"
+    )
+    mixin._run_node_command = MagicMock(return_value="")
+    return mixin, node, vm
+
+
+def _create(mixin, guest_os: str, image_url: str = _DEBIAN_URL, **kwargs):
+    with patch("time.sleep"):
+        return mixin.create_vm_from_cloud_init(
+            name="bsd-vm",
+            image_url=image_url,
+            cpu=2,
+            memory=2048,
+            nics=[{"bridge": "vmbr0"}],
+            cloud_init_config={"hostname": "bsd-vm"},
+            guest_os=guest_os,
+            **kwargs,
+        )
+
+
+def _all_config_posts(vm) -> dict:
+    merged: dict = {}
+    for call in vm.config.post.call_args_list:
+        merged.update(call.kwargs)
+    return merged
+
+
+def _written_snippet(mixin, name: str) -> str | None:
+    """The content of the snippet *name* the driver wrote over SSH, if any."""
+    for call in mixin._run_node_command.call_args_list:
+        command = call.args[0]
+        if f"snippets/{name}" in command and "base64 -d" in command:
+            encoded = command.split("echo ", 1)[1].split(" |", 1)[0]
+            return base64.b64decode(encoded).decode()
+    return None
+
+
+class TestWhichGuestsProxmoxProvisions:
+    def test_all_qemu_guests(self):
+        assert ProxmoxDriver.GUEST_AGENTS == QEMU_GUEST_AGENTS
+
+    def test_an_unknown_guest_is_refused_before_anything_is_created(self):
+        mixin, node, _ = _mixin()
+        with pytest.raises(ValueError, match="windows"):
+            _create(mixin, "windows")
+        mixin._api.cluster.nextid.get.assert_not_called()
+        node.qemu.post.assert_not_called()
+
+
+class TestVmShellPerGuest:
+    @pytest.mark.parametrize(
+        ("guest_os", "ostype"), [("linux", "l26"), ("freebsd", "other"), ("openbsd", "other")]
+    )
+    def test_ostype(self, guest_os, ostype):
+        mixin, node, _ = _mixin()
+        _create(mixin, guest_os)
+        assert node.qemu.post.call_args.kwargs["ostype"] == ostype
+
+    @pytest.mark.parametrize("guest_os", ["linux", "freebsd"])
+    def test_agent_over_virtio_serial(self, guest_os):
+        mixin, node, _ = _mixin()
+        _create(mixin, guest_os)
+        shell = node.qemu.post.call_args.kwargs
+        assert shell["agent"] == "1"
+        assert "serial0" not in shell
+
+    def test_openbsd_agent_on_the_second_isa_serial_port(self):
+        """OpenBSD's qemu-ga cannot use virtio-serial, and the image keeps its
+        console on com0: the agent only answers as cua01, behind serial0."""
+        mixin, node, _ = _mixin()
+        _create(mixin, "openbsd")
+        shell = node.qemu.post.call_args.kwargs
+        assert shell["agent"] == "1,type=isa"
+        assert shell["serial0"] == "socket"
+
+
+class TestNetworkConfigPerGuest:
+    def test_linux_keeps_proxmoxs_own_network_config(self):
+        mixin, _, vm = _mixin()
+        _create(mixin, "linux")
+        config = _all_config_posts(vm)
+        assert config["cicustom"] == "user=local:snippets/101-user-data.yaml"
+        assert config["ipconfig0"] == "ip=dhcp"
+        assert _written_snippet(mixin, "101-network-config.yaml") is None
+
+    @pytest.mark.parametrize("guest_os", ["freebsd", "openbsd"])
+    def test_a_bsd_guest_gets_a_v2_network_config_matched_by_its_mac(self, guest_os):
+        """FreeBSD's nuageinit fails on the v1 Proxmox writes and then skips
+        runcmd, which is what starts the guest agent."""
+        mixin, _, vm = _mixin(net0="virtio=BC:24:11:AA:BB:02,bridge=vmbr0")
+        _create(mixin, guest_os)
+        config = _all_config_posts(vm)
+        assert config["cicustom"] == (
+            "user=local:snippets/101-user-data.yaml,network=local:snippets/101-network-config.yaml"
+        )
+        assert yaml.safe_load(_written_snippet(mixin, "101-network-config.yaml")) == {
+            "version": 2,
+            "ethernets": {"nic0": {"match": {"macaddress": "bc:24:11:aa:bb:02"}, "dhcp4": True}},
+        }
+
+
+class TestPackedImages:
+    """Proxmox's download-url unpacks ISOs only, so a packed image goes over SSH."""
+
+    def test_an_xz_image_is_unpacked_on_the_node_and_the_unpacked_file_imported(self):
+        mixin, _, _ = _mixin()
+        mixin._run_node_command = MagicMock(
+            side_effect=lambda cmd, timeout: "MISSING" if "test -f" in cmd else ""
+        )
+        mixin._download_cloud_image = MagicMock(
+            return_value="/var/lib/vz/template/netork-images/k-FreeBSD-15.1-ufs.qcow2.xz"
+        )
+        _create(mixin, "freebsd", image_url=_FREEBSD_URL, image_checksum="sha256:abc")
+
+        mixin._download_cloud_image.assert_called_once()
+        assert mixin._download_cloud_image.call_args.args[1] == "sha256:abc"  # the .xz is verified
+        commands = [c.args[0] for c in mixin._run_node_command.call_args_list]
+        unpack = next(c for c in commands if c.startswith("xz -dc "))
+        assert "k-FreeBSD-15.1-ufs.qcow2.xz" in unpack
+        importdisk = next(c for c in commands if c.startswith("qm importdisk"))
+        assert ".qcow2.xz" not in importdisk
+        assert ".qcow2 " in importdisk
+
+    def test_an_unpacked_image_already_on_the_node_is_not_downloaded_again(self):
+        mixin, _, _ = _mixin()
+        mixin._run_node_command = MagicMock(
+            side_effect=lambda cmd, timeout: "EXISTS" if "test -f" in cmd else ""
+        )
+        _create(mixin, "freebsd", image_url=_FREEBSD_URL)
+
+        mixin._download_cloud_image.assert_not_called()
+        commands = [c.args[0] for c in mixin._run_node_command.call_args_list]
+        assert not any(c.startswith("xz -dc ") for c in commands)
+
+    def test_a_packed_image_never_goes_through_an_import_storage(self):
+        mixin, node, _ = _mixin()
+        node.storage.get.return_value = [
+            {"storage": "local-lvm", "content": "images,rootdir", "enabled": 1},
+            {"storage": "local", "content": "snippets,import", "enabled": 1},
+        ]
+        mixin._run_node_command = MagicMock(
+            side_effect=lambda cmd, timeout: "MISSING" if "test -f" in cmd else ""
+        )
+        _create(mixin, "freebsd", image_url=_FREEBSD_URL)
+        # storage(name)("download-url").post(...) is the import-storage download.
+        node.storage.return_value.return_value.post.assert_not_called()
+        mixin._download_cloud_image.assert_called_once()
+
+
+class TestLoopbackIsNeverTheVmsAddress:
+    """OpenBSD's agent lists lo0 first (#793); Linux names it lo."""
+
+    def _status(self, interfaces: list) -> dict:
+        mixin = ProxmoxVMProvisionMixin()
+        node = MagicMock()
+        mixin._node_api = MagicMock(return_value=node)
+        node.qemu.return_value.config.get.return_value = {"net0": "virtio,bridge=vmbr0"}
+        node.qemu.return_value.agent.return_value.get.return_value = {"result": interfaces}
+        with patch("time.sleep"):
+            return mixin.get_vm_status("101", wait_for_ip=True, timeout=30, poll_interval=1)
+
+    def test_openbsd_answer(self):
+        """The raw answer of OpenBSD 7.9's qemu-ga, from the spike."""
+        result = self._status(
+            [
+                {
+                    "name": "lo0",
+                    "ip-addresses": [
+                        {"ip-address-type": "ipv6", "ip-address": "::1", "prefix": 128},
+                        {"ip-address-type": "ipv6", "ip-address": "fe80:3::1", "prefix": 64},
+                        {"ip-address-type": "ipv4", "ip-address": "127.0.0.1", "prefix": 8},
+                    ],
+                    "hardware-address": "00:00:00:00:00:00",
+                },
+                {
+                    "name": "vio0",
+                    "ip-addresses": [
+                        {"ip-address-type": "ipv4", "ip-address": "10.0.2.15", "prefix": 24}
+                    ],
+                    "hardware-address": "bc:24:11:aa:bb:04",
+                },
+                {"name": "enc0", "hardware-address": "00:00:00:00:00:00"},
+                {"name": "pflog0", "hardware-address": "00:00:00:00:00:00"},
+            ]
+        )
+        assert result["ip_address"] == "10.0.2.15"
+        assert result["mac_address"] == "bc:24:11:aa:bb:04"
+
+    def test_any_loopback_address_is_skipped_whatever_the_interface_is_called(self):
+        result = self._status(
+            [
+                {
+                    "name": "lo1",
+                    "ip-addresses": [{"ip-address-type": "ipv4", "ip-address": "127.0.0.2"}],
+                },
+                {
+                    "name": "vtnet0",
+                    "ip-addresses": [{"ip-address-type": "ipv4", "ip-address": "10.0.0.5"}],
+                },
+            ]
+        )
+        assert result["ip_address"] == "10.0.0.5"
+
+
+class TestDestroyRemovesEverySnippet:
+    def test_user_and_network_snippets_are_read_before_the_vm_is_deleted(self):
+        """After the delete the config is gone, and the snippets stayed behind
+        (napalm-proxmox#15)."""
+        mixin = ProxmoxVMProvisionMixin()
+        node = MagicMock()
+        mixin._node_api = MagicMock(return_value=node)
+        vm = node.qemu.return_value
+        order: list[str] = []
+        vm.config.get.side_effect = lambda: (
+            order.append("config")
+            or {
+                "cicustom": "user=local:snippets/101-user-data.yaml,"
+                "network=local:snippets/101-network-config.yaml"
+            }
+        )
+        vm.delete.side_effect = lambda **kw: order.append("delete")
+        node.tasks.return_value.status.get.return_value = {"status": "stopped", "exitstatus": "OK"}
+
+        with patch("time.sleep"):
+            mixin.destroy_vm("101")
+
+        assert order.index("config") < order.index("delete")
+        deleted = [c.args[0] for c in node.storage.return_value.content.call_args_list]
+        assert deleted == ["snippets/101-user-data.yaml", "snippets/101-network-config.yaml"]