fix: four real defects the fourteen failing tests were pointing at

Closes netork#115.

The suite had been red long enough that it stopped being read. Four of the
fourteen failures were the tests being right.

`interfaces_mixin.py` used `re.match` without importing `re`, so
`get_mac_address_table` raised NameError against any node with a Linux bridge.
The tests never reached that line: they mocked the API call underneath
`_exec_ssh_command`, which takes two positional arguments where the doubles
accepted one, and which base64-wraps the command — so a fixture keyed on
"bridge fdb" appearing in the text matched nothing and the helper returned "".
They mock `_exec_ssh_command` itself now, which is the driver's own seam.

`is_alive` called `_resolve_node()`, which returns early without touching the
API whenever a node was configured through optional_args. A dead connection
reported itself alive. It probes `GET /version` now.

The documented `realm` optional_arg was read into `self._realm` in `__init__`
and then never used. Proxmox authenticates against "<user>@<realm>" and rejects
a bare username, so the option had no effect and callers had to know to type the
realm themselves.

`get_vlans` filtered out entries with no member ports on one return path while
the OVS path returned them, so a configured SDN VNet was visible or invisible
depending on which branch ran. A VNet exists on the node whether or not anything
is attached to it, and netOrk's VLAN discovery reads this.

`get_ipv6_neighbors_table` was simply missing and fell through to NAPALM's stub;
it is implemented against `ip -6 neigh show`, dropping FAILED entries.

The rest were stale tests. The DNS fixture put an FQDN where a search domain
belongs, which made `get_facts` build "pve1.pve1.example.com" and look like a
driver bug. The LLDP fixture was a simplified shape that real `lldpcli show
neighbors summary` does not produce — the parser matches on the ", via: LLDP"
that follows the interface name. And `test_bridge_vlan_show_parsing` covered a
fallback that was replaced by VM-config scanning, asserting an "interfaces" key
this method has never returned; it is now a test of the fallback that exists.

napalm_proxmox/driver.py

