"""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