164 lines
5.5 KiB
Python
164 lines
5.5 KiB
Python
"""Tests for OVS-related helpers and MAC/ARP methods."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import pytest
|
|
|
|
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:
|
|
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 ""
|
|
|
|
driver._node_api().execute.post.side_effect = lambda command: {
|
|
"data": _exec(command)
|
|
}
|
|
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 ""
|
|
|
|
driver._node_api().execute.post.side_effect = lambda command: {
|
|
"data": _exec(command)
|
|
}
|
|
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
|