NetworkInterface added

lisa/sut_orchestrator/hyperv/features.py

@@ -0,0 +1,68 @@
+from typing import Type, Any, List
+from lisa import schema
+from lisa.feature import Feature
+from lisa.tools import IpInfo, HyperV
+
+
+class NetworkInterface(Feature):
+    @classmethod
+    def settings_type(cls) -> Type[schema.FeatureSettings]:
+        return schema.NetworkInterfaceOptionSettings
+
+    @classmethod
+    def can_disable(cls) -> bool:
+        return False
+
+    def enabled(self) -> bool:
+        return True
+
+    def switch_sriov(
+        self, enable: bool, wait: bool = True, reset_connections: bool = True
+    ) -> None:
+        raise NotImplementedError("SR-IOV is not supported on Hyper-V platform.")
+
+    def is_enabled_sriov(self) -> bool:
+        return False
+
+    def create_route_table(
+        self,
+        nic_name: str,
+        route_name: str,
+        subnet_mask: str,
+        dest_hop: str,
+        em_first_hop: str = "",
+        next_hop_type: str = "",
+    ) -> None:
+        raise NotImplementedError(
+            "Route table creation is not supported on Hyper-V platform."
+        )
+
+    def switch_ip_forwarding(self, enable: bool, private_ip_addr: str = "") -> None:
+        raise NotImplementedError("IP forwarding is not supported on Hyper-V platform.")
+
+    def attach_nics(
+        self, extra_nic_count: int, enable_accelerated_networking: bool = True
+    ) -> None:
+        hyperv: HyperV = self._node.tools[HyperV]
+        for i in range(extra_nic_count):
+            nic_name = f"{self._node.name}-extra-{i}"
+            hyperv.add_network_adapter(self._node.name, nic_name)
+
+    def remove_extra_nics(self) -> None:
+        hyperv: HyperV = self._node.tools[HyperV]
+        hyperv.remove_extra_network_adapters(self._node.name)
+
+    def reload_module(self) -> None:
+        raise NotImplementedError("Module reload is not supported on Hyper-V platform.")
+
+    def get_nic_count(self, is_sriov_enabled: bool = True) -> int:
+        hyperv = self._node.tools[HyperV]
+        return hyperv.get_network_adapter_count(self._node.name)
+
+    def get_all_primary_nics_ip_info(self) -> List[IpInfo]:
+        hyperv = self._node.tools[HyperV]
+        return hyperv.get_all_primary_nics_ip_info(self._node.name)
+
+    def _initialize(self, *args: Any, **kwargs: Any) -> None:
+        self.origin_extra_synthetic_nics_count = 0
+        self.origin_extra_sriov_nics_count = 0

lisa/tools/hyperv.py

@@ -5,7 +5,7 @@
 import time
 from dataclasses import dataclass
 from enum import Enum
-from typing import Dict, Optional
+from typing import Dict, List, Optional
 
 from assertpy import assert_that
 from dataclasses_json import dataclass_json
@@ -15,6 +15,7 @@
 from lisa.operating_system import Windows
 from lisa.tools.powershell import PowerShell
 from lisa.tools.windows_feature import WindowsFeatureManagement
+from lisa.tools import IpInfo
 from lisa.util import LisaException
 from lisa.util.process import Process
 
@@ -445,6 +446,53 @@ def enable_internal_dhcp(self, dhcp_scope_name: str = "DHCPInternalNAT") -> None
         # Restart the DHCP server to apply the changes
         service.restart_service("dhcpserver")
 
+    def add_network_adapter(self, vm_name: str, nic_name: str) -> None:
+        self.node.tools[PowerShell].run_cmdlet(
+            f"Add-VMNetworkAdapter -VMName {vm_name} -Name {nic_name}",
+            force_run=True,
+        )
+
+    def remove_extra_network_adapters(self, vm_name: str) -> None:
+        adapters = self.node.tools[PowerShell].run_cmdlet(
+            f"Get-VMNetworkAdapter -VMName {vm_name} "
+            "| Where-Object {{$_.Name -like '*-extra-*'}}",
+            force_run=True,
+            output_json=True,
+        )
+        for adapter in adapters:
+            self.node.tools[PowerShell].run_cmdlet(
+                f"Remove-VMNetworkAdapter -VMName {vm_name} -Name {adapter['Name']}",
+                force_run=True,
+            )
+
+    def get_network_adapter_count(self, vm_name: str) -> int:
+        output = self.node.tools[PowerShell].run_cmdlet(
+            f"Get-VMNetworkAdapter -VMName {vm_name}",
+            force_run=True,
+            output_json=True,
+        )
+        return len(output)
+
+    def get_all_primary_nics_ip_info(self, vm_name: str) -> List[IpInfo]:
+        output = self.node.tools[PowerShell].run_cmdlet(
+            f"Get-VMNetworkAdapter -VMName {vm_name} "
+            "| Select-Object Name, MacAddress, IPAddresses",
+            force_run=True,
+            output_json=True,
+        )
+        ip_info_list = []
+        for adapter in output:
+            for ip in adapter["IPAddresses"]:
+                if re.match(self.IP_REGEX, ip):
+                    ip_info_list.append(
+                        IpInfo(
+                            mac_addr=adapter["MacAddress"],
+                            ip_addr=ip,
+                            nic_name=adapter["Name"],
+                        )
+                    )
+        return ip_info_list
+
     def _install(self) -> bool:
         assert isinstance(self.node.os, Windows)