diff --git a/lisa/sut_orchestrator/hyperv/features.py b/lisa/sut_orchestrator/hyperv/features.py new file mode 100644 index 0000000000..5a37705d84 --- /dev/null +++ b/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 diff --git a/lisa/tools/hyperv.py b/lisa/tools/hyperv.py index d328aa72f5..40e8e0adc2 100644 --- a/lisa/tools/hyperv.py +++ b/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)