Christian Manivong
20fcf2ebc3
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.
166 lines
5.8 KiB
Python
166 lines
5.8 KiB
Python
"""Tests for OVS-related helpers and MAC/ARP methods."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import pytest
|
|
from unittest.mock import patch
|
|
|
|
from napalm_proxmox import utils
|
|
|
|
|
|
PROC_NET_ARP = """\
|
|
IP address HW type Flags HW address Mask Device
|
|
10.0.0.254 0x1 0x2 aa:bb:cc:dd:ee:ff * eth0
|
|
10.0.0.10 0x1 0x2 11:22:33:44:55:66 * vmbr0
|
|
10.0.0.20 0x1 0x0 00:00:00:00:00:00 * eth0
|
|
"""
|
|
|
|
OVS_VSCTL_SHOW = """\
|
|
Bridge vmbr1
|
|
Port eth1
|
|
Port vnet0
|
|
Port vmbr1
|
|
Bridge vmbr2
|
|
Port eth2
|
|
"""
|
|
|
|
BRIDGE_FDB_OUTPUT = """\
|
|
aa:bb:cc:dd:ee:01 dev eth0 vlan 10 master vmbr0 permanent
|
|
cc:dd:ee:ff:00:11 dev eth0 vlan 20 master vmbr0
|
|
"""
|
|
|
|
OVS_FDB_SHOW = """\
|
|
LOCAL 0 aa:bb:cc:dd:ee:03 0
|
|
1 10 11:22:33:44:55:66 100
|
|
2 20 66:55:44:33:22:11 200
|
|
"""
|
|
|
|
|
|
class TestNormalizeMAC:
|
|
def test_colon_separated(self):
|
|
assert utils.normalize_mac("AA:BB:CC:DD:EE:FF") == "aa:bb:cc:dd:ee:ff"
|
|
|
|
def test_dash_separated(self):
|
|
assert utils.normalize_mac("AA-BB-CC-DD-EE-FF") == "aa:bb:cc:dd:ee:ff"
|
|
|
|
def test_empty_string(self):
|
|
assert utils.normalize_mac("") == ""
|
|
|
|
def test_already_normalized(self):
|
|
assert utils.normalize_mac("aa:bb:cc:dd:ee:ff") == "aa:bb:cc:dd:ee:ff"
|
|
|
|
|
|
class TestParseARPOutput:
|
|
def test_parses_valid_entries(self):
|
|
entries = utils.parse_arp_output(PROC_NET_ARP)
|
|
# 10.0.0.20 has 00:00:00:00:00:00 → excluded by ARP parse (not by proc/net/arp parser)
|
|
# proc/net/arp in driver excludes 00:00..., parse_arp_output is for `arp -n` format
|
|
# The proc/net/arp format is handled inline in driver.
|
|
# parse_arp_output handles `arp -n` which uses "HWtype HWaddress" columns
|
|
# Let's test with arp -n format
|
|
arp_n = (
|
|
"Address HWtype HWaddress Flags Iface\n"
|
|
"10.0.0.254 ether aa:bb:cc:dd:ee:ff C eth0\n"
|
|
"10.0.0.10 ether 11:22:33:44:55:66 C vmbr0\n"
|
|
)
|
|
entries = utils.parse_arp_output(arp_n)
|
|
assert len(entries) == 2
|
|
assert entries[0]["ip"] == "10.0.0.254"
|
|
assert entries[0]["mac"] == "aa:bb:cc:dd:ee:ff"
|
|
assert entries[0]["interface"] == "eth0"
|
|
|
|
def test_skips_incomplete(self):
|
|
arp_n = (
|
|
"Address HWtype HWaddress Flags Iface\n"
|
|
"10.0.0.1 ether (incomplete) C eth0\n"
|
|
)
|
|
entries = utils.parse_arp_output(arp_n)
|
|
assert len(entries) == 0
|
|
|
|
|
|
class TestParseOVSVsctlShow:
|
|
def test_bridges_found(self):
|
|
result = utils.parse_ovs_vsctl_show(OVS_VSCTL_SHOW)
|
|
assert "vmbr1" in result
|
|
assert "vmbr2" in result
|
|
|
|
def test_ports_found(self):
|
|
result = utils.parse_ovs_vsctl_show(OVS_VSCTL_SHOW)
|
|
assert "eth1" in result["vmbr1"]["ports"]
|
|
assert "vnet0" in result["vmbr1"]["ports"]
|
|
# The bridge self-port should not appear in ports list
|
|
assert "vmbr1" not in result["vmbr1"]["ports"]
|
|
|
|
def test_second_bridge_ports(self):
|
|
result = utils.parse_ovs_vsctl_show(OVS_VSCTL_SHOW)
|
|
assert "eth2" in result["vmbr2"]["ports"]
|
|
|
|
|
|
class TestGetARPTable:
|
|
PROC_ARP = (
|
|
"IP address HW type Flags HW address Mask Device\n"
|
|
"10.0.0.254 0x1 0x2 aa:bb:cc:dd:ee:ff * eth0\n"
|
|
"10.0.0.10 0x1 0x2 11:22:33:44:55:66 * vmbr0\n"
|
|
"10.0.0.20 0x1 0x0 00:00:00:00:00:00 * eth0\n"
|
|
)
|
|
|
|
def test_arp_entries_returned(self, driver):
|
|
driver._node_api().execute.post.return_value = {"data": self.PROC_ARP}
|
|
result = driver.get_arp_table()
|
|
assert len(result) == 2 # 00:00:00:00:00:00 excluded
|
|
|
|
def test_mac_normalized(self, driver):
|
|
driver._node_api().execute.post.return_value = {"data": self.PROC_ARP}
|
|
result = driver.get_arp_table()
|
|
macs = {e["mac"] for e in result}
|
|
assert "aa:bb:cc:dd:ee:ff" in macs
|
|
|
|
def test_vrf_filter(self, driver):
|
|
driver._node_api().execute.post.return_value = {"data": self.PROC_ARP}
|
|
result = driver.get_arp_table(vrf="vmbr0")
|
|
assert all(e["interface"] == "vmbr0" for e in result)
|
|
|
|
def test_empty_when_exec_unavailable(self, driver):
|
|
driver._node_api().execute.post.return_value = {"data": ""}
|
|
result = driver.get_arp_table()
|
|
assert result == []
|
|
|
|
|
|
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"
|
|
)
|
|
|
|
OVS_APPCTL = (
|
|
" LOCAL 0 aa:bb:cc:dd:ee:03 0\n"
|
|
" 1 10 11:22:33:44:55:66 100\n"
|
|
)
|
|
|
|
def test_linux_bridge_entries(self, driver):
|
|
def _exec(cmd):
|
|
if "bridge fdb" in cmd:
|
|
return self.BRIDGE_FDB
|
|
return ""
|
|
|
|
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
|
|
|
|
def test_static_flag(self, driver):
|
|
def _exec(cmd):
|
|
if "bridge fdb" in cmd:
|
|
return self.BRIDGE_FDB
|
|
return ""
|
|
|
|
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
|