feat: Add MAC address support for interface retrieval in NetworkAdapterOwner

Signed-off-by: Lasota, Adrian <[email protected]>

Author: adrianlasota

examples/linux_owner_example.py

@@ -5,7 +5,7 @@
 
 from mfd_network_adapter.network_adapter_owner.linux import LinuxNetworkAdapterOwner
 from mfd_connect import RPyCConnection
-from mfd_typing import PCIAddress
+from mfd_typing import PCIAddress, MACAddress
 
 owner = LinuxNetworkAdapterOwner(connection=RPyCConnection("10.11.12.13"))
 
@@ -38,3 +38,5 @@
 # utils - get_same_pci_bus_interfaces example
 interface = owner.get_interface(interface_name="eth1")
 owner.utils.get_same_pci_bus_interfaces(interface=interface)  # get interfaces on the same PCI bus as eth1 interface
+
+interface = owner.get_interfaces(mac_address=MACAddress("00:1A:2B:3C:4D:5E")) # get interfaces by MAC address

mfd_network_adapter/network_adapter_owner/base.py

@@ -11,7 +11,7 @@
 
 from mfd_common_libs import log_levels, add_logging_level
 from mfd_const import MANAGEMENT_NETWORK, Family, Speed
-from mfd_typing import OSName, PCIDevice, PCIAddress, VendorID
+from mfd_typing import OSName, PCIDevice, PCIAddress, VendorID, MACAddress
 from mfd_typing.network_interface import InterfaceInfo, WindowsInterfaceInfo, LinuxInterfaceInfo
 
 from .exceptions import NetworkAdapterConnectedOSNotSupported, NetworkAdapterIncorrectData
@@ -376,6 +376,7 @@ def get_interfaces(
         interface_names: Optional[List[str]] = None,
         random_interface: Optional[bool] = None,
         all_interfaces: Optional[bool] = None,
+        mac_address: MACAddress | None = None,
     ) -> List["NetworkInterface"]:
         """
         Get Network Interface objects.
@@ -388,6 +389,7 @@ def get_interfaces(
         3) (`pci_device`|`family`|`speed`|`family`+`speed`) + (`random_interface`|`all_interfaces`)
         4) (`random_interface`|`all_interfaces`)
         5) `interface_names`
+        6) `mac_address`
 
         :param pci_address: PCI address
         :param pci_device: PCI device
@@ -397,6 +399,7 @@ def get_interfaces(
         :param interface_names: Names of the interfaces
         :param random_interface: Flag - random interface
         :param all_interfaces: Flag - all interfaces
+        :param mac_address: MAC Address of the interface
         :return: List of Network Interface objects depending on passed args
         """
         all_interfaces_info: List[InterfaceInfoType] = self._get_all_interfaces_info()
@@ -410,6 +413,7 @@ def get_interfaces(
             interface_names=interface_names,
             random_interface=random_interface,
             all_interfaces=all_interfaces,
+            mac_address=mac_address,
         )
 
         if not filtered_info:
@@ -427,6 +431,7 @@ def get_interface(
         interface_index: Optional[int] = None,
         interface_name: Optional[str] = None,
         namespace: Optional[str] = None,
+        mac_address: MACAddress | None = None,
     ) -> "NetworkInterface":
         """
         Get single interface of network adapter.
@@ -435,6 +440,7 @@ def get_interface(
             1) interface_name
             2) pci_address
             3) pci_device / family / speed + interface_index
+            4) mac_address
 
         :param pci_address: PCI address
         :param pci_device: PCI device
@@ -443,6 +449,7 @@ def get_interface(
         :param interface_index: Index of interface, like 0 - first interface of adapter
         :param interface_name: Name of the interface
         :param namespace: Linux namespace, in which cmd will be executed
+        :param mac_address: MAC Address of the interface
         :return: Network Interface
         """
         all_interfaces_info: List[InterfaceInfoType] = self._get_all_interfaces_info()
@@ -455,6 +462,7 @@ def get_interface(
             interface_indexes=[interface_index] if interface_index is not None else [],
             interface_names=[interface_name] if interface_name is not None else [],
             all_interfaces=True,
+            mac_address=mac_address,
         )
 
         if len(filtered_info) > 1:
