Hyperv improvements for DHCP and default switch

Hyperv improvements for DHCP and default switch

Added below new functionality:
    1 get_default_internal_switch
    2 configure_dhcp

lisa/sut_orchestrator/hyperv/platform_.py

@@ -230,8 +230,7 @@ def _deploy_environment(self, environment: Environment, log: Logger) -> None:
         vm_name_prefix = f"{normalized_name}-e{environment.id}"
 
         hv = self._server.tools[HyperV]
-        default_switch = hv.get_default_external_switch()
-        assert default_switch, "No external switch found"
+        default_switch = hv.get_default_switch()
 
         extra_args = {
             x.command.lower(): x.args for x in self._hyperv_runbook.extra_args

lisa/tools/hyperv.py

@@ -4,14 +4,17 @@
 import re
 import time
 from dataclasses import dataclass, field
+from enum import Enum
 from typing import Dict, Optional
 
 from assertpy import assert_that
 from dataclasses_json import config, dataclass_json
 
+from lisa.base_tools import Service
 from lisa.executable import Tool
 from lisa.operating_system import Windows
 from lisa.tools.powershell import PowerShell
+from lisa.tools.windows_feature import WindowsFeatureManagement
 from lisa.util import LisaException
 from lisa.util.process import Process
 
@@ -22,6 +25,11 @@ class VMSwitch:
     name: str = field(metadata=config(field_name="Name"))
 
 
+class HypervSwitchType(Enum):
+    INTERNAL = "Internal"
+    EXTERNAL = "External"
+
+
 class HyperV(Tool):
     # 192.168.5.12
     IP_REGEX = r"\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}"
@@ -201,15 +209,21 @@ def enable_device_passthrough(self, name: str, mmio_mb: int = 5120) -> None:
             force_run=True,
         )
 
-    def get_default_external_switch(self) -> Optional[VMSwitch]:
+    def get_default_switch(
+        self, switch_type: HypervSwitchType = HypervSwitchType.EXTERNAL
+    ) -> Optional[VMSwitch]:
+        if switch_type not in (HypervSwitchType.INTERNAL, HypervSwitchType.EXTERNAL):
+            raise LisaException(f"Unknown switch type {switch_type}")
+
+        # get default switch of type `switch_type` from hyperv
         switch_json = self.node.tools[PowerShell].run_cmdlet(
-            'Get-VMSwitch | Where-Object {$_.SwitchType -eq "External"} '
+            f'Get-VMSwitch | Where-Object {{$_.SwitchType -eq "{switch_type}"}}'
             "| Select -First 1 | select Name | ConvertTo-Json",
             force_run=True,
         )
 
         if not switch_json:
-            return None
+            raise LisaException(f"Could not find default switch of type {switch_type}")
 
         return VMSwitch.from_json(switch_json)  # type: ignore
 
@@ -228,13 +242,13 @@ def delete_switch(self, name: str) -> None:
                 force_run=True,
             )
 
-    def create_switch(self, name: str) -> None:
+    def create_switch(self, name: str, switch_type: str = "Internal") -> None:
         # remove switch if it exists
         self.delete_switch(name)
 
         # create a new switch
         self.node.tools[PowerShell].run_cmdlet(
-            f"New-VMSwitch -Name {name} -SwitchType Internal",
+            f"New-VMSwitch -Name {name} -SwitchType {switch_type}",
             force_run=True,
         )
 
@@ -253,16 +267,17 @@ def exists_nat(self, name: str) -> bool:
         )
         return bool(output.strip() != "")
 
-    def delete_nat(self, name: str) -> None:
+    def delete_nat(self, name: str, fail_on_error: bool = True) -> None:
         if self.exists_nat(name):
             self.node.tools[PowerShell].run_cmdlet(
                 f"Remove-NetNat -Name {name} -Confirm:$false",
                 force_run=True,
+                fail_on_error=fail_on_error,
             )
 
     def create_nat(self, name: str, ip_range: str) -> None:
         # delete NAT if it exists
-        self.delete_nat(name)
+        self.delete_nat(name, fail_on_error=False)
 
         # create a new NAT
         self.node.tools[PowerShell].run_cmdlet(
@@ -381,30 +396,58 @@ def create_virtual_disk(self, name: str, pool_name: str, columns: int = 2) -> No
             force_run=True,
         )
 
-    def _check_exists(self) -> bool:
-        try:
-            self.node.tools[PowerShell].run_cmdlet(
-                "Get-VM",
-                force_run=True,
-            )
-            self._log.debug("Hyper-V is installed")
-            return True
-        except LisaException as e:
-            self._log.debug(f"Hyper-V not installed: {e}")
-            return False
+    def configure_dhcp(self, dhcp_scope_name: str = "DHCPInternalNAT") -> None:
+        powershell = self.node.tools[PowerShell]
+        service: Service = self.node.tools[Service]
+
+        # Install DHCP server
+        self.node.tools[WindowsFeatureManagement].install_feature("DHCP")
+
+        # Restart the DHCP server to make it available
+        service.restart_service("dhcpserver")
+
+        # check if DHCP server is already configured
+        output = powershell.run_cmdlet(
+            "Get-DhcpServerv4Scope",
+            force_run=True,
+            output_json=True,
+            fail_on_error=False,
+        )
+        if output:
+            return
+
+        # Configure the DHCP server to use the internal NAT network
+        powershell.run_cmdlet(
+            f'Add-DhcpServerV4Scope -Name "{dhcp_scope_name}" -StartRange 192.168.0.50 -EndRange 192.168.0.100 -SubnetMask 255.255.255.0',  # noqa: E501
+            force_run=True,
+        )
+
+        # Set the DHCP server options
+        powershell.run_cmdlet(
+            "Set-DhcpServerV4OptionValue -Router 192.168.0.1 -DnsServer 168.63.129.16",
+            force_run=True,
+        )
+
+        # Restart the DHCP server to apply the changes
+        service.restart_service("dhcpserver")
 
     def _install(self) -> bool:
         assert isinstance(self.node.os, Windows)
 
+        # check if Hyper-V is already installed
+        if self._check_exists():
+            return True
+
         # enable hyper-v
-        self.node.tools[PowerShell].run_cmdlet(
-            "Install-WindowsFeature -Name Hyper-V -IncludeManagementTools",
-            force_run=True,
-        )
+        self.node.tools[WindowsFeatureManagement].install_feature("Hyper-V")
 
         # reboot node
         self.node.reboot()
 
+        # wait for Hyper-V services to start
+        self.node.tools[Service].wait_for_service_start("vmms")
+        self.node.tools[Service].wait_for_service_start("vmcompute")
+
         return self._check_exists()
 
     def _run_hyperv_cmdlet(
@@ -422,3 +465,6 @@ def _run_hyperv_cmdlet(
                 f"{cmd} {args} {extra_args.get(cmd.lower(), '')}", force_run=force_run
             )
         )
+
+    def _check_exists(self) -> bool:
+        return self.node.tools[WindowsFeatureManagement].is_installed("Hyper-V")