Compare commits
2
Commits
39f8d80352
...
20fcf2ebc3
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
20fcf2ebc3 | ||
|
|
51f67704e1 |
@@ -78,6 +78,10 @@ class ProxmoxDriver(
|
|||||||
|
|
||||||
VENDOR = "Proxmox"
|
VENDOR = "Proxmox"
|
||||||
DRIVER_NAME = "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 = [
|
PORT_SPECS = [
|
||||||
PortSpec("https", 8006, weight=8.0),
|
PortSpec("https", 8006, weight=8.0),
|
||||||
]
|
]
|
||||||
@@ -137,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,
|
||||||
@@ -206,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
|
||||||
|
|||||||
@@ -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.
|
||||||
|
|
||||||
|
|||||||
@@ -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
@@ -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"}
|
||||||
|
|
||||||
|
|||||||
@@ -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):
|
||||||
|
|||||||
+14
-3
@@ -263,17 +263,28 @@ 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"
|
||||||
|
"-------------------------------------------------------------------------------\n"
|
||||||
|
"Interface: eth0, via: LLDP, RID: 1, Time: 0 day, 00:11:22\n"
|
||||||
|
" Chassis:\n"
|
||||||
" SysName: sw01.example.com\n"
|
" SysName: sw01.example.com\n"
|
||||||
|
" Port:\n"
|
||||||
" PortID: ifname GigabitEthernet1/0/1\n"
|
" PortID: ifname GigabitEthernet1/0/1\n"
|
||||||
" Interface: eth1\n"
|
"-------------------------------------------------------------------------------\n"
|
||||||
|
"Interface: eth1, via: LLDP, RID: 2, Time: 0 day, 00:11:22\n"
|
||||||
|
" Chassis:\n"
|
||||||
" SysName: sw02.example.com\n"
|
" SysName: sw02.example.com\n"
|
||||||
|
" Port:\n"
|
||||||
" PortID: ifname GigabitEthernet1/0/2\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"
|
||||||
|
|||||||
@@ -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,9 +148,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()
|
||||||
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
@@ -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
|
||||||
|
|||||||
Reference in New Issue
Block a user