@@ -479,6 +487,7 @@ def _filter_interfaces_info(
         interface_names: Optional[List[str]] = None,
         random_interface: Optional[bool] = None,
         all_interfaces: Optional[bool] = None,
+        mac_address: MACAddress | None = None,
     ) -> List[InterfaceInfoType]:
         """
         Filter list based on passed criteria.
@@ -492,6 +501,7 @@ def _filter_interfaces_info(
         :param interface_names: Names of the interfaces
         :param random_interface: Flag - random interface
         :param all_interfaces: Flag - all interfaces
+        :param mac_address: MAC Address of the interface
         :return: Filtered list of InterfaceInfo objects
         """
         self._validate_filtering_args(
@@ -506,6 +516,8 @@ def _filter_interfaces_info(
             selected = [info for info in all_interfaces_info if info.name in interface_names]
         elif family is not None or speed is not None:
             selected = self._get_info_by_speed_and_family(all_interfaces_info, family=family, speed=speed)
+        elif mac_address is not None:
+            selected = [info for info in all_interfaces_info if info.mac_address == mac_address]
         else:
             selected = all_interfaces_info
 
@@ -565,6 +577,7 @@ def _validate_filtering_args(
         family: Optional[str] = None,
         speed: Optional[str] = None,
         interface_names: Optional[List[str]] = None,
+        mac_address: MACAddress | None = None,
     ) -> None:
         """Validate passed args based on expected combinations."""
         passed_combinations_amount = sum(
@@ -573,6 +586,7 @@ def _validate_filtering_args(
                 pci_device is not None,
                 interface_names is not None and interface_names != [],
                 family is not None or speed is not None,
+                mac_address is not None,
             ]
         )
 
@@ -586,7 +600,12 @@ def _validate_filtering_args(
             return
 
         NetworkAdapterOwner._log_selection_criteria(
-            pci_address=pci_address, pci_device=pci_device, interface_names=interface_names, family=family, speed=speed
+            pci_address=pci_address,
+            pci_device=pci_device,
+            interface_names=interface_names,
+            family=family,
+            speed=speed,
+            mac_address=mac_address,
         )
 
     def _get_all_interfaces_info(self) -> List[InterfaceInfoType]:

mfd_network_adapter/network_adapter_owner/esxi.py

@@ -184,6 +184,7 @@ def _filter_interfaces_info(
         interface_names: Optional[List[str]] = None,
         random_interface: Optional[bool] = None,
         all_interfaces: Optional[bool] = None,
+        mac_address: MACAddress | None = None,
     ) -> List[InterfaceInfo]:
         """
         Filter all interfaces based on selected criteria.
@@ -197,6 +198,7 @@ def _filter_interfaces_info(
         :param interface_names: Names of the interfaces
         :param random_interface: Flag - random interface
         :param all_interfaces: Flag - all interfaces
+        :param mac_address: MAC Address of the interface
         :return: List of Network Interface objects depending on passed args
         """
         selected = []
@@ -225,6 +227,8 @@ def _filter_interfaces_info(
                 continue
             if interface_names and interface.name not in interface_names:
                 continue
+            if mac_address and interface.mac_address != mac_address:
+                continue
 
             selected.append(interface)
 
@@ -245,6 +249,7 @@ def get_interface(
         interface_index: Optional[int] = None,
         interface_name: Optional[str] = None,
         namespace: Optional[str] = None,
+        mac_address: MACAddress | None = None,
     ) -> "ESXiNetworkInterface":
         """
         Get single interface of network adapter.
@@ -253,6 +258,7 @@ def get_interface(
             1) interface_name
             2) pci_address
             3) pci_device / family / speed + interface_index
+            4) mac_address
 
         :param pci_address: PCI address
         :param pci_device: PCI device
@@ -261,6 +267,7 @@ def get_interface(
         :param interface_index: Index of interface, like 0 - first interface of adapter
         :param interface_name: Name of the interface
         :param namespace: Linux namespace, in which cmd will be executed
+        :param mac_address: MAC Address of the interface
         :return: Network Interface
         """
         interface_indexes = [interface_index] if interface_index is not None else []
@@ -274,6 +281,7 @@ def get_interface(
             speed=speed,
             interface_indexes=interface_indexes,
             interface_names=interface_names,
+            mac_address=mac_address,
         )
         if len(interfaces) < 1:
             raise NetworkAdapterNotFound("Could not find adapter with selected parameters")

tests/unit/test_mfd_network_adapter/test_network_adapter_owner/test_esxi_network_owner.py

@@ -78,7 +78,7 @@ class TestESXiNetworkOwner:
         vmnic6  0000:31:00.2 igbn        Down 0Mbps      Half   00:00:00:00:00:00 1500   Intel(R) I350 Gigabit Network Connection
         vmnic7  0000:31:00.3 igbn        Down 0Mbps      Half   00:00:00:00:00:00 1500   Intel(R) I350 Gigabit Network Connection
         vmnic8  0000:b1:00.0 ixgben      Up   10000Mbps  Full   00:00:00:00:00:00 1500   Intel(R) 82599 10 Gigabit Dual Port Network Connection
-        vmnic9  0000:b1:00.1 ixgben      Up   10000Mbps  Full   00:00:00:00:00:00 1500   Intel(R) 82599 10 Gigabit Dual Port Network Connection
+        vmnic9  0000:b1:00.1 ixgben      Up   10000Mbps  Full   00:00:00:00:00:01 1500   Intel(R) 82599 10 Gigabit Dual Port Network Connection
         """  # noqa: E501
     )
 
@@ -173,6 +173,14 @@ def test_filter_interfaces_pci_address(self, owner2):
         assert len(devices) == 1
         assert devices[0].name == "vmnic13"
 
+    def test_filter_interfaces_mac_address(self, owner2):
+        all_interfaces_info = owner2._get_all_interfaces_info()
+        devices = owner2._filter_interfaces_info(
+            all_interfaces_info=all_interfaces_info, mac_address=MACAddress("00:00:00:00:00:01")
+        )
+        assert len(devices) == 1
+        assert devices[0].name == "vmnic9"
+
     def test_filter_interfaces_pci_device_all_1(self, owner2):
         all_interfaces_info = owner2._get_all_interfaces_info()
         devices = owner2._filter_interfaces_info(