initial commit

2026-05-29 09:24:39 +02:00
commit f3ecf14c8d
18 changed files with 4206 additions and 0 deletions
@@ -0,0 +1,163 @@
+"""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