diff --git a/napalm_proxmox/driver.py b/napalm_proxmox/driver.py index d2b7748..4ea172c 100644 --- a/napalm_proxmox/driver.py +++ b/napalm_proxmox/driver.py @@ -141,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 "@" 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, @@ -210,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 diff --git a/napalm_proxmox/interfaces_mixin.py b/napalm_proxmox/interfaces_mixin.py index 3716da4..782f194 100644 --- a/napalm_proxmox/interfaces_mixin.py +++ b/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() + # " dev lladdr " — 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. diff --git a/napalm_proxmox/sdn_mixin.py b/napalm_proxmox/sdn_mixin.py index e09531c..f77143e 100644 --- a/napalm_proxmox/sdn_mixin.py +++ b/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 { - vid: entry for vid, entry in result.items() - if entry.get("tagged") or entry.get("untagged") - } + # Deliberately unfiltered. This used to drop entries with no member + # ports, which hid every configured SDN VNet that no VM happened to be + # 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. diff --git a/tests/conftest.py b/tests/conftest.py index d2729e9..1bd20a0 100644 --- a/tests/conftest.py +++ b/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"} diff --git a/tests/test_connection.py b/tests/test_connection.py index a36844e..e17fc43 100644 --- a/tests/test_connection.py +++ b/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 + # "!" 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): diff --git a/tests/test_misc.py b/tests/test_misc.py index 05f5134..d005032 100644 --- a/tests/test_misc.py +++ b/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" - " SysName: sw01.example.com\n" - " PortID: ifname GigabitEthernet1/0/1\n" - " Interface: eth1\n" - " SysName: sw02.example.com\n" - " PortID: ifname GigabitEthernet1/0/2\n" + "LLDP neighbors:\n" + "-------------------------------------------------------------------------------\n" + "Interface: eth0, via: LLDP, RID: 1, Time: 0 day, 00:11:22\n" + " Chassis:\n" + " SysName: sw01.example.com\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} - result = driver.get_lldp_neighbors() + with patch.object(driver, "_exec_ssh_command", return_value=self.LLDP_SUMMARY): + result = driver.get_lldp_neighbors() assert "eth0" in result assert result["eth0"][0]["hostname"] == "sw01.example.com" diff --git a/tests/test_ovs_and_arp.py b/tests/test_ovs_and_arp.py index b514bb6..32f9053 100644 --- a/tests/test_ovs_and_arp.py +++ b/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: { - "data": _exec(command) - } - result = driver.get_mac_address_table() + with patch.object(driver, "_exec_ssh_command", side_effect=_exec): + 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: { - "data": _exec(command) - } - result = driver.get_mac_address_table() + with patch.object(driver, "_exec_ssh_command", side_effect=_exec): + 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 diff --git a/tests/test_sdn.py b/tests/test_sdn.py index 3c68d97..b24bed4 100644 --- a/tests/test_sdn.py +++ b/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): - # Simulate bridge vlan output - bridge_output = ( - "vmbr0 1\n" - " 10\n" - " 20\n" - "eth0 1\n" - ) - driver._node_api().execute.post.return_value = {"data": bridge_output} + def test_membership_derived_from_vm_configs(self, driver): + """Without OVS ports, VLAN membership comes from each VM's netN config. + + This replaces a test for a `bridge vlan show` fallback that no longer + exists — it also asserted an "interfaces" key this method has never + produced, so it could not have passed against any version of the code. + """ + node = driver._node_api() + 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 - entry = result.get("1", {}) - assert "vmbr0" in entry.get("interfaces", []) + assert "vmbr0" in result["10"]["untagged"] + + 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