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

@@ -36,8 +36,10 @@ logger = logging.getLogger(__name__)
 
 from napalm_device_types import (
     FingerprintRule,
+    HostStatusMixin,
     HypervisorDriver,
     KernelFactsMixin,
+    ListeningSocketsMixin,
     PortSpec,
     SystemdServicesMixin,
 )
@@ -83,7 +85,9 @@ class ProxmoxDriver(
     ProxmoxRoutingMixin,
     ProxmoxSystemMixin,
     KernelFactsMixin,
+    ListeningSocketsMixin,
     SystemdServicesMixin,
+    HostStatusMixin,
     HypervisorDriver,
 ):
     """NAPALM driver for Proxmox VE nodes."""
@@ -94,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__)
@@ -229,6 +231,15 @@ class ProxmoxSystemMixin:
         """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)
     # ------------------------------------------------------------------ #
@@ -362,22 +373,47 @@ class ProxmoxSystemMixin:
     # ------------------------------------------------------------------ #
 
     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,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
+                # 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
@@ -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._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}")
@@ -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:
+            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")
 
@@ -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
+                    # 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.2.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_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_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_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