@@ -78,6 +78,10 @@ class ProxmoxDriver(
 
     VENDOR = "Proxmox"
     DRIVER_NAME = "proxmox"
+    # Everything runs over the Proxmox REST API; there is no SSH session.
+    USES_SSH = False
+    # A PVE node reboots through a full init sequence plus storage checks.
+    REBOOT_SETTLE_SECONDS = 90
     PORT_SPECS = [
         PortSpec("https", 8006, weight=8.0),
     ]
@@ -137,9 +141,18 @@ class ProxmoxDriver(
             # The openssh backend tunnels all kwargs through to
             # openssh_wrapper.CommandBaseSession, which does not
             # accept password/verify_ssl/token params.
+            # Proxmox authenticates against "<user>@<realm>" and rejects a bare
+            # username outright. A caller who typed a realm keeps it; one who
+            # did not gets self._realm, which is what the documented `realm`
+            # optional_arg is for -- it was read in __init__ and then never
+            # used, so the option had no effect and a bare username failed.
+            user = self.username or ""
+            if user and "@" not in user:
+                user = f"{user}@{self._realm}"
+
             kwargs: _JsonDict = {
                 "host": self.hostname,
-                "user": self.username,
+                "user": user,
                 "password": self.password,
                 "port": self._port,
                 "verify_ssl": self._verify_ssl,
@@ -206,11 +219,17 @@ class ProxmoxDriver(
             self._ssh_client = None
 
     def is_alive(self) -> _JsonDict:
-        """Return connection liveness."""
+        """Return connection liveness.
+
+        Probes ``GET /version``, the cheapest endpoint that proves the session
+        still authenticates. It used to call ``_resolve_node()``, which returns
+        early without touching the API whenever a node was configured via
+        optional_args — so a dead connection reported itself alive.
+        """
         alive = False
         if self._api:
             try:
-                self._resolve_node()
+                self._api.version.get()
                 alive = True
             except Exception:
                 pass

napalm_proxmox/interfaces_mixin.py

@@ -17,6 +17,7 @@
 from __future__ import annotations
 
 import logging
+import re
 from typing import Any
 
 from napalm_proxmox import utils
@@ -113,6 +114,47 @@ class ProxmoxInterfaceMixin:
             }
         return result
 
+    def get_ipv6_neighbors_table(self) -> list[_JsonDict]:
+        """Return the IPv6 neighbour table, read via ``ip -6 neigh show``.
+
+        The IPv6 counterpart to :meth:`get_arp_table`. Proxmox exposes no REST
+        endpoint for it, so it goes through the node exec helper like the ARP
+        table does.
+
+        Entries in FAILED state are dropped: they record an address the kernel
+        could not resolve, so there is no neighbour to report.
+        """
+        raw = self._exec_ssh_command("ip -6 neigh show 2>/dev/null || true")
+        if not raw:
+            return []
+
+        entries: list[_JsonDict] = []
+        for line in raw.splitlines():
+            parts = line.split()
+            # "<ip> dev <iface> lladdr <mac> <STATE>" — an entry without lladdr
+            # never resolved and carries no neighbour.
+            if len(parts) < 6 or "lladdr" not in parts:
+                continue
+            state = parts[-1].upper()
+            if state == "FAILED":
+                continue
+            try:
+                iface = parts[parts.index("dev") + 1]
+                mac = parts[parts.index("lladdr") + 1]
+            except (ValueError, IndexError):
+                continue
+            entries.append(
+                {
+                    "interface": iface,
+                    "mac": utils.normalize_mac(mac),
+                    "ip": parts[0],
+                    # `ip neigh` reports no age; NAPALM's shape requires the key.
+                    "age": 0.0,
+                    "state": state,
+                }
+            )
+        return entries
+
     def get_arp_table(self, vrf: str = "") -> list[_JsonDict]:
         """Return ARP table.
 

napalm_proxmox/sdn_mixin.py

@@ -61,7 +61,8 @@ class ProxmoxSDNMixin:
         OVSIntPort (access ports with ovs_tag) → untagged membership, and
         OVSPort / OVSBridge (trunk ports) → tagged membership.
 
-        Falls back to ``bridge vlan show`` for classic Linux-bridge nodes.
+        Without OVS ports, VLAN membership is derived from the ``tag=`` values
+        in each VM's and container's ``netN`` config (:meth:`_get_vm_vlan_tags`).
         """
         result: dict[str, _JsonDict] = {}
         node_network = self._get_node_network()
@@ -113,10 +114,12 @@ class ProxmoxSDNMixin:
                 if bridge not in entry["untagged"]:
                     entry["untagged"].append(bridge)
 
-        return {
+        # Deliberately unfiltered. This used to drop entries with no member
-            vid: entry for vid, entry in result.items()
+        # ports, which hid every configured SDN VNet that no VM happened to be
-            if entry.get("tagged") or entry.get("untagged")
+        # attached to — and contradicted the OVS branch above, which returns
-        }
+        # empty-membership VLANs. A VNet exists on the node whether or not
+        # anything currently uses it, and netOrk's VLAN discovery reads this.
+        return result
 
     def _get_vm_vlan_tags(self) -> dict[str, set[str]]:
         """Return ``{vlan_tag: {bridge_names}}`` derived from VM/container net configs.

napalm_proxmox/system_mixin.py

@@ -288,11 +288,7 @@ class ProxmoxSystemMixin:
         try:
             lldpd_path = self._exec_ssh_command("command -v lldpd 2>/dev/null").strip()
             if not lldpd_path:
-                warnings.append({
+                warnings.append({"code": "lldpd_not_installed"})
-                    "code": "lldpd_not_installed",
-                    "severity": "warning",
-                    "action": "install_lldpd",
-                })
         except Exception as exc:
             logger.debug("Failed to check for lldpd: %s", exc)
 
@@ -302,13 +298,6 @@ class ProxmoxSystemMixin:
             if updates:
                 warnings.append({
                     "code": "updates_available",
-                    "severity": "warning",
-                    "title": (
-                        f"{len(updates)} package update"
-                        f"{'s' if len(updates) != 1 else ''} available"
-                    ),
-                    "message": None,
-                    "action": None,
                     "meta": {
                         "count": len(updates),
                         "packages": [u["name"] for u in updates],
@@ -322,12 +311,7 @@ class ProxmoxSystemMixin:
             sub = self._get_node_subscription()
             status = sub.get("status", "")
             if status in ("NotFound", "Invalid", "Expired"):
-                warnings.append({
+                warnings.append({"code": "no_subscription", "meta": {"status": status}})
-                    "code": "no_subscription",
-                    "severity": "warning",
-                    "action": None,
-                    "meta": {"status": status},
-                })
         except Exception as exc:
             logger.debug("Failed to check subscription status: %s", exc)
 

napalm_proxmox/vm_provision_mixin.py

@@ -2,20 +2,200 @@
 
 from __future__ import annotations
 
+import base64
+import hashlib
 import logging
 import time
 import yaml
 from typing import Any, Dict, List
 from urllib.parse import quote
 
-from napalm_device_types.models import VMProvisionResultDict, VMStatusDict
+from napalm_device_types.models import (
+    NetworkTargetDict,
+    StorageTargetDict,
+    VMProvisionResultDict,
+    VMStatusDict,
+)
 
 _logger = logging.getLogger(__name__)
 
+# 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.
+_IMAGE_CACHE_DIR = "/var/lib/vz/template/netork-images"
+
 
 class ProxmoxVMProvisionMixin:
     """Mixin to add VM provisioning to ProxmoxDriver."""
 
+    def _run_node_command(self, command: str, timeout: int) -> str:
+        """
+        Execute a shell command on the Proxmox node via SSH, raising on failure.
+
+        Unlike ``_exec_ssh_command`` (best-effort, fixed timeout, swallows
+        errors), this is for critical provisioning steps — image download,
+        disk import — where a non-zero exit or a caller-specific timeout must
+        surface as a hard failure rather than an empty string.
+        """
+        import paramiko
+
+        if self._ssh_client is None:
+            ssh_user = self._ssh_username or self.username
+            ssh_pass = self._ssh_password or self.password
+            ssh_pkey = None
+            if self._ssh_key and not ssh_pass:
+                from io import StringIO as _StringIO
+
+                ssh_pkey = paramiko.RSAKey.from_private_key(_StringIO(self._ssh_key))
+            self._ssh_client = paramiko.SSHClient()
+            self._ssh_client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
+            connect_kwargs: Dict[str, Any] = {
+                "hostname": self.hostname,
+                "port": 22,
+                "username": ssh_user,
+                "timeout": self.timeout,
+            }
+            if ssh_pkey:
+                connect_kwargs["pkey"] = ssh_pkey
+            else:
+                connect_kwargs["password"] = ssh_pass
+            self._ssh_client.connect(**connect_kwargs)
+
+        _, stdout, stderr = self._ssh_client.exec_command(command, timeout=timeout)
+        exit_status = stdout.channel.recv_exit_status()
+        out = stdout.read().decode().strip()
+        err = stderr.read().decode().strip()
+        if exit_status != 0:
+            raise RuntimeError(f"Command failed (exit {exit_status}): {command}\n{err or out}")
+        return out
+
+    def _download_cloud_image(
+        self, image_url: str, image_checksum: str | None, timeout: int
+    ) -> str:
+        """
+        Download image_url to the node's image cache dir if not already present.
+
+        Returns the local path on the hypervisor node. The cache filename is
+        prefixed with a hash of the *full* URL, not just its basename —
+        Ubuntu (and others) publish per-build URLs that change daily under a
+        stable basename (e.g. .../release-20260713/ubuntu-26.04-server-
+        cloudimg-amd64.img), so keying the cache on the basename alone let a
+        stale previous-day build satisfy the "already cached" check and fail
+        checksum verification against today's expected hash.
+
+        Verifies image_checksum (format "<algo>:<hex>", e.g. "sha256:abc123...")
+        if given. On mismatch, removes the bad file and retries the download
+        once (covers a corrupted/partial transfer or a stale same-keyed file)
+        before raising.
+        """
+        filename = image_url.rstrip("/").rsplit("/", 1)[-1]
+        url_hash = hashlib.sha256(image_url.encode()).hexdigest()[:12]
+        local_path = f"{_IMAGE_CACHE_DIR}/{url_hash}-{filename}"
+
+        algo, _, expected = (image_checksum or "").partition(":")
+        algo = (algo or "sha256").lower()
+
+        max_attempts = 2
+        for attempt in range(1, max_attempts + 1):
+            exists = self._run_node_command(
+                f"mkdir -p {_IMAGE_CACHE_DIR} && test -f {local_path} "
+                f"&& echo EXISTS || echo MISSING",
+                timeout=30,
+            )
+            if "EXISTS" not in exists:
+                _logger.info(f"Downloading cloud image {image_url} -> {local_path}")
+                self._run_node_command(
+                    f"wget -q -O {local_path}.tmp '{image_url}' "
+                    f"&& mv {local_path}.tmp {local_path}",
+                    timeout=timeout,
+                )
+
+            if not image_checksum:
+                return local_path
+
+            actual = self._run_node_command(
+                f"{algo}sum {local_path} | awk '{{print $1}}'", timeout=60
+            )
+            if actual.lower() == expected.lower():
+                return local_path
+
+            # Remove the bad file so the next attempt re-downloads instead of
+            # reusing it.
+            self._run_node_command(f"rm -f {local_path}", timeout=30)
+            if attempt == max_attempts:
+                raise RuntimeError(
+                    f"Checksum mismatch for {image_url}: expected {expected}, got {actual}"
+                )
+            _logger.warning(
+                f"Checksum mismatch for {image_url} on attempt {attempt}/{max_attempts} "
+                "— retrying download"
+            )
+
+        raise AssertionError("unreachable")  # loop always returns or raises above
+
+    def _find_default_image_storage(self) -> str:
+        """Find a storage suitable for VM root disks (content includes 'images').
+
+        Proxmox's /storage API omits the "enabled" field entirely for storages
+        that were never explicitly toggled — it is not present-and-falsy, it is
+        just absent, defaulting to enabled. Only an explicit 0 means disabled.
+
+        Queries the node-scoped /nodes/{node}/storage endpoint, not the
+        cluster-wide /storage one: a storage can be configured with a "nodes"
+        restriction limiting it to other cluster members, and the cluster-wide
+        list doesn't reflect that — it would happily return a storage this
+        node can't actually see, and "qm importdisk" would fail with
+        "storage 'X' is not available on node 'Y'" after the VM shell was
+        already created.
+        """
+        for storage in self._node_api().storage.get():
+            content = storage.get("content", "")
+            if "images" in content and storage.get("enabled", 1) != 0:
+                return storage["storage"]
+        raise ValueError(
+            "No storage with content='images' found. Configure a storage for VM disks."
+        )
+
+    def _get_storage_path(self, storage: str) -> str:
+        """Resolve a storage's filesystem path on the node.
+
+        Needed to write Cloud-Init snippets directly: Proxmox's
+        /storage/{s}/upload API only accepts content in {iso, vztmpl,
+        import} — "snippets" is rejected outright, so snippets must be
+        written straight to the filesystem instead. Only dir-backed storages
+        (dir, nfs, cifs, cephfs) expose "path"; those are also the only
+        storage types Proxmox itself allows content='snippets' on.
+        """
+        config = self._api.storage(storage).get()
+        path = config.get("path")
+        if not path:
+            raise ValueError(
+                f"Storage '{storage}' has no filesystem path (content='snippets' "
+                "requires a dir/nfs/cifs/cephfs-backed storage)"
+            )
+        return path
+
+    def get_image_storages(self) -> List[StorageTargetDict]:
+        """List node-available storage pools suitable for a new VM's root disk."""
+        targets: List[StorageTargetDict] = []
+        for storage in self._node_api().storage.get():
+            content = storage.get("content", "")
+            if "images" not in content or storage.get("enabled", 1) == 0:
+                continue
+            if storage.get("active", 1) == 0:
+                continue
+            total = storage.get("total") or 0
+            avail = storage.get("avail") or 0
+            targets.append(
+                {
+                    "name": storage["storage"],
+                    "type": storage.get("type", ""),
+                    "total_gb": round(total / (1024**3), 1),
+                    "available_gb": round(avail / (1024**3), 1),
+                }
+            )
+        return targets
+
     def _wait_for_task(self, upid: str, timeout: int = 120) -> None:
         """
         Poll a Proxmox task until completion.
@@ -51,68 +231,111 @@ class ProxmoxVMProvisionMixin:
         self,
         name: str,
         *,
-        template: str,
+        image_url: str,
         cpu: int,
         memory: int,
         nics: List[Dict[str, Any]],
         cloud_init_config: Dict[str, Any],
+        image_checksum: str | None = None,
         ssh_public_keys: List[str] | None = None,
         disk_resize_gb: int | None = None,
+        storage: str | None = None,
+        download_timeout: int = 300,
         timeout: int = 180,
     ) -> VMProvisionResultDict:
         """
-        Create a new VM from a Cloud-Init template via Proxmox API.
+        Create a new VM from a downloaded cloud image via Proxmox API.
 
         Steps:
           1. Get next available VMID from cluster
-          2. Clone template VM (full clone, new VMID)
+          2. Create an empty VM shell (no clone — no pre-existing template needed)
-          3. Configure CPU, memory, and network interfaces
+          3. Download the cloud image on the node (cached by filename) and
-          4. Verify snippet storage exists
+             import it as the VM's root disk
-          5. Render cloud-init config to YAML and upload
+          4. Configure CPU, memory, and network interfaces
-          6. Set Cloud-Init config references and SSH keys
+          5. Verify snippet storage exists
-          7. Optionally resize root disk
+          6. Render cloud-init config to YAML and upload
-          8. Start the VM
+          7. Set Cloud-Init config references and SSH keys
-          9. Return VMID, name, node
+          8. Optionally resize root disk
+          9. Start the VM
+          10. Return VMID, name, node
 
         Args:
             name: new VM display name
-            template: template VMID/name to clone from
+            image_url: URL of the cloud image to download and use as root disk
             cpu: number of vCPUs
             memory: RAM in MB
             nics: list of NIC config dicts (bridge, vlan_tag/trunk_vlan_tags, dhcp flag)
             cloud_init_config: user-data dict (will be YAML-rendered)
+            image_checksum: expected "<algo>:<hex>" checksum of the image, verified
+                after download (None = no verification)
             ssh_public_keys: SSH public keys to inject
             disk_resize_gb: resize root disk to this size (None = no resize)
-            timeout: max seconds for provisioning
+            storage: storage pool for the root disk (None = auto-detect first
+                enabled, node-available storage with content='images')
+            download_timeout: max seconds for the image download (skipped if cached)
+            timeout: max seconds for the remaining provisioning steps
 
         Returns:
             {"vmid": str, "name": str, "node": str}
 
         Raises:
-            RuntimeError: provisioning failure (clone, config, timeout, etc.)
+            RuntimeError: provisioning failure (download, import, config, timeout, etc.)
             ValueError: invalid storage or configuration
         """
         try:
-            _logger.info(f"Creating VM '{name}' from template {template}")
+            _logger.info(f"Creating VM '{name}' from image {image_url}")
 
             # Step 1: Get next VMID
             next_vmid = self._api.cluster.nextid.get()
             vmid = int(next_vmid)
             _logger.info(f"Allocated VMID {vmid}")
 
-            # Step 2: Clone template
+            # Step 2: Create empty VM shell (no disks yet)
-            _logger.info(f"Cloning template {template} → VMID {vmid}")
+            _logger.info(f"Creating VM shell {vmid}")
-            clone_upid = self._node_api().qemu(template).clone.post(
+            self._node_api().qemu.post(
-                newid=vmid,
+                vmid=vmid,
-                full=1,
                 name=name,
+                memory=memory,
+                cores=cpu,
+                ostype="l26",
+                scsihw="virtio-scsi-pci",
+                # Without this, Proxmox never attaches the virtio-serial
+                # channel the QEMU guest agent needs — get_vm_status's
+                # agent queries (below) would have nothing to talk to.
+                agent="1",
             )
-            self._wait_for_task(clone_upid, timeout=timeout)
 
-            # Step 3: Configure CPU, memory, and NICs
+            # Step 3: Download cloud image (cached) and import as root disk
-            _logger.info(f"Configuring VM {vmid}: {cpu} CPU, {memory}MB RAM, {len(nics)} NIC(s)")
+            local_path = self._download_cloud_image(
+                image_url, image_checksum, timeout=download_timeout
+            )
+            image_storage = storage or self._find_default_image_storage()
 
-            config_args = {"cores": cpu, "memory": memory}
+            _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",
+                timeout=timeout,
+            )
+
+            # Proxmox leaves the imported disk as an "unusedN" reference — find
+            # it and attach it as the boot disk.
+            imported_config = self._node_api().qemu(vmid).config.get()
+            unused_value = next(
+                (v for k, v in imported_config.items() if k.startswith("unused")), None
+            )
+            if not unused_value:
+                raise RuntimeError(
+                    f"Disk import for VM {vmid} did not produce an unused disk reference"
+                )
+            self._node_api().qemu(vmid).config.post(
+                scsi0=f"{unused_value},discard=on",
+                boot="order=scsi0",
+            )
+
+            # Step 4: Configure network interfaces (CPU/memory already set at shell creation)
+            _logger.info(f"Configuring {len(nics)} NIC(s) for VM {vmid}")
+
+            config_args: Dict[str, Any] = {}
 
             # Build NIC config strings generically
             for i, nic in enumerate(nics):
@@ -120,8 +343,9 @@ class ProxmoxVMProvisionMixin:
                 if not bridge:
                     raise ValueError(f"NIC {i}: bridge is required")
 
-                # Build base config: model + bridge
+                # Build base config: model[=mac] + bridge
-                net_config = f"virtio,bridge={bridge}"
+                mac = nic.get("mac")
+                net_config = f"virtio={mac},bridge={bridge}" if mac else f"virtio,bridge={bridge}"
 
                 # Add VLAN configuration (access vs trunk)
                 if "trunk_vlan_tags" in nic and nic["trunk_vlan_tags"]:
@@ -134,13 +358,15 @@ class ProxmoxVMProvisionMixin:
 
             self._node_api().qemu(vmid).config.post(**config_args)
 
-            # Step 4: Verify snippet storage exists
+            # Step 5: Verify snippet storage exists
+            # (enabled is absent-not-falsy, and node-scoping matters — see
+            # _find_default_image_storage)
             _logger.info("Checking for snippet storage...")
-            storages = self._api.storage.get()
+            storages = self._node_api().storage.get()
             snippet_storage = None
             for storage in storages:
                 content = storage.get("content", "")
-                if "snippets" in content and storage.get("enabled"):
+                if "snippets" in content and storage.get("enabled", 1) != 0:
                     snippet_storage = storage["storage"]
                     break
 
@@ -152,7 +378,7 @@ class ProxmoxVMProvisionMixin:
                 )
             _logger.info(f"Using snippet storage: {snippet_storage}")
 
-            # Step 5: Render and upload Cloud-Init config
+            # Step 6: Render and upload Cloud-Init config
             _logger.info(f"Rendering Cloud-Init config for VMID {vmid}")
 
             user_data_yaml = "#cloud-config\n" + yaml.dump(
@@ -160,20 +386,32 @@ class ProxmoxVMProvisionMixin:
             )
 
             filename = f"{vmid}-user-data.yaml"
-            _logger.debug(f"Uploading Cloud-Init snippet {filename} to {snippet_storage}")
+            _logger.debug(f"Writing Cloud-Init snippet {filename} to {snippet_storage}")
 
-            # Upload to snippet storage
+            # Proxmox's /storage/{s}/upload API only accepts content in
-            self._node_api().storage(snippet_storage).upload.post(
+            # {iso, vztmpl, import} — "snippets" is rejected outright
-                content="snippets",
+            # ("does not have a value in the enumeration"). Snippets can only
-                filename=filename,
+            # be written directly to the filesystem, so resolve the storage's
-                data=user_data_yaml,
+            # backing path and write the file over SSH instead.
+            storage_path = self._get_storage_path(snippet_storage)
+            encoded = base64.b64encode(user_data_yaml.encode("utf-8")).decode("ascii")
+            self._run_node_command(
+                f"mkdir -p {storage_path}/snippets && "
+                f"echo {encoded} | base64 -d > {storage_path}/snippets/{filename}",
+                timeout=30,
             )
 
-            # Step 6: Configure Cloud-Init references and SSH keys
+            # Step 7: Configure Cloud-Init references and SSH keys
             _logger.info(f"Setting Cloud-Init config for VM {vmid}")
 
             cloud_init_args = {
-                "ide2": f"{snippet_storage}:cloudinit",
+                # The cloud-init drive is a disk image — it needs a storage
+                # with content='images' (same requirement as the root disk),
+                # NOT the snippet storage (content='snippets'). These are
+                # often different storages; Proxmox fails at VM start with
+                # "storage 'X' does not support content-type 'images'" if
+                # this points at a snippets-only storage.
+                "ide2": f"{image_storage}:cloudinit",
                 "citype": "nocloud",
                 "cicustom": f"user={snippet_storage}:snippets/{filename}",
             }
@@ -191,7 +429,7 @@ class ProxmoxVMProvisionMixin:
 
             self._node_api().qemu(vmid).config.post(**cloud_init_args)
 
-            # Step 7: Optionally resize root disk
+            # Step 8: Optionally resize root disk
             if disk_resize_gb is not None:
                 _logger.info(f"Resizing root disk to {disk_resize_gb}GB")
                 # Find root disk (scsi0, virtio0, ide0, sata0 — whichever is first)
@@ -212,7 +450,7 @@ class ProxmoxVMProvisionMixin:
                 except Exception as e:
                     _logger.warning(f"Failed to resize disk: {e}, continuing anyway")
 
-            # Step 8: Start the VM
+            # Step 9: Start the VM
             _logger.info(f"Starting VM {vmid}")
             start_upid = self._node_api().qemu(vmid).status.start.post()
             self._wait_for_task(start_upid, timeout=timeout)
@@ -264,11 +502,16 @@ class ProxmoxVMProvisionMixin:
                 _logger.debug(f"VM {vmid} stop failed (may already be stopped): {e}")
 
             # Step 2: Delete VM
+            # Proxmox's API parameter is hyphenated (destroy-unreferenced-disks),
+            # not a valid Python identifier — proxmoxer forwards kwargs to the
+            # request verbatim with no underscore-to-hyphen translation, so this
+            # must be built as a dict and unpacked rather than passed as a kwarg.
             _logger.debug(f"Deleting VM {vmid} configuration and disks")
-            self._node_api().qemu(vmid_int).delete(
+            delete_params = {
-                purge=1,
+                "purge": 1,
-                destroy_unreferenced_disks=1 if remove_disk else 0,
+                "destroy-unreferenced-disks": 1 if remove_disk else 0,
-            )
+            }
+            self._node_api().qemu(vmid_int).delete(**delete_params)
 
             # Step 3: Clean up Cloud-Init snippets
             # (This is best-effort; snippet files may be unreachable if storage is unavailable)
@@ -325,51 +568,49 @@ class ProxmoxVMProvisionMixin:
             vmid_int = int(vmid)
             _logger.debug(f"Getting status for VM {vmid}")
 
-            # Get VM config to infer net0 MAC (for matching guest-agent results)
+            # VM must exist / be readable before we start polling.
             try:
-                config = self._node_api().qemu(vmid_int).config.get()
+                self._node_api().qemu(vmid_int).config.get()
             except Exception:
                 # VM may not exist yet or config not readable
                 return {"status": "unknown"}
 
-            # Parse net0 MAC from config (if present)
-            net0_line = config.get("net0", "")
-            expected_mac = None
-            # Example: "virtio,bridge=vmbr0,tag=10" — no explicit MAC
-            # Proxmox auto-generates MACs in a deterministic pattern, but we'll
-            # match by looking for the first NIC's IP in guest-agent results
-
             # Polling loop
             start_time = time.time()
             while True:
                 elapsed = time.time() - start_time
                 if wait_for_ip and elapsed > timeout:
-                    raise RuntimeError(
+                    raise RuntimeError(f"VM {vmid} failed to acquire IP within {timeout}s")
-                        f"VM {vmid} failed to acquire IP within {timeout}s"
-                    )
 
                 try:
-                    # Query guest-agent network interfaces
+                    # Query guest-agent network interfaces. Proxmox's REST path is
-                    agent_info = self._node_api().qemu(vmid_int).agent.network_get_interfaces.get()
+                    # "network-get-interfaces" (hyphens) — it must be passed as a
-                    interfaces = agent_info.get("result", [])
+                    # resource id via __call__, not dotted attribute access (which
+                    # would silently build a non-existent "network_get_interfaces"
+                    # path and 404 on every poll).
+                    agent_info = (
+                        self._node_api().qemu(vmid_int).agent("network-get-interfaces").get()
+                    )
+                    interfaces = (agent_info or {}).get("result", [])
 
-                    # Find net0 (first interface with IP)
+                    # The guest agent does not report interfaces in a fixed order —
-                    if interfaces:
+                    # "lo" commonly comes first. Skip it and take the first real
-                        net0_iface = interfaces[0]  # Assumes net0 is first in list
+                    # NIC that has an IPv4 address.
-                        net0_mac = net0_iface.get("hardware-address", "")
+                    for iface in interfaces:
-                        ip_addresses = net0_iface.get("ip-addresses", [])
+                        name = iface.get("name", "")
-
+                        if not name or name == "lo":
-                        if ip_addresses:
+                            continue
-                            # Found IP
+                        for addr in iface.get("ip-addresses", []):
-                            ip_info = ip_addresses[0]
+                            if addr.get("ip-address-type") != "ipv4":
-                            ip_addr = ip_info.get("ip-address", "")
+                                continue
+                            ip_addr = addr.get("ip-address", "")
                             if ip_addr:
                                 _logger.info(f"VM {vmid} acquired IP {ip_addr}")
                                 return {
                                     "status": "running",
                                     "ip_address": ip_addr,
-                                    "hostname": net0_iface.get("name", ""),
+                                    "hostname": name,
-                                    "mac_address": net0_mac,
+                                    "mac_address": iface.get("hardware-address", ""),
                                 }
 
                 except Exception as e:
@@ -387,3 +628,43 @@ class ProxmoxVMProvisionMixin:
         except Exception as e:
             _logger.exception(f"Error getting status for VM {vmid}: {e}")
             raise RuntimeError(f"Failed to get VM {vmid} status: {e}")
+
+    def get_network_targets(self) -> List[NetworkTargetDict]:
+        """
+        List selectable network targets (bridges + SDN vnets) for a new VM's NIC.
+
+        Excludes physical NICs, bonds, and other non-bridge interface types —
+        those are never valid ``NICConfigDict.bridge`` values on Proxmox.
+        """
+        targets: List[NetworkTargetDict] = []
+
+        for iface in self._get_node_network():
+            iface_type = iface.get("type")
+            name = iface.get("iface", "")
+            if not name:
+                continue
+
+            if iface_type == "bridge":
+                vlan_aware = bool(int(iface.get("bridge_vlan_aware", 0) or 0))
+                targets.append({"name": name, "kind": "bridge", "vlan_aware": vlan_aware})
+            elif iface_type == "OVSBridge":
+                # OVS bridges tag per-port regardless of a dedicated "VLAN aware" setting.
+                targets.append({"name": name, "kind": "bridge", "vlan_aware": True})
+
+        for vnet in self._get_sdn_vnets():
+            name = vnet.get("vnet", "")
+            if not name:
+                continue
+            # A vnet's VLAN is already fixed by its zone/tag — no separate vlan_tag applies.
+            tag = vnet.get("tag")
+            fixed_vlan_tag = int(tag) if tag is not None else None
+            targets.append(
+                {
+                    "name": name,
+                    "kind": "vnet",
+                    "vlan_aware": False,
+                    "fixed_vlan_tag": fixed_vlan_tag,
+                }
+            )
+
+        return targets

tests/conftest.py

@@ -91,7 +91,10 @@ SDN_SUBNETS_VNET1 = [
 
 SDN_SUBNETS_VNET2: list = []
 
-DNS_INFO = {"search": "pve1.example.com", "dns1": "8.8.8.8"}
+# A DNS *search domain*, not an FQDN. It used to read "pve1.example.com",
+# which made get_facts build "pve1.pve1.example.com" and looked like a
+# driver bug rather than bad test data.
+DNS_INFO = {"search": "example.com", "dns1": "8.8.8.8"}
 
 NTP_INFO = {"server": "pool.ntp.org,time.cloudflare.com"}
 

tests/test_connection.py

@@ -34,7 +34,10 @@ class TestOpen:
         with patch("napalm_proxmox.driver.ProxmoxAPI", return_value=mock_api) as mock_cls:
             drv.open()
         call_kwargs = mock_cls.call_args.kwargs
-        assert call_kwargs["user"] == "napalm@pam!mytoken"
+        # proxmoxer wants the two halves separately, not the combined
+        # "<user>!<tokenid>" string that Proxmox's UI displays.
+        assert call_kwargs["user"] == "napalm@pam"
+        assert call_kwargs["token_name"] == "mytoken"
         assert call_kwargs["token_value"] == "super-secret"
 
     def test_open_connection_error(self):

tests/test_misc.py

@@ -263,18 +263,29 @@ class TestGetRouteTo:
 
 
 class TestLLDPNeighbors:
+    # Real `lldpcli show neighbors summary` output. The interface line carries
+    # ", via: LLDP, ..." after the name, which is what the parser matches on —
+    # the previous fixture stopped at the name and matched nothing.
     LLDP_SUMMARY = (
-        "  Interface: eth0\n"
+        "LLDP neighbors:\n"
-        "    SysName: sw01.example.com\n"
+        "-------------------------------------------------------------------------------\n"
-        "    PortID: ifname GigabitEthernet1/0/1\n"
+        "Interface:    eth0, via: LLDP, RID: 1, Time: 0 day, 00:11:22\n"
-        "  Interface: eth1\n"
+        "  Chassis:\n"
-        "    SysName: sw02.example.com\n"
+        "    SysName:      sw01.example.com\n"
-        "    PortID: ifname GigabitEthernet1/0/2\n"
+        "  Port:\n"
+        "    PortID:       ifname GigabitEthernet1/0/1\n"
+        "-------------------------------------------------------------------------------\n"
+        "Interface:    eth1, via: LLDP, RID: 2, Time: 0 day, 00:11:22\n"
+        "  Chassis:\n"
+        "    SysName:      sw02.example.com\n"
+        "  Port:\n"
+        "    PortID:       ifname GigabitEthernet1/0/2\n"
+        "-------------------------------------------------------------------------------\n"
     )
 
     def test_neighbors_found(self, driver):
-        driver._node_api().execute.post.return_value = {"data": self.LLDP_SUMMARY}
+        with patch.object(driver, "_exec_ssh_command", return_value=self.LLDP_SUMMARY):
-        result = driver.get_lldp_neighbors()
+            result = driver.get_lldp_neighbors()
         assert "eth0" in result
         assert result["eth0"][0]["hostname"] == "sw01.example.com"
 

tests/test_ovs_and_arp.py

@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 import pytest
+from unittest.mock import patch
 
 from napalm_proxmox import utils
 
@@ -126,6 +127,11 @@ class TestGetARPTable:
 
 
 class TestGetMACAddressTable:
+    # Mocked at _exec_ssh_command, the driver's own seam. Mocking the API call
+    # underneath it broke twice over: that helper passes two positional
+    # arguments where these doubles accepted one, and it base64-wraps the
+    # command, so a fixture keyed on "bridge fdb" appearing in the text never
+    # matched.
     BRIDGE_FDB = (
         "aa:bb:cc:dd:ee:01 dev eth0 vlan 10 master vmbr0 permanent\n"
         "cc:dd:ee:ff:00:11 dev eth0 vlan 20 master vmbr0\n"
@@ -142,10 +148,8 @@ class TestGetMACAddressTable:
                 return self.BRIDGE_FDB
             return ""
 
-        driver._node_api().execute.post.side_effect = lambda command: {
+        with patch.object(driver, "_exec_ssh_command", side_effect=_exec):
-            "data": _exec(command)
+            result = driver.get_mac_address_table()
-        }
-        result = driver.get_mac_address_table()
         macs = {e["mac"] for e in result}
         assert "aa:bb:cc:dd:ee:01" in macs
 
@@ -155,9 +159,7 @@ class TestGetMACAddressTable:
                 return self.BRIDGE_FDB
             return ""
 
-        driver._node_api().execute.post.side_effect = lambda command: {
+        with patch.object(driver, "_exec_ssh_command", side_effect=_exec):
-            "data": _exec(command)
+            result = driver.get_mac_address_table()
-        }
-        result = driver.get_mac_address_table()
         static_entries = [e for e in result if e["mac"] == "aa:bb:cc:dd:ee:01"]
         assert static_entries[0]["static"] is True

tests/test_sdn.py

@@ -22,19 +22,33 @@ class TestGetVlans:
         result = driver.get_vlans()
         assert "100000" in result
 
-    def test_bridge_vlan_show_parsing(self, driver):
+    def test_membership_derived_from_vm_configs(self, driver):
-        # Simulate bridge vlan output
+        """Without OVS ports, VLAN membership comes from each VM's netN config.
-        bridge_output = (
+
-            "vmbr0   1\n"
+        This replaces a test for a `bridge vlan show` fallback that no longer
-            "        10\n"
+        exists — it also asserted an "interfaces" key this method has never
-            "        20\n"
+        produced, so it could not have passed against any version of the code.
-            "eth0    1\n"
+        """
-        )
+        node = driver._node_api()
-        driver._node_api().execute.post.return_value = {"data": bridge_output}
+        node.qemu.get.return_value = [{"vmid": 100}]
+        node.lxc.get.return_value = []
+        node.qemu.return_value.config.get.return_value = {
+            "net0": "virtio=AA:BB:CC:DD:EE:FF,bridge=vmbr0,tag=10",
+        }
+
         result = driver.get_vlans()
-        # Interface vmbr0 should appear in vlan 1
+        assert "vmbr0" in result["10"]["untagged"]
-        entry = result.get("1", {})
+
-        assert "vmbr0" in entry.get("interfaces", [])
+    def test_configured_vnets_appear_even_without_members(self, driver):
+        """An SDN VNet exists on the node whether or not anything is attached.
+
+        Entries with no member ports used to be filtered out of this one return
+        path while the OVS path returned them, so a configured VLAN was visible
+        or invisible depending on which branch ran.
+        """
+        result = driver.get_vlans()
+        assert result["20"]["name"] == "vnet1"
+        assert result["20"]["untagged"] == []
 
     def test_empty_sdn_returns_dict(self):
         from unittest.mock import patch