feat: Add MAC address support for interface retrieval in NetworkAdapt…

README.md

@@ -119,6 +119,7 @@ To filter out specific Network Interfaces you can use following combinations of
 3) (`pci_device`|`family`|`speed`|`family`+`speed`) + (`random_interface`|`all_interfaces`)
 4) (`random_interface`|`all_interfaces`) 
 5) `interface_names`
+6) `mac_address`
 
 - `family`:
   - key of `DEVICE_IDS` from `mfd-const` e.g. `CPK`, `FVL`
@@ -144,6 +145,7 @@ def get_interfaces(
         random_interface: Optional[bool] = None,
         all_interfaces: Optional[bool] = None,
         namespace: Optional[str] = None,
+        mac_address:  MACAddress | None = None
     ) -> List["NetworkInterface"]
 ```
 
@@ -159,6 +161,7 @@ Expected combinations are:
 1) interface_name
 2) pci_address
 3) pci_device / family / speed + interface_index
+4) mac_address
 
 * source code:
 ```python
@@ -172,6 +175,7 @@ def get_interface(
         interface_index: Optional[int] = None,
         interface_name: Optional[str] = None,
         namespace: Optional[str] = None,
+        mac_address:  MACAddress | None = None
     ) -> "NetworkInterface"
 ```
 

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
@@ -49,8 +49,8 @@
     from .feature.ans import AnsFeatureType
     from .feature.cpu import CPUFeatureType
     from .feature.mac import MACFeatureType
-    from .feature.geneve import GeneveFeatureType
-    from .feature.gtp import GTPFeatureType
+    from .feature.geneve import GeneveTunnelFeatureType
+    from .feature.gtp import GTPTunnelFeatureType
 
 logger = logging.getLogger(__name__)
 add_logging_level(level_name="MODULE_DEBUG", level_value=log_levels.MODULE_DEBUG)
@@ -103,7 +103,7 @@ def __init__(self, *, connection: "Connection", **kwargs):
         # features of owner to be lazy initialized
         self._arp: "ARPFeatureType | None" = None
         self._dcb: "LinuxDcb | WindowsDcb | None" = None
-        self._driver: "DriverFeatureType " | None = None
+        self._driver: "DriverFeatureType | None" = None
         self._firewall: "FirewallFeatureType | None" = None
         self._ip: "IPFeatureType | None" = None
         self._nm: "NMFeatureType | None" = None
@@ -122,8 +122,8 @@ def __init__(self, *, connection: "Connection", **kwargs):
         self._ans: "AnsFeatureType | None" = None
         self._cpu: "CPUFeatureType | None" = None
         self._mac: "MACFeatureType | None" = None
-        self._geneve: "GeneveFeatureType | None" = None
-        self._gtp: "GTPFeatureType | None" = None
+        self._geneve: "GeneveTunnelFeatureType | None" = None
+        self._gtp: "GTPTunnelFeatureType | None" = None
 
     @property
     def arp(self) -> "ARPFeatureType":
@@ -136,7 +136,7 @@ def arp(self) -> "ARPFeatureType":
         return self._arp
 
     @property
-    def dcb(self) -> Union["Dcb", "LinuxDcb", "WindowsDcb"]:
+    def dcb(self) -> "Dcb | LinuxDcb | WindowsDcb":
         """DCB feature."""
         if self._dcb is None:
             from mfd_dcb import Dcb
@@ -336,7 +336,7 @@ def mac(self) -> "MACFeatureType":
         return self._mac
 
     @property
-    def geneve(self) -> "GeneveFeatureType":
+    def geneve(self) -> "GeneveTunnelFeatureType":
         """Geneve Tunnel feature."""
         if self._geneve is None:
             from .feature.geneve import BaseGeneveTunnelFeature
@@ -346,7 +346,7 @@ def geneve(self) -> "GeneveFeatureType":
         return self._geneve
 
     @property
-    def gtp(self) -> "GTPFeatureType":
+    def gtp(self) -> "GTPTunnelFeatureType":
         """GTP Tunnel feature."""
         if self._gtp is None:
             from .feature.gtp import BaseGTPTunnelFeature
@@ -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")

mfd_network_adapter/network_adapter_owner/feature/arp/__init__.py

@@ -2,12 +2,10 @@
 # SPDX-License-Identifier: MIT
 """Module for ARP feature."""
 
-from typing import Union
-
 from .base import BaseARPFeature
 from .esxi import ESXiARPFeature
 from .freebsd import FreeBSDARPFeature
 from .linux import LinuxARPFeature
 from .windows import WindowsARPFeature
 
-ARPFeatureType = Union[LinuxARPFeature, WindowsARPFeature, FreeBSDARPFeature, ESXiARPFeature]
+ARPFeatureType = BaseARPFeature | LinuxARPFeature | WindowsARPFeature | FreeBSDARPFeature | ESXiARPFeature

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(