feat(vm-provision): provision FreeBSD and OpenBSD guests

create_vm_from_cloud_init(guest_os=...) with GUEST_AGENTS from
napalm-device-types 3.0 (QEMU_GUEST_AGENTS: Linux, FreeBSD, OpenBSD). An
unknown guest_os is refused before anything is created. Found on FreeBSD
15.1 and OpenBSD 7.9 cloud images (NetOrk/netork#793):

- OS type `other` for the BSDs. OpenBSD's qemu-ga only works over ISA
  serial, and only as the second port: the image keeps its console on
  com0, so the VM gets `serial0: socket` next to `agent: 1,type=isa`.
- A BSD guest gets a network-config v2 snippet of its own
  (`cicustom: user=...,network=...`, MAC read back from net0). FreeBSD's
  nuageinit fails on the v1 Proxmox generates and then skips runcmd,
  which starts the guest agent. Linux keeps Proxmox's own.
- Packed images (FreeBSD's .qcow2.xz) are unpacked on the node after the
  packed file's checksum is verified; only the unpacked file is cached.
  download-url unpacks ISOs only, so they always take the SSH path.

Fixed on the way:
- get_vm_status skipped loopback only when it was called `lo`; OpenBSD's
  agent lists `lo0` first, so 127.0.0.1 would have been the VM's IP.
  Loopback is now recognised by its address.
- destroy_vm read cicustom after deleting the VM, when its config was gone,
  so snippets stayed on the node; it now reads them first and removes all
  of them (#15).

.github/workflows/ci.yml

@@ -0,0 +1,48 @@
+name: CI
+
+on:
+  push:
+    branches: ["**"]
+  pull_request:
+    branches: ["**"]
+
+jobs:
+  test:
+    runs-on: ubuntu-latest
+    strategy:
+      fail-fast: false
+      matrix:
+        python-version: ["3.10", "3.11", "3.12"]
+    steps:
+      - name: Checkout
+        uses: actions/checkout@v4
+
+      - name: Setup Python
+        uses: actions/setup-python@v5
+        with:
+          python-version: ${{ matrix.python-version }}
+          cache: pip
+
+      - name: Install package with dev extras
+        run: |
+          python -m pip install --upgrade pip
+          # napalm-device-types lives in git.netork.io/NAPALM, not on PyPI: without this
+          # pip looks there, finds an unrelated 0.1.0 and the job dies before any test.
+          python -m pip install "napalm-device-types @ git+https://git.netork.io/NAPALM/napalm-device-types.git"
+          python -m pip install -e ".[dev]"
+
+      - name: Run unit tests
+        run: |
+          python -m pytest -q --tb=short
+
+      - name: Build wheel and sdist
+        run: |
+          python -m pip install build
+          python -m build
+
+      - name: Upload dist artifacts
+        # v4 refuses to run on Gitea ("not currently supported on GHES").
+        uses: actions/upload-artifact@v3
+        with:
+          name: dist-${{ matrix.python-version }}
+          path: dist/*

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/driver.py

@@ -34,7 +34,15 @@ from typing import Any
 
 logger = logging.getLogger(__name__)
 
-from napalm_device_types import FingerprintRule, HypervisorDriver, PortSpec
+from napalm_device_types import (
+    FingerprintRule,
+    HostStatusMixin,
+    HypervisorDriver,
+    KernelFactsMixin,
+    ListeningSocketsMixin,
+    PortSpec,
+    SystemdServicesMixin,
+)
 from napalm.base.exceptions import ConnectionException
 
 try:
@@ -76,6 +84,10 @@ class ProxmoxDriver(
     ProxmoxVMProvisionMixin,
     ProxmoxRoutingMixin,
     ProxmoxSystemMixin,
+    KernelFactsMixin,
+    ListeningSocketsMixin,
+    SystemdServicesMixin,
+    HostStatusMixin,
     HypervisorDriver,
 ):
     """NAPALM driver for Proxmox VE nodes."""
@@ -86,6 +98,10 @@ class ProxmoxDriver(
     USES_SSH = False
     # A PVE node reboots through a full init sequence plus storage checks.
     REBOOT_SETTLE_SECONDS = 90
+    #: "Upgrade everything" must be a full upgrade on Proxmox VE: a plain
+    #: ``apt-get upgrade`` holds back what needs new or removed packages and can
+    #: leave the node half updated. netOrk reads this when it upgrades the host.
+    FULL_UPGRADE = True
     PORT_SPECS = [
         PortSpec("https", 8006, weight=8.0),
     ]

napalm_proxmox/system_mixin.py

@@ -20,6 +20,8 @@ import logging
 import re
 from typing import Any
 
+from napalm_device_types import APT_UPGRADABLE_COMMAND, parse_apt_upgradable
+
 from napalm_proxmox import utils
 
 logger = logging.getLogger(__name__)
@@ -221,6 +223,23 @@ class ProxmoxSystemMixin:
 
         return result
 
+    # ------------------------------------------------------------------ #
+    # Kernel facts (KernelFactsMixin supplies get_kernel_facts)
+    # ------------------------------------------------------------------ #
+
+    def _run_kernel_facts_command(self, command: str) -> str:
+        """The transport for ``KernelFactsMixin.get_kernel_facts``: the exec path."""
+        return self._exec_ssh_command(command)
+
+    # ------------------------------------------------------------------ #
+    # Listening sockets (ListeningSocketsMixin supplies get_listening_sockets)
+    # ------------------------------------------------------------------ #
+
+    def _run_listening_sockets_command(self, command: str, *, privileged: bool) -> str:
+        """The transport for ``ListeningSocketsMixin.get_listening_sockets``:
+        the exec path, which runs as root either way."""
+        return str(self._exec_ssh_command(command))
+
     # ------------------------------------------------------------------ #
     # Packages (Debian APT)
     # ------------------------------------------------------------------ #
@@ -338,101 +357,63 @@ class ProxmoxSystemMixin:
         return warnings
 
     # ------------------------------------------------------------------ #
-    # Services (systemd)
+    # Services (systemd, through napalm-device-types' SystemdServicesMixin)
     # ------------------------------------------------------------------ #
 
-    def get_services(self) -> list[_JsonDict]:
+    def _run_service_command(self, command: str, *, privileged: bool, timeout: int) -> str:
-        """Return systemd services with running and enabled state.
+        """The transport for ``SystemdServicesMixin``: the exec path, as root.
 
-        Uses two ``systemctl`` invocations combined in a single SSH command:
+        *privileged* needs nothing more on a node the driver reaches as root, and
-        - ``list-unit-files`` for the static enabled/disabled state
+        the exec path keeps its own timeout.
-        - ``list-units`` for the live running state
         """
-        raw = self._exec_ssh_command(
+        return str(self._exec_ssh_command(command))
-            "{ systemctl list-unit-files --type=service --no-pager --no-legend --full 2>/dev/null;"
-            " echo '---UNITS---';"
-            " systemctl list-units --type=service --all --no-pager --no-legend --full 2>/dev/null;"
-            " } || true"
-        )
-
-        # Parse enabled state from list-unit-files
-        enabled_map: dict[str, bool] = {}
-        section = "files"
-        for line in raw.splitlines():
-            if line.strip() == "---UNITS---":
-                section = "units"
-                continue
-            parts = line.strip().split(None, 1)
-            if len(parts) < 1:
-                continue
-            unit = parts[0].lstrip("\u25cf").strip()
-            if not unit.endswith(".service"):
-                continue
-            name = unit[: -len(".service")]
-            if section == "files":
-                state = parts[1].strip() if len(parts) > 1 else ""
-                enabled_map[name] = state in ("enabled", "enabled-runtime", "static")
-
-        # Parse running state from list-units
-        running_map: dict[str, bool] = {}
-        section = "files"
-        for line in raw.splitlines():
-            if line.strip() == "---UNITS---":
-                section = "units"
-                continue
-            if section != "units":
-                continue
-            parts = line.strip().lstrip("\u25cf").strip().split(None, 4)
-            if len(parts) < 4:
-                continue
-            unit = parts[0]
-            if not unit.endswith(".service"):
-                continue
-            name = unit[: -len(".service")]
-            sub_state = parts[3]
-            running_map[name] = sub_state == "running"
-
-        all_names = sorted(set(enabled_map) | set(running_map))
-        return [
-            {
-                "name": name,
-                "running": running_map.get(name, False),
-                "enabled": enabled_map.get(name, False),
-                "pid": 0,
-            }
-            for name in all_names
-        ]
-
-    def manage_service(self, name: str, action: str) -> _JsonDict:
-        """Start / stop / restart / enable / disable a systemd service."""
-        if not re.match(r'^[a-zA-Z0-9_\-\.@]+$', name):
-            raise ValueError(f"Invalid service name: {name!r}")
-        if action not in ('start', 'stop', 'restart', 'enable', 'disable'):
-            raise ValueError(f"Invalid action: {action!r}")
-        output = self._exec_ssh_command(f"systemctl {action} {name}.service 2>&1 || true")
-        return {"success": True, "output": output}
 
     # ------------------------------------------------------------------ #
     # Available updates
     # ------------------------------------------------------------------ #
 
     def get_available_updates(self) -> list[_JsonDict]:
-        """Return list of upgradable packages from the Proxmox APT API."""
+        """Return the node's upgradable packages, with origin and security status.
-        updates: list[_JsonDict] = []
+
+        ``apt list --upgradable`` over the exec path names each candidate's suite
+        (``trixie-security``); the APT API names only an Origin ("Debian",
+        "Proxmox") and is the fallback, with the security status unknown.
+
+        :raises Exception: when neither answers -- never an empty list for
+            "could not read".
+        """
         try:
-            for upd in self._api.nodes(self._node_name).apt.update.get():
+            updates = parse_apt_upgradable(self._exec_ssh_command(APT_UPGRADABLE_COMMAND) or "")
-                pkg = upd.get("Package", "")
+        except ValueError as exc:
-                if not pkg:
+            logger.debug("apt list over the exec path failed, using the API: %s", exc)
-                    continue
+            updates = self._updates_from_api()
-                updates.append({
-                    "name": pkg,
-                    "current_version": upd.get("OldVersion", ""),
-                    "new_version": upd.get("Version", ""),
-                })
-        except Exception as exc:
-            logger.debug("Failed to fetch available updates: %s", exc)
         return sorted(updates, key=lambda u: u["name"])
 
+    def _updates_from_api(self) -> list[_JsonDict]:
+        return [
+            {
+                "name": upd["Package"],
+                "current_version": upd.get("OldVersion", ""),
+                "new_version": upd.get("Version", ""),
+                "origin": upd.get("Origin"),
+                "security": None,
+            }
+            for upd in self._api.nodes(self._node_name).apt.update.get()  # type: ignore[union-attr]
+            if upd.get("Package")
+        ]
+
+    def refresh_available_updates(self) -> _JsonDict:
+        """Resynchronise the node's package index (``POST nodes/{n}/apt/update``)."""
+        try:
+            task = self._api.nodes(self._node_name).apt.update.post()  # type: ignore[union-attr]
+        except Exception as exc:
+            return {"success": False, "output": str(exc)}
+        return {"success": True, "output": f"Package index refresh started ({task})"}
+
+    def _run_host_status_command(self, command: str) -> str:
+        """The transport for ``HostStatusMixin.get_host_status``: the exec path."""
+        return str(self._exec_ssh_command(command))
+
     def apply_updates(self, packages: list[str]) -> _JsonDict:
         """Upgrade the given packages via ``apt-get install`` over SSH."""
         for pkg in packages:

napalm_proxmox/vm_provision_mixin.py

@@ -4,22 +4,74 @@ 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 typing import 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
+    # pyproject.toml, and nothing here needs it at runtime.
+    from napalm_device_types.models import VMCpuTypeDict
 
 _logger = logging.getLogger(__name__)
 
+# The CPU models a new VM may be given. Each lists the /proc/cpuinfo flags it
+# adds on top of QEMU's qemu64 baseline: what the node's CPU must have for the
+# model to start at all, and what a guest can count on. The sets follow
+# Proxmox's own x86-64-v* definitions (qemu-server, PVE/QemuServer/CPUConfig.pm).
+#
+# Leaving the model out of qemu.post is not neutral: Proxmox then falls back to
+# kvm64, which lacks even AES-NI, let alone the AVX MongoDB 5.0+ needs
+# (netOrk#494). The default below is what the Proxmox GUI picks since PVE 8.
+_X86_64_V2_AES_FLAGS = ("aes", "popcnt", "pni", "sse4_1", "sse4_2", "ssse3")
+_X86_64_V3_FLAGS = _X86_64_V2_AES_FLAGS + (
+    "avx",
+    "avx2",
+    "bmi1",
+    "bmi2",
+    "f16c",
+    "fma",
+    "abm",
+    "movbe",
+    "xsave",
+)
+_DEFAULT_CPU_TYPE = "x86-64-v2-AES"
+_CPU_MODELS = (
+    (
+        "x86-64-v2-AES",
+        _X86_64_V2_AES_FLAGS,
+        "Proxmox's own default: runs on practically any x86-64 server CPU and "
+        "can live-migrate between different ones. No AVX.",
+    ),
+    (
+        "x86-64-v3",
+        _X86_64_V3_FLAGS,
+        "Adds AVX and AVX2 (which MongoDB 5.0 and later need). Every node the VM "
+        "may run on needs an Intel Haswell or AMD Excavator CPU (2013) or newer.",
+    ),
+)
+_HOST_CPU_DESCRIPTION = (
+    "This node's CPU, passed through unchanged: fastest, with every feature it "
+    "has, but the VM can only live-migrate to nodes with the same CPU."
+)
+
 # Downloaded cloud images are cached here on the hypervisor node, keyed by
 # filename, so provisioning multiple VMs from the same image only pays the
 # download cost once.
@@ -31,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\-.+=_]")
@@ -66,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.
@@ -169,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".
 
@@ -231,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",
@@ -351,6 +474,56 @@ class ProxmoxVMProvisionMixin:
             )
         return targets
 
+    def get_vm_cpu_types(self) -> list[VMCpuTypeDict]:
+        """List the CPU models a new VM may be given, judged against this node's CPU."""
+        return self._cpu_types_for(self._node_cpu_flags())
+
+    def _node_cpu_flags(self) -> set[str]:
+        status = self._node_api().status.get() or {}
+        return set(str((status.get("cpuinfo") or {}).get("flags", "")).split())
+
+    @staticmethod
+    def _cpu_types_for(node_flags: set[str]) -> list[VMCpuTypeDict]:
+        types: list[VMCpuTypeDict] = [
+            {
+                "name": name,
+                "description": description,
+                "features": list(flags),
+                "available": set(flags) <= node_flags,
+                "default": name == _DEFAULT_CPU_TYPE,
+            }
+            for name, flags, description in _CPU_MODELS
+        ]
+        types.append(
+            {
+                "name": "host",
+                "description": _HOST_CPU_DESCRIPTION,
+                "features": sorted(node_flags),
+                "available": True,
+                "default": False,
+            }
+        )
+        return types
+
+    def _resolve_cpu_type(self, cpu_type: str | None) -> str:
+        """The model to create the VM with. A model asked for by name is checked
+        against this node's CPU first: one it cannot run would fail only at VM
+        start, after the disk import, leaving a half-built VM behind."""
+        if cpu_type is None:
+            return _DEFAULT_CPU_TYPE
+        node_flags = self._node_cpu_flags()
+        offered = {t["name"]: t for t in self._cpu_types_for(node_flags)}
+        entry = offered.get(cpu_type)
+        if entry is None:
+            raise ValueError(f"Unknown CPU type {cpu_type!r}; choose one of {', '.join(offered)}")
+        if not entry["available"]:
+            missing = sorted(set(entry["features"]) - node_flags)
+            raise ValueError(
+                f"CPU type {cpu_type!r} needs {', '.join(missing)}, which the CPU of "
+                f"node {self._node_name} does not have"
+            )
+        return cpu_type
+
     def _wait_for_task(self, upid: str, timeout: int = 120) -> None:
         """
         Poll a Proxmox task until completion.
@@ -395,6 +568,8 @@ class ProxmoxVMProvisionMixin:
         ssh_public_keys: List[str] | None = None,
         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:
@@ -427,6 +602,11 @@ class ProxmoxVMProvisionMixin:
             disk_resize_gb: resize root disk to this size (None = no resize)
             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
 
@@ -435,10 +615,18 @@ class ProxmoxVMProvisionMixin:
 
         Raises:
             RuntimeError: provisioning failure (download, import, config, timeout, etc.)
-            ValueError: invalid storage or configuration
+            ValueError: invalid storage or configuration, or a cpu_type that is
+                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
             next_vmid = self._api.cluster.nextid.get()
@@ -452,12 +640,12 @@ class ProxmoxVMProvisionMixin:
                 name=name,
                 memory=memory,
                 cores=cpu,
-                ostype="l26",
+                cpu=cpu_model,
                 scsihw="virtio-scsi-pci",
-                # Without this, Proxmox never attaches the virtio-serial
+                # Includes agent: without it Proxmox never attaches the
-                # channel the QEMU guest agent needs — get_vm_status's
+                # channel the QEMU guest agent needs — get_vm_status's agent
-                # agent queries (below) would have nothing to talk to.
+                # queries (below) would have nothing to talk to.
-                agent="1",
+                **_GUEST_VM_SETTINGS[guest_os],
             )
 
             # Step 3: Download cloud image (cached) and import as root disk
@@ -520,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._write_snippet(storage_path, filename, user_data_yaml)
-            self._run_node_command(
+            cicustom = f"user={snippet_storage}:snippets/{filename}"
-                f"mkdir -p {storage_path}/snippets && "
+
-                f"echo {encoded} | base64 -d > {storage_path}/snippets/{filename}",
+            # Proxmox writes network-config v1, and FreeBSD's nuageinit fails
-                timeout=30,
+            # 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}")
@@ -547,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)
@@ -635,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
@@ -649,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:
+            for storage, filepath in snippets:
-                config = self._node_api().qemu(vmid_int).config.get()
+                _logger.debug(f"Deleting snippet {filepath} from {storage}")
-                cicustom = config.get("cicustom", "")
+                try:
-                if "snippets/" in cicustom:
+                    self._node_api().storage(storage).content(filepath).delete()
-                    parts = cicustom.split("=")
+                except Exception as e:
-                    if len(parts) >= 2:
+                    _logger.warning(f"Failed to delete snippet {filepath}: {e}")
-                        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")
 
@@ -671,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,
@@ -728,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
+                    # loopback commonly comes first, named "lo" on Linux and
-                    # NIC that has an IPv4 address.
+                    # "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.0.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_full_upgrade.py

@@ -0,0 +1,11 @@
+"""Proxmox VE says that "upgrade everything" must be a full upgrade on it.
+
+A plain ``apt-get upgrade`` can leave a node half updated; Proxmox documents
+``apt full-upgrade``. netOrk reads ``FULL_UPGRADE`` to choose.
+"""
+
+from napalm_proxmox.driver import ProxmoxDriver
+
+
+def test_a_full_upgrade_is_declared():
+    assert ProxmoxDriver.FULL_UPGRADE is True

tests/test_kernel_facts.py

@@ -0,0 +1,46 @@
+"""`get_kernel_facts`: what the node's kernel has built and loaded.
+
+A Proxmox node runs its own kernel under every guest, which makes it the host
+where a kernel CVE's preconditions matter most. The command and its parse are
+napalm-device-types'; the driver only carries the command over its exec path.
+"""
+
+from __future__ import annotations
+
+import base64
+import gzip
+from unittest.mock import patch
+
+import pytest
+
+from napalm_device_types import KernelFactsMixin
+from napalm_device_types.kernel import KERNEL_FACTS_COMMAND
+from napalm_proxmox.driver import ProxmoxDriver
+
+REPORT = (
+    "[release]\n6.8.12-4-pve\n[loaded]\nkvm_intel\n[builtin]\nkernel/net/ipv4/tcp_cubic.ko\n"
+    "[available]\nkernel/net/tipc/tipc.ko\n[config]\nCONFIG_TIPC=m\n"
+)
+WIRE = "KFACTS_BEGIN\n" + base64.encodebytes(gzip.compress(REPORT.encode())).decode() + "KFACTS_END"
+
+
+def test_the_driver_declares_the_contract():
+    assert issubclass(ProxmoxDriver, KernelFactsMixin)
+
+
+def test_it_runs_the_shared_command(driver):
+    with patch.object(driver, "_exec_ssh_command", return_value=WIRE) as exec_:
+        facts = driver.get_kernel_facts()
+
+    exec_.assert_called_once_with(KERNEL_FACTS_COMMAND)
+    assert facts["release"] == "6.8.12-4-pve"
+    assert facts["loaded"] == ["kvm_intel"]
+    assert facts["builtin"] == ["tcp_cubic"]
+    assert facts["available"] == ["tipc"]
+    assert facts["config"] == {"CONFIG_TIPC": "m"}
+
+
+def test_output_without_a_report_raises(driver):
+    with patch.object(driver, "_exec_ssh_command", return_value=""):
+        with pytest.raises(ValueError):
+            driver.get_kernel_facts()

tests/test_listening_sockets.py

@@ -0,0 +1,37 @@
+"""`get_listening_sockets`: what listens on the node, and which service it is.
+
+Whether pveproxy, a Ceph manager or a guest-facing service is reachable from
+outside the node is decided by the address it listens on. The command and its
+parse are napalm-device-types'; the driver only carries the command over its
+exec path, which already runs as root.
+"""
+
+from __future__ import annotations
+
+from unittest.mock import patch
+
+from napalm_device_types import ListeningSocketsMixin
+from napalm_proxmox.driver import ProxmoxDriver
+
+WIRE = (
+    "SOCK_BEGIN\n[ss]\n"
+    'tcp LISTEN 0 4096 *:8006 *:* users:(("pveproxy worker",pid=2101,fd=6))\n'
+    "__SS_RC=0\n[cgroups]\n"
+    "2101 0::/system.slice/pveproxy.service\n"
+    "SOCK_END\n"
+)
+
+
+def test_the_driver_declares_the_contract():
+    assert issubclass(ProxmoxDriver, ListeningSocketsMixin)
+
+
+def test_it_reads_as_root_over_the_exec_path(driver):
+    with patch.object(driver, "_exec_ssh_command", return_value=WIRE) as exec_:
+        reading = driver.get_listening_sockets()
+
+    [command] = [c.args[0] for c in exec_.call_args_list]
+    assert command.startswith("sh -c '")
+    assert reading["attributed"] is True
+    [socket] = reading["sockets"]
+    assert (socket["address"], socket["port"], socket["unit"]) == ("*", 8006, "pveproxy")

tests/test_services.py

@@ -0,0 +1,68 @@
+"""Services on a Proxmox node: systemd, through napalm-device-types' mixin.
+
+The listing used to read ``list-unit-files``' second column, which since
+systemd 245 is followed by a preset column -- so ``enabled`` was false for
+every service on every node (#6). An action ended in ``|| true`` and reported
+success whatever happened. Both now come from the shared mixin; the driver
+only carries the command over its exec path, as root.
+"""
+
+from __future__ import annotations
+
+from unittest.mock import patch
+
+import pytest
+from napalm_device_types import SystemdServicesMixin
+from napalm_device_types.systemd import SYSTEMD_SERVICES_COMMAND, service_action_command
+
+from napalm_proxmox.driver import ProxmoxDriver
+
+REPORT = (
+    "SVC_BEGIN\n[files]\npveproxy.service enabled enabled\n[units]\n"
+    "MainPID=1234\nId=pveproxy.service\nNames=pveproxy.service\nLoadState=loaded\n"
+    "ActiveState=active\nSubState=running\nUnitFileState=enabled\n"
+    "[generated]\nSVC_END\n"
+)
+
+
+def test_the_driver_uses_the_shared_mixin():
+    assert issubclass(ProxmoxDriver, SystemdServicesMixin)
+    assert ProxmoxDriver.get_services is SystemdServicesMixin.get_services
+    assert ProxmoxDriver.manage_service is SystemdServicesMixin.manage_service
+
+
+def test_listing_runs_the_shared_command(driver):
+    with patch.object(driver, "_exec_ssh_command", return_value=REPORT) as exec_:
+        services = driver.get_services()
+
+    exec_.assert_called_once_with(SYSTEMD_SERVICES_COMMAND)
+    assert services == [{"name": "pveproxy", "running": True, "enabled": True, "pid": 1234}]
+
+
+def test_an_unreadable_listing_raises_instead_of_reporting_no_services(driver):
+    with patch.object(driver, "_exec_ssh_command", return_value=""):
+        with pytest.raises(ValueError):
+            driver.get_services()
+
+
+def test_an_action_reports_its_real_outcome(driver):
+    failed = "Failed to restart nope.service: Unit nope.service not found.\n__SVC_RC=5"
+    with patch.object(driver, "_exec_ssh_command", return_value=failed) as exec_:
+        result = driver.manage_service("nope", "restart")
+
+    exec_.assert_called_once_with(service_action_command("nope", "restart"))
+    assert result["success"] is False
+    assert "not found" in result["output"]
+
+
+def test_a_successful_action(driver):
+    with patch.object(driver, "_exec_ssh_command", return_value="__SVC_RC=0"):
+        assert driver.manage_service("pveproxy", "restart") == {"success": True, "output": ""}
+
+
+def test_an_invalid_name_never_reaches_the_node(driver):
+    with patch.object(driver, "_exec_ssh_command") as exec_:
+        with pytest.raises(ValueError):
+            driver.manage_service("pveproxy; reboot", "stop")
+
+    exec_.assert_not_called()

tests/test_updates.py

@@ -0,0 +1,120 @@
+"""Pending updates on a Proxmox node: from where, whether they are security fixes,
+and whether the node needs a reboot.
+
+The node's APT API names only an Origin ("Debian", "Proxmox"), the same for the
+main and the security archive. ``apt list --upgradable`` over the exec path
+names the suite (``trixie-security``), so that is read first; the API remains
+the fallback, with the security status left unknown. A reader that cannot read
+raises: an empty list would tell netOrk that nothing is pending.
+"""
+
+from __future__ import annotations
+
+from unittest.mock import MagicMock, patch
+
+import pytest
+from napalm_device_types import HostStatusMixin
+from napalm_device_types.host_status import HOST_STATUS_COMMAND
+from napalm_device_types.package_updates import APT_UPGRADABLE_COMMAND
+
+from napalm_proxmox.driver import ProxmoxDriver
+
+APT = (
+    "libssl3t64/stable-security 3.5.1-1+deb13u2 amd64 [upgradable from: 3.5.1-1+deb13u1]\n"
+    "ceph-common/stable 20.2.4-pve5 amd64 [upgradable from: 20.2.4-pve4]\n"
+)
+API_ENTRY = {
+    "Package": "librados2",
+    "OldVersion": "20.2.4-pve4",
+    "Version": "20.2.4-pve5",
+    "Origin": "Proxmox",
+}
+
+
+def _api_updates(driver, entries=None, error=None):
+    api = MagicMock()
+    getter = api.nodes.return_value.apt.update.get
+    if error:
+        getter.side_effect = error
+    else:
+        getter.return_value = entries or []
+    driver._api = api
+    return api
+
+
+class TestAvailableUpdates:
+    def test_apt_over_the_exec_path_names_the_suite(self, driver):
+        with patch.object(driver, "_exec_ssh_command", return_value=APT + "__APT_RC=0\n") as exec_:
+            updates = {u["name"]: u for u in driver.get_available_updates()}
+
+        exec_.assert_called_once_with(APT_UPGRADABLE_COMMAND)
+        assert updates["libssl3t64"]["security"] is True
+        assert updates["ceph-common"]["security"] is False
+        assert updates["ceph-common"]["origin"] == "stable"
+
+    def test_the_api_is_the_fallback_with_security_unknown(self, driver):
+        _api_updates(driver, [API_ENTRY])
+        with patch.object(driver, "_exec_ssh_command", return_value=""):
+            updates = driver.get_available_updates()
+
+        assert updates == [
+            {
+                "name": "librados2",
+                "current_version": "20.2.4-pve4",
+                "new_version": "20.2.4-pve5",
+                "origin": "Proxmox",
+                "security": None,
+            }
+        ]
+
+    def test_a_failed_apt_falls_back_too(self, driver):
+        _api_updates(driver, [API_ENTRY])
+        with patch.object(driver, "_exec_ssh_command", return_value="E: lock\n__APT_RC=100\n"):
+            assert [u["name"] for u in driver.get_available_updates()] == ["librados2"]
+
+    def test_nothing_readable_raises_instead_of_reporting_nothing(self, driver):
+        _api_updates(driver, error=RuntimeError("API timeout"))
+        with patch.object(driver, "_exec_ssh_command", return_value=""):
+            with pytest.raises(RuntimeError):
+                driver.get_available_updates()
+
+    def test_nothing_pending_is_an_empty_list(self, driver):
+        with patch.object(driver, "_exec_ssh_command", return_value="__APT_RC=0\n"):
+            assert driver.get_available_updates() == []
+
+
+class TestRefresh:
+    def test_the_node_refreshes_its_index_through_the_api(self, driver):
+        api = _api_updates(driver)
+        api.nodes.return_value.apt.update.post.return_value = "UPID:pve1:0001"
+
+        result = driver.refresh_available_updates()
+
+        assert result["success"] is True
+        api.nodes.return_value.apt.update.post.assert_called_once()
+
+    def test_a_refused_refresh_says_why(self, driver):
+        api = _api_updates(driver)
+        api.nodes.return_value.apt.update.post.side_effect = RuntimeError(
+            "403 Permission check failed"
+        )
+
+        result = driver.refresh_available_updates()
+
+        assert result == {"success": False, "output": "403 Permission check failed"}
+
+
+class TestHostStatus:
+    def test_the_node_is_read_over_the_exec_path(self, driver):
+        report = (
+            "HSTAT_BEGIN\n[kernel]\n7.0.14-19-pve\n[modules]\n7.0.14-19-pve\n7.0.2-6-pve\n"
+            "[timers]\napt-daily-upgrade.timer enabled\nHSTAT_END\n"
+        )
+        with patch.object(driver, "_exec_ssh_command", return_value=report) as exec_:
+            status = driver.get_host_status()
+
+        exec_.assert_called_once_with(HOST_STATUS_COMMAND)
+        assert status["reboot_required"] is False
+
+    def test_the_driver_declares_the_contract(self):
+        assert issubclass(ProxmoxDriver, HostStatusMixin)

tests/test_vm_cpu_types.py

@@ -0,0 +1,153 @@
+"""Which virtual CPU model a new VM gets.
+
+Created without a ``cpu`` argument, a Proxmox VM falls back to ``kvm64``:
+no AVX, no AES-NI. MongoDB 5.0 and later will not even start on it, which is
+how a Graylog provisioned through netOrk failed (netOrk#494). The driver now
+always names a model, defaulting to ``x86-64-v2-AES`` as the Proxmox GUI does,
+and lists the alternatives with what each needs from the node's CPU.
+
+The flag sets below are trimmed from real ``/nodes/{node}/status`` answers in a
+mixed cluster: a Celeron J3455 (no AVX at all) and an i7-7700 (AVX2).
+"""
+
+from __future__ import annotations
+
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from napalm_proxmox.vm_provision_mixin import ProxmoxVMProvisionMixin
+
+CELERON_J3455 = (
+    "fpu vme de pse tsc msr pae mce cx8 apic sep mtrr pge mca cmov pat pse36 clflush "
+    "mmx fxsr sse sse2 ss ht syscall nx pdpe1gb rdtscp lm constant_tsc pni pclmulqdq "
+    "ssse3 cx16 sse4_1 sse4_2 x2apic movbe popcnt aes rdrand lahf_lm 3dnowprefetch "
+    "erms mpx rdseed smap clflushopt sha_ni xsaveopt xsavec xgetbv1"
+)
+CORE_I7_7700 = (
+    "fpu vme de pse tsc msr pae mce cx8 apic sep mtrr pge mca cmov pat pse36 clflush "
+    "mmx fxsr sse sse2 ss ht syscall nx pdpe1gb rdtscp lm constant_tsc pni pclmulqdq "
+    "ssse3 fma cx16 sse4_1 sse4_2 x2apic movbe popcnt aes xsave avx f16c rdrand "
+    "lahf_lm abm 3dnowprefetch fsgsbase bmi1 hle avx2 smep bmi2 erms invpcid rtm mpx "
+    "rdseed adx smap clflushopt xsaveopt xsavec xgetbv1 xsaves"
+)
+
+
+def _mixin_on(flags: str) -> tuple[ProxmoxVMProvisionMixin, MagicMock, MagicMock]:
+    """A mixin whose node reports `flags` and that can run a full create."""
+    mixin = ProxmoxVMProvisionMixin()
+    mixin._node_name = "pve1"
+
+    api = MagicMock()
+    api.cluster.nextid.get.return_value = 120
+    api.storage.return_value.get.return_value = {"path": "/var/lib/vz"}
+
+    node = MagicMock()
+    node.status.get.return_value = {"cpuinfo": {"model": "test", "flags": flags}}
+    node.storage.get.return_value = [
+        {"storage": "local-lvm", "type": "lvmthin", "content": "images,rootdir", "enabled": 1},
+        {"storage": "snippets", "type": "dir", "content": "snippets", "enabled": 1},
+    ]
+    vm = MagicMock()
+    vm.config.get.return_value = {
+        "unused0": "local-lvm:vm-120-disk-0",
+        "scsi0": "local-lvm:vm-120-disk-0",
+    }
+    vm.status.start.post.return_value = "UPID:pve1:1:start"
+    node.qemu.return_value = vm
+    task = MagicMock()
+    task.status.get.return_value = {"status": "stopped", "exitstatus": "OK"}
+    node.tasks.return_value = task
+
+    mixin._api = api
+    mixin._node_api = MagicMock(return_value=node)
+    mixin._download_cloud_image = MagicMock(return_value="/var/lib/vz/template/x.qcow2")
+    mixin._run_node_command = MagicMock(return_value="")
+    return mixin, api, node
+
+
+def _create(mixin: ProxmoxVMProvisionMixin, **kwargs):
+    with patch("time.sleep"):
+        return mixin.create_vm_from_cloud_init(
+            name="graylog-01",
+            image_url="https://cloud-images.ubuntu.com/noble/current/noble-server-cloudimg-amd64.img",
+            cpu=2,
+            memory=4096,
+            nics=[{"bridge": "vmbr0"}],
+            cloud_init_config={"hostname": "graylog-01"},
+            **kwargs,
+        )
+
+
+def _by_name(types):
+    return {t["name"]: t for t in types}
+
+
+class TestListing:
+    def test_offers_the_three_models_in_order(self):
+        mixin, _, _ = _mixin_on(CORE_I7_7700)
+        assert [t["name"] for t in mixin.get_vm_cpu_types()] == [
+            "x86-64-v2-AES",
+            "x86-64-v3",
+            "host",
+        ]
+
+    def test_exactly_one_default_and_it_is_what_the_gui_uses(self):
+        mixin, _, _ = _mixin_on(CORE_I7_7700)
+        defaults = [t["name"] for t in mixin.get_vm_cpu_types() if t["default"]]
+        assert defaults == ["x86-64-v2-AES"]
+
+    def test_v3_gives_avx_and_runs_on_a_core_i7(self):
+        mixin, _, _ = _mixin_on(CORE_I7_7700)
+        v3 = _by_name(mixin.get_vm_cpu_types())["x86-64-v3"]
+        assert v3["available"] is True
+        assert {"avx", "avx2"} <= set(v3["features"])
+
+    def test_v3_is_unavailable_on_a_celeron_without_avx(self):
+        mixin, _, _ = _mixin_on(CELERON_J3455)
+        types = _by_name(mixin.get_vm_cpu_types())
+        assert types["x86-64-v3"]["available"] is False
+        assert types["x86-64-v2-AES"]["available"] is True
+
+    def test_v2_aes_has_no_avx(self):
+        mixin, _, _ = _mixin_on(CORE_I7_7700)
+        assert "avx" not in _by_name(mixin.get_vm_cpu_types())["x86-64-v2-AES"]["features"]
+
+    def test_host_passes_the_nodes_own_flags_through(self):
+        """On a CPU without AVX, `host` gives none either -- a role that needs
+        AVX must not be told `host` would help there."""
+        mixin, _, _ = _mixin_on(CELERON_J3455)
+        host = _by_name(mixin.get_vm_cpu_types())["host"]
+        assert host["available"] is True
+        assert "avx" not in host["features"]
+        assert "aes" in host["features"]
+
+    def test_every_entry_explains_itself(self):
+        mixin, _, _ = _mixin_on(CORE_I7_7700)
+        assert all(t["description"] for t in mixin.get_vm_cpu_types())
+
+
+class TestCreate:
+    def test_names_the_default_model_instead_of_leaving_kvm64(self):
+        mixin, _, node = _mixin_on(CORE_I7_7700)
+        _create(mixin)
+        assert node.qemu.post.call_args[1]["cpu"] == "x86-64-v2-AES"
+
+    def test_passes_a_chosen_model_through(self):
+        mixin, _, node = _mixin_on(CORE_I7_7700)
+        _create(mixin, cpu_type="host")
+        assert node.qemu.post.call_args[1]["cpu"] == "host"
+
+    def test_refuses_a_model_the_node_cannot_run_before_creating_anything(self):
+        mixin, api, node = _mixin_on(CELERON_J3455)
+        with pytest.raises(ValueError, match="avx"):
+            _create(mixin, cpu_type="x86-64-v3")
+        api.cluster.nextid.get.assert_not_called()
+        node.qemu.post.assert_not_called()
+
+    def test_refuses_an_unknown_model_before_creating_anything(self):
+        mixin, api, node = _mixin_on(CORE_I7_7700)
+        with pytest.raises(ValueError, match="kvm64"):
+            _create(mixin, cpu_type="kvm64")
+        api.cluster.nextid.get.assert_not_called()
+        node.qemu.post.assert_not_called()

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"]

tests/test_vm_provision_mixin.py

@@ -459,7 +459,13 @@ def test_destroy_vm_success():
 
 
 def test_destroy_vm_already_stopped():
-    """destroy_vm succeeds even if VM already stopped."""
+    """destroy_vm succeeds even if VM already stopped.
+
+    Proxmox refuses to stop a VM that is not running, so the stop call raises.
+    (A bare MagicMock as the stop task never reports "stopped": _wait_for_task
+    then spun for the full timeout with sleep patched out, and every recorded
+    mock call made the test take 60 s and several GB, enough for CI to kill it.)
+    """
     mixin = ProxmoxVMProvisionMixin()
 
     mock_node = MagicMock()
@@ -468,6 +474,7 @@ def test_destroy_vm_already_stopped():
     # Mock VM operations
     mock_vm = MagicMock()
     mock_node.qemu.return_value = mock_vm
+    mock_vm.status.stop.post.side_effect = Exception("VM 101 not running")
     mock_vm.config.get.return_value = {}
     mock_vm.delete.return_value = None