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.
This commit is contained in:
Christian Manivong
2026-08-21 13:22:03 +07:00
parent 51f67704e1
commit 20fcf2ebc3
8 changed files with 131 additions and 38 deletions
+18 -3
View File
@@ -141,9 +141,18 @@ class ProxmoxDriver(
# The openssh backend tunnels all kwargs through to # The openssh backend tunnels all kwargs through to
# openssh_wrapper.CommandBaseSession, which does not # openssh_wrapper.CommandBaseSession, which does not
# accept password/verify_ssl/token params. # 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 = { kwargs: _JsonDict = {
"host": self.hostname, "host": self.hostname,
"user": self.username, "user": user,
"password": self.password, "password": self.password,
"port": self._port, "port": self._port,
"verify_ssl": self._verify_ssl, "verify_ssl": self._verify_ssl,
@@ -210,11 +219,17 @@ class ProxmoxDriver(
self._ssh_client = None self._ssh_client = None
def is_alive(self) -> _JsonDict: 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 alive = False
if self._api: if self._api:
try: try:
self._resolve_node() self._api.version.get()
alive = True alive = True
except Exception: except Exception:
pass pass
+42
View File
@@ -17,6 +17,7 @@
from __future__ import annotations from __future__ import annotations
import logging import logging
import re
from typing import Any from typing import Any
from napalm_proxmox import utils from napalm_proxmox import utils
@@ -113,6 +114,47 @@ class ProxmoxInterfaceMixin:
} }
return result 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]: def get_arp_table(self, vrf: str = "") -> list[_JsonDict]:
"""Return ARP table. """Return ARP table.
+8 -5
View File
@@ -61,7 +61,8 @@ class ProxmoxSDNMixin:
OVSIntPort (access ports with ovs_tag) → untagged membership, and OVSIntPort (access ports with ovs_tag) → untagged membership, and
OVSPort / OVSBridge (trunk ports) → tagged membership. 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] = {} result: dict[str, _JsonDict] = {}
node_network = self._get_node_network() node_network = self._get_node_network()
@@ -113,10 +114,12 @@ class ProxmoxSDNMixin:
if bridge not in entry["untagged"]: if bridge not in entry["untagged"]:
entry["untagged"].append(bridge) 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]]: def _get_vm_vlan_tags(self) -> dict[str, set[str]]:
"""Return ``{vlan_tag: {bridge_names}}`` derived from VM/container net configs. """Return ``{vlan_tag: {bridge_names}}`` derived from VM/container net configs.
+4 -1
View File
@@ -91,7 +91,10 @@ SDN_SUBNETS_VNET1 = [
SDN_SUBNETS_VNET2: list = [] 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"} NTP_INFO = {"server": "pool.ntp.org,time.cloudflare.com"}
+4 -1
View File
@@ -34,7 +34,10 @@ class TestOpen:
with patch("napalm_proxmox.driver.ProxmoxAPI", return_value=mock_api) as mock_cls: with patch("napalm_proxmox.driver.ProxmoxAPI", return_value=mock_api) as mock_cls:
drv.open() drv.open()
call_kwargs = mock_cls.call_args.kwargs 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" assert call_kwargs["token_value"] == "super-secret"
def test_open_connection_error(self): def test_open_connection_error(self):
+19 -8
View File
@@ -263,18 +263,29 @@ class TestGetRouteTo:
class TestLLDPNeighbors: 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 = ( 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): 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 "eth0" in result
assert result["eth0"][0]["hostname"] == "sw01.example.com" assert result["eth0"][0]["hostname"] == "sw01.example.com"
+10 -8
View File
@@ -3,6 +3,7 @@
from __future__ import annotations from __future__ import annotations
import pytest import pytest
from unittest.mock import patch
from napalm_proxmox import utils from napalm_proxmox import utils
@@ -126,6 +127,11 @@ class TestGetARPTable:
class TestGetMACAddressTable: 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 = ( BRIDGE_FDB = (
"aa:bb:cc:dd:ee:01 dev eth0 vlan 10 master vmbr0 permanent\n" "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" "cc:dd:ee:ff:00:11 dev eth0 vlan 20 master vmbr0\n"
@@ -142,10 +148,8 @@ class TestGetMACAddressTable:
return self.BRIDGE_FDB return self.BRIDGE_FDB
return "" 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} macs = {e["mac"] for e in result}
assert "aa:bb:cc:dd:ee:01" in macs assert "aa:bb:cc:dd:ee:01" in macs
@@ -155,9 +159,7 @@ class TestGetMACAddressTable:
return self.BRIDGE_FDB return self.BRIDGE_FDB
return "" 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"] static_entries = [e for e in result if e["mac"] == "aa:bb:cc:dd:ee:01"]
assert static_entries[0]["static"] is True assert static_entries[0]["static"] is True
+26 -12
View File
@@ -22,19 +22,33 @@ class TestGetVlans:
result = driver.get_vlans() result = driver.get_vlans()
assert "100000" in result 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() 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): def test_empty_sdn_returns_dict(self):
from unittest.mock import patch from unittest.mock import patch