diff --git a/README.md b/README.md index f06acaf..f08b120 100644 --- a/README.md +++ b/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" ``` diff --git a/examples/linux_owner_example.py b/examples/linux_owner_example.py index 93fce40..c90c2f2 100644 --- a/examples/linux_owner_example.py +++ b/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 \ No newline at end of file diff --git a/mfd_network_adapter/network_adapter_owner/base.py b/mfd_network_adapter/network_adapter_owner/base.py index cfbb4e9..99b4b88 100644 --- a/mfd_network_adapter/network_adapter_owner/base.py +++ b/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]: diff --git a/mfd_network_adapter/network_adapter_owner/esxi.py b/mfd_network_adapter/network_adapter_owner/esxi.py index fb48fe0..737932c 100644 --- a/mfd_network_adapter/network_adapter_owner/esxi.py +++ b/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") diff --git a/mfd_network_adapter/network_adapter_owner/feature/arp/__init__.py b/mfd_network_adapter/network_adapter_owner/feature/arp/__init__.py index fdf9813..419f0e3 100644 --- a/mfd_network_adapter/network_adapter_owner/feature/arp/__init__.py +++ b/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 diff --git a/tests/unit/test_mfd_network_adapter/test_network_adapter_owner/test_esxi_network_owner.py b/tests/unit/test_mfd_network_adapter/test_network_adapter_owner/test_esxi_network_owner.py index dc81ff7..48c5e08 100644 --- a/tests/unit/test_mfd_network_adapter/test_network_adapter_owner/test_esxi_network_owner.py +++ b/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(