Compare commits
2
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
20fcf2ebc3 | ||
|
|
51f67704e1 |
@@ -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
|
||||
|
||||
@@ -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.
|
||||
|
||||
|
||||
@@ -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.
|
||||
|
||||
+4
-1
@@ -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"}
|
||||
|
||||
|
||||
@@ -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):
|
||||
|
||||
+19
-8
@@ -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"
|
||||
|
||||
|
||||
@@ -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
|
||||
|
||||
+26
-12
@@ -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
|
||||
|
||||
Reference in New Issue
Block a user