net, vm import: clone source VM before migration

tests/network/provider_migration/conftest.py

@@ -1,3 +1,5 @@
+import contextlib
+import uuid
 from collections.abc import Generator
 from typing import Final
 
@@ -19,13 +21,18 @@
 from tests.network.libs import cluster_user_defined_network as libcudn
 from tests.network.libs.label_selector import LabelSelector
 from tests.network.libs.vm_factory import udn_vm
-from tests.network.provider_migration.libprovider import SourceHypervisorProvider, extract_vm_primary_network_data
+from tests.network.provider_migration.libprovider import (
+    SourceHypervisorProvider,
+    VmNotFoundError,
+    extract_vm_primary_network_data,
+)
 from utilities.bitwarden import get_cnv_tests_secret_by_name
 from utilities.constants import OS_FLAVOR_FEDORA
 
 CUDN_LABEL: Final[dict] = {"cudn": "mtv"}
 CUDN_SUBNET_IPV4: Final[str] = "192.168.100.0/24"
 MTV_RESOURCE_TIMEOUT_SEC: Final[int] = 30
+SOURCE_VM_NAME: Final[str] = f"cnv-vm-for-mtv-import-{uuid.uuid4().hex[:8]}"
 VDDK_IMAGE: Final[str] = "quay.io/libvirt_v2v_cnv/vddk:8.0.1"
 
 
@@ -70,10 +77,11 @@ def source_vm_network_data(source_hypervisor_data: dict) -> Generator[tuple[str,
         password=source_hypervisor_data["password"],
     ) as sp:
         try:
-            vm = sp.power_on_vm(vm_name=source_hypervisor_data["vm_name"])
+            vm = sp.clone_vm(template_name=source_hypervisor_data["vm_name"], clone_name=SOURCE_VM_NAME, power_on=True)
             yield extract_vm_primary_network_data(vm=vm)
         finally:
-            sp.power_off_vm(vm_name=source_hypervisor_data["vm_name"])
+            with contextlib.suppress(VmNotFoundError):
+                sp.delete_vm(vm_name=SOURCE_VM_NAME)
 
 
 @pytest.fixture(scope="module")
@@ -239,7 +247,7 @@ def mtv_migration_plan_to_cudn_ns(
         destination_provider_name=mtv_target_provider.name,
         destination_provider_namespace=mtv_target_provider.namespace,
         target_namespace=cudn_namespace.name,
-        virtual_machines_list=[{"name": source_hypervisor_data["vm_name"]}],
+        virtual_machines_list=[{"name": SOURCE_VM_NAME}],
         type="cold",
         target_power_state="on",
         preserve_static_ips=True,
@@ -271,7 +279,7 @@ def imported_cudn_vm(
     admin_client: DynamicClient, cudn_namespace: Namespace, source_hypervisor_data: dict, mtv_migration_to_cudn_ns: None
 ) -> Generator[BaseVirtualMachine]:
     vm = BaseVirtualMachine.from_existing(
-        name=source_hypervisor_data["vm_name"],
+        name=SOURCE_VM_NAME,
         namespace=cudn_namespace.name,
         client=admin_client,
         os_distribution=OS_FLAVOR_FEDORA,

tests/network/provider_migration/libprovider.py

@@ -2,6 +2,7 @@
 from functools import cache
 
 from pyVim.connect import Disconnect, SmartConnect
+from pyVim.task import WaitForTask
 from pyVmomi import vim
 from timeout_sampler import TimeoutExpiredError, retry
 
@@ -17,7 +18,8 @@ class IfaceNotFoundError(Exception):
 
 
 class SourceHypervisorProvider:
-    """A simple source provider context manager to manage VMs. Supports only basic operations: power on/off."""
+    """A simple source provider context manager to manage VMs. Supports only basic operations: power on/off,
+    clone and delete."""
 
     def __init__(self, host: str, username: str, password: str) -> None:
         self.host = host
@@ -64,6 +66,29 @@ def power_off_vm(self, vm_name: str) -> None:
             LOGGER.error(f"Timeout while trying to power off VM '{vm_name}'")
             raise
 
+    def clone_vm(self, template_name: str, clone_name: str, power_on: bool = False) -> vim.VirtualMachine:
+        LOGGER.info(f"Cloning VM '{template_name}' to '{clone_name}'")
+        template_vm = self.get_vm_by_name(name=template_name)
+        clone_spec = vim.vm.CloneSpec(location=vim.vm.RelocateSpec(), powerOn=False)
+        task = template_vm.Clone(name=clone_name, folder=template_vm.parent, spec=clone_spec)
+        WaitForTask(task=task, maxWaitTime=600)
+        LOGGER.info(f"Successfully cloned VM '{clone_name}'")
+
+        return self.power_on_vm(vm_name=clone_name) if power_on else self.get_vm_by_name(name=clone_name)
+
+    def delete_vm(self, vm_name: str) -> None:
+        LOGGER.info(f"Deleting VM '{vm_name}'")
+        vm = self.get_vm_by_name(name=vm_name)
+        try:
+            if vm.runtime.powerState != vim.VirtualMachine.PowerState.poweredOff:
+                LOGGER.info(f"Powering off VM '{vm_name}' before deletion")
+                self.power_off_vm(vm_name=vm_name)
+            WaitForTask(task=vm.Destroy_Task(), maxWaitTime=120)
+            LOGGER.info(f"Successfully deleted VM '{vm_name}'")
+        except Exception:
+            LOGGER.error(f"Couldn't delete VM '{vm_name}'. Please, ensure it was deleted or delete it manually.")
+            raise
+
     def get_vm_by_name(self, name: str) -> vim.VirtualMachine:
         for dc in self.content.rootFolder.childEntity:
             for vm in dc.vmFolder.childEntity: