[CCLM] Verify VM boot_id, add post-CCLM validation tests (#4108)

##### Short description:
- Check VM boot_id before and after CCLM
- Allow connection to the VM Console in the remote cluster with the
kubeconfig
- Add post-CCLM validation tests

Jira: https://issues.redhat.com/browse/CNV-60016

##### More details:
Split from
#3926

##### What this PR does / why we need it:

##### Which issue(s) this PR fixes:

##### Special notes for reviewer:
Assisted by Claude AI and Cursor AI

##### jira-ticket:
<!-- full-ticket-url needs to be provided. This would add a link to the
pull request to the jira and close it when the pull request is merged
If the task is not tracked by a Jira ticket, just write "NONE".
-->


<!-- This is an auto-generated comment: release notes by coderabbit.ai
-->
## Summary by CodeRabbit

* **Tests**
* Expanded cross-cluster VM migration test coverage with additional VM
scenarios and state verification.
* Added post-migration validation for VM boot integrity and file
persistence.
  * New tests for VM state transitions and cleanup operations.

* **Refactor**
* Consolidated test constants for broader reusability across test
modules.
* Refactored VM file verification utilities for improved test
modularity.
  * Enhanced console utility support for remote cluster operations.
<!-- end of auto-generated comment: release notes by coderabbit.ai -->

Author: jpeimer

tests/storage/constants.py

@@ -18,3 +18,6 @@
 HTTPS = "https"
 
 QUAY_FEDORA_CONTAINER_IMAGE = f"docker://{Images.Fedora.FEDORA_CONTAINER_IMAGE}"
+
+TEST_FILE_NAME = "test-file.txt"
+TEST_FILE_CONTENT = "test-content"

tests/storage/cross_cluster_live_migration/conftest.py

@@ -1,9 +1,12 @@
 import logging
+import os
 import re
+import tempfile
 from copy import deepcopy
 
 import pytest
 import requests
+import yaml
 from kubernetes.dynamic.exceptions import NotFoundError
 from ocp_resources.data_source import DataSource
 from ocp_resources.forklift_controller import ForkliftController
@@ -25,8 +28,10 @@
 )
 from pytest_testconfig import config as py_config
 
+from tests.storage.constants import TEST_FILE_CONTENT, TEST_FILE_NAME
 from tests.storage.cross_cluster_live_migration.utils import (
     enable_feature_gate_and_configure_hco_live_migration_network,
+    get_vm_boot_id_via_console,
 )
 from utilities.constants import (
     OS_FLAVOR_RHEL,
@@ -38,7 +43,7 @@
     Images,
 )
 from utilities.infra import create_ns, get_hyperconverged_resource
-from utilities.storage import data_volume_template_with_source_ref_dict
+from utilities.storage import data_volume_template_with_source_ref_dict, write_file
 from utilities.virt import VirtualMachineForTests, running_vm
 
 LOGGER = logging.getLogger(__name__)
@@ -109,6 +114,50 @@ def remote_cluster_auth_token(remote_admin_client):
     raise NotFoundError("Unable to extract authentication token from remote admin client")
 
 
+@pytest.fixture(scope="session")
+def remote_cluster_kubeconfig(remote_admin_client, remote_cluster_auth_token):
+    """
+    Generate a kubeconfig file from the remote admin client credentials.
+    Returns the path to the generated kubeconfig file.
+    """
+    # Extract cluster information from the client
+    cluster_host = remote_admin_client.configuration.host
+    cluster_name = "remote-cluster"
+    user_name = "remote-admin"
+    context_name = "remote-context"
+
+    # Create kubeconfig structure
+    kubeconfig_dict = {
+        "apiVersion": "v1",
+        "kind": "Config",
+        "clusters": [
+            {
+                "name": cluster_name,
+                "cluster": {
+                    "server": cluster_host,
+                    "insecure-skip-tls-verify": True,
+                },
+            }
+        ],
+        "users": [{"name": user_name, "user": {"token": remote_cluster_auth_token}}],
+        "contexts": [{"name": context_name, "context": {"cluster": cluster_name, "user": user_name}}],
+        "current-context": context_name,
+    }
+
+    # Use TemporaryDirectory context manager for automatic cleanup
+    with tempfile.TemporaryDirectory(suffix="-remote-kubeconfig") as temp_dir:
+        kubeconfig_path = os.path.join(temp_dir, "kubeconfig")
+
+        # Write kubeconfig file with secure permissions (0o600 = rw-------)
+        # Using os.open with O_CREAT ensures the file is created with restricted permissions from the start
+        file_descriptor = os.open(kubeconfig_path, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600)
+        with os.fdopen(file_descriptor, "w") as kubeconfig_file:
+            yaml.safe_dump(data=kubeconfig_dict, stream=kubeconfig_file)
+
+        LOGGER.info(f"Created remote cluster kubeconfig at: {kubeconfig_path}")
+        yield kubeconfig_path
+
+
 @pytest.fixture(scope="session")
 def remote_cluster_hco_namespace(remote_admin_client):
     return Namespace(client=remote_admin_client, name=py_config["hco_namespace"], ensure_exists=True)
@@ -402,7 +451,7 @@ def vm_for_cclm_from_template_with_data_source(
         ),
         memory_guest=Images.Rhel.DEFAULT_MEMORY_SIZE,
     ) as vm:
-        running_vm(vm=vm, check_ssh_connectivity=False)  # False because we can't ssh to a VM in the remote cluster
+        vm.start()
         yield vm
 
 
@@ -422,7 +471,7 @@ def vm_for_cclm_with_instance_type(
             storage_class=py_config["default_storage_class"],
         ),
     ) as vm:
-        running_vm(vm=vm, check_ssh_connectivity=False)  # False because we can't ssh to a VM in the remote cluster
+        vm.start()
         yield vm
 
 
@@ -437,6 +486,53 @@ def vms_for_cclm(request):
     yield vms
 
 
+@pytest.fixture(scope="class")
+def booted_vms_for_cclm(vms_for_cclm):
+    for vm in vms_for_cclm:
+        running_vm(vm=vm, check_ssh_connectivity=False)  # False because we can't ssh to a VM in the remote cluster
+    return vms_for_cclm
+
+
+@pytest.fixture(scope="class")
+def vms_boot_id_before_cclm(booted_vms_for_cclm, remote_cluster_kubeconfig):
+    return {
+        vm.name: get_vm_boot_id_via_console(vm=vm, kubeconfig=remote_cluster_kubeconfig) for vm in booted_vms_for_cclm
+    }
+
+
+@pytest.fixture(scope="class")
+def written_file_to_vms_before_cclm(booted_vms_for_cclm, remote_cluster_kubeconfig):
+    for vm in booted_vms_for_cclm:
+        write_file(
+            vm=vm,
+            filename=TEST_FILE_NAME,
+            content=TEST_FILE_CONTENT,
+            stop_vm=False,
+            kubeconfig=remote_cluster_kubeconfig,
+        )
+    return booted_vms_for_cclm
+
+
+@pytest.fixture(scope="class")
+def local_vms_after_cclm_migration(admin_client, namespace, vms_for_cclm):
+    """
+    Returns List of VirtualMachineForTests objects referencing VMs in the local cluster
+    """
+    local_vms = []
+    for vm in vms_for_cclm:
+        local_vm = VirtualMachineForTests(
+            name=vm.name,
+            namespace=namespace.name,
+            os_flavor=vm.os_flavor,
+            client=admin_client,
+            generate_unique_name=False,
+        )
+        local_vm.username = vm.username
+        local_vm.password = vm.password
+        local_vms.append(local_vm)
+    return local_vms
+
+
 @pytest.fixture(scope="class")
 def mtv_migration_plan(
     admin_client,

tests/storage/cross_cluster_live_migration/test_cclm.py

@@ -1,10 +1,16 @@
 import pytest
 
-from tests.storage.cross_cluster_live_migration.utils import verify_compute_live_migration_after_cclm
+from tests.storage.constants import TEST_FILE_CONTENT, TEST_FILE_NAME
+from tests.storage.cross_cluster_live_migration.utils import (
+    assert_vms_are_stopped,
+    assert_vms_can_be_deleted,
+    verify_compute_live_migration_after_cclm,
+    verify_vms_boot_id_after_cross_cluster_live_migration,
+)
+from tests.storage.utils import check_file_in_vm
 from utilities.constants import TIMEOUT_10MIN
 
-TESTS_CLASS_NAME_VM_FROM_TEMPLATE_WITH_DATA_SOURCE = "CCLMvmFromTemplateWithDataSource"
-TESTS_CLASS_NAME_VM_WITH_INSTANCE_TYPE = "CCLMvmWithInstanceType"
+TESTS_CLASS_NAME_SEVERAL_VMS = "TestCCLMSeveralVMs"
 
 pytestmark = [
     pytest.mark.cclm,
@@ -21,18 +27,23 @@
     "vms_for_cclm",
     [
         pytest.param(
-            {"vms_fixtures": ["vm_for_cclm_from_template_with_data_source"]},
+            {
+                "vms_fixtures": [
+                    "vm_for_cclm_from_template_with_data_source",
+                    "vm_for_cclm_with_instance_type",
+                ]
+            },
         )
     ],
     indirect=True,
 )
-class TestCCLMvmFromTemplateWithDataSource:
-    @pytest.mark.polarion("CNV-11910")
-    @pytest.mark.dependency(
-        name=f"{TESTS_CLASS_NAME_VM_FROM_TEMPLATE_WITH_DATA_SOURCE}::test_migrate_vm_from_remote_to_local_cluster"
-    )
+class TestCCLMSeveralVMs:
+    @pytest.mark.polarion("CNV-11995")
+    @pytest.mark.dependency(name=f"{TESTS_CLASS_NAME_SEVERAL_VMS}::test_migrate_vm_from_remote_to_local_cluster")
     def test_migrate_vm_from_remote_to_local_cluster(
         self,
+        written_file_to_vms_before_cclm,
+        vms_boot_id_before_cclm,
         mtv_migration,
     ):
         mtv_migration.wait_for_condition(
@@ -42,42 +53,35 @@ def test_migrate_vm_from_remote_to_local_cluster(
             stop_condition=mtv_migration.Status.FAILED,
         )
 
-    @pytest.mark.dependency(
-        depends=[f"{TESTS_CLASS_NAME_VM_FROM_TEMPLATE_WITH_DATA_SOURCE}::test_migrate_vm_from_remote_to_local_cluster"]
-    )
-    @pytest.mark.polarion("CNV-12038")
-    def test_compute_live_migrate_vms_after_cclm(self, admin_client, namespace, vms_for_cclm):
-        verify_compute_live_migration_after_cclm(client=admin_client, namespace=namespace, vms_list=vms_for_cclm)
+    @pytest.mark.dependency(depends=[f"{TESTS_CLASS_NAME_SEVERAL_VMS}::test_migrate_vm_from_remote_to_local_cluster"])
+    @pytest.mark.polarion("CNV-11910")
+    def test_verify_vms_not_rebooted_after_migration(self, local_vms_after_cclm_migration, vms_boot_id_before_cclm):
+        verify_vms_boot_id_after_cross_cluster_live_migration(
+            local_vms=local_vms_after_cclm_migration, initial_boot_id=vms_boot_id_before_cclm
+        )
 
+    @pytest.mark.dependency(depends=[f"{TESTS_CLASS_NAME_SEVERAL_VMS}::test_migrate_vm_from_remote_to_local_cluster"])
+    @pytest.mark.polarion("CNV-14332")
+    def test_verify_file_persisted_after_migration(self, local_vms_after_cclm_migration):
+        for vm in local_vms_after_cclm_migration:
+            check_file_in_vm(
+                vm=vm,
+                file_name=TEST_FILE_NAME,
+                file_content=TEST_FILE_CONTENT,
+                username=vm.username,
+                password=vm.password,
+            )
 
-@pytest.mark.parametrize(
-    "vms_for_cclm",
-    [
-        pytest.param(
-            {"vms_fixtures": ["vm_for_cclm_with_instance_type"]},
-        ),
-    ],
-    indirect=True,
-)
-class TestCCLMvmWithInstanceType:
-    @pytest.mark.polarion("CNV-12013")
-    @pytest.mark.dependency(
-        name=f"{TESTS_CLASS_NAME_VM_WITH_INSTANCE_TYPE}::test_migrate_vm_from_remote_to_local_cluster"
-    )
-    def test_migrate_vm_from_remote_to_local_cluster(
-        self,
-        mtv_migration,
-    ):
-        mtv_migration.wait_for_condition(
-            condition=mtv_migration.Condition.Type.SUCCEEDED,
-            status=mtv_migration.Condition.Status.TRUE,
-            timeout=TIMEOUT_10MIN,
-            stop_condition=mtv_migration.Status.FAILED,
-        )
+    @pytest.mark.dependency(depends=[f"{TESTS_CLASS_NAME_SEVERAL_VMS}::test_migrate_vm_from_remote_to_local_cluster"])
+    @pytest.mark.polarion("CNV-14333")
+    def test_source_vms_are_stopped_after_cclm(self, vms_for_cclm):
+        assert_vms_are_stopped(vms=vms_for_cclm)
+
+    @pytest.mark.dependency(depends=[f"{TESTS_CLASS_NAME_SEVERAL_VMS}::test_migrate_vm_from_remote_to_local_cluster"])
+    @pytest.mark.polarion("CNV-12038")
+    def test_compute_live_migrate_vms_after_cclm(self, local_vms_after_cclm_migration):
+        verify_compute_live_migration_after_cclm(local_vms=local_vms_after_cclm_migration)
 
-    @pytest.mark.dependency(
-        depends=[f"{TESTS_CLASS_NAME_VM_WITH_INSTANCE_TYPE}::test_migrate_vm_from_remote_to_local_cluster"]
-    )
-    @pytest.mark.polarion("CNV-12474")
-    def test_compute_live_migrate_vms_after_cclm(self, admin_client, namespace, vms_for_cclm):
-        verify_compute_live_migration_after_cclm(client=admin_client, namespace=namespace, vms_list=vms_for_cclm)
+    @pytest.mark.polarion("CNV-14334")
+    def test_source_vms_can_be_deleted(self, vms_for_cclm):
+        assert_vms_can_be_deleted(vms=vms_for_cclm)

tests/storage/cross_cluster_live_migration/utils.py

@@ -7,6 +7,7 @@
 from ocp_resources.namespace import Namespace
 from ocp_resources.network_attachment_definition import NetworkAttachmentDefinition
 
+from utilities import console
 from utilities.constants import VIRT_HANDLER
 from utilities.hco import ResourceEditorValidateHCOReconcile
 from utilities.infra import get_daemonset_by_name
@@ -73,32 +74,74 @@ def enable_feature_gate_and_configure_hco_live_migration_network(
         )
 
 
-def verify_compute_live_migration_after_cclm(
-    client: DynamicClient, namespace: Namespace, vms_list: list[VirtualMachineForTests]
-) -> None:
+def verify_compute_live_migration_after_cclm(local_vms: list[VirtualMachineForTests]) -> None:
     """
     Verify compute live migration for VMs after Cross-Cluster Live Migration (CCLM).
 
-    This function creates local VM references for each VM that was migrated from the remote cluster,
-    preserves their credentials, and attempts to perform compute live migration on each VM.
-
     Args:
-        client: DynamicClient
-        namespace: The namespace where the VMs are located in the target cluster
-        vms_list: List of VirtualMachineForTests objects to be migrated
+        local_vms: List of VirtualMachineForTests objects in the local cluster
 
     Raises:
         AssertionError: If any VM migration fails, with details of all failed migrations
     """
     vms_failed_migration = {}
-    for vm in vms_list:
-        local_vm = VirtualMachineForTests(
-            name=vm.name, namespace=namespace.name, client=client, generate_unique_name=False
-        )
-        local_vm.username = vm.username
-        local_vm.password = vm.password
+    for vm in local_vms:
         try:
-            migrate_vm_and_verify(vm=local_vm, check_ssh_connectivity=True)
+            migrate_vm_and_verify(vm=vm, check_ssh_connectivity=True)
         except Exception as migration_exception:
-            vms_failed_migration[local_vm.name] = migration_exception
+            vms_failed_migration[vm.name] = migration_exception
     assert not vms_failed_migration, f"Failed VM migrations: {vms_failed_migration}"
+
+
+def get_vm_boot_id_via_console(
+    vm: VirtualMachineForTests, kubeconfig: str | None = None, username: str | None = None, password: str | None = None
+) -> str:
+    with console.Console(vm=vm, kubeconfig=kubeconfig, username=username, password=password) as vm_console:
+        vm_console.sendline("cat /proc/sys/kernel/random/boot_id")
+        vm_console.expect([r"#", r"\$"])
+        raw_output = vm_console.before
+        LOGGER.info(f"Boot ID from VM console (raw output): '{raw_output}'")
+        return raw_output
+
+
+def verify_vms_boot_id_after_cross_cluster_live_migration(
+    local_vms: list[VirtualMachineForTests],
+    initial_boot_id: dict[str, str],
+) -> None:
+    """
+    Verify that VMs have not rebooted after cross-cluster live migration.
+
+    Args:
+        local_vms: List of VirtualMachineForTests objects in the local cluster
+        initial_boot_id: Dictionary mapping VM names to their initial boot IDs
+
+    Raises:
+        AssertionError: If any VM has rebooted (boot ID changed)
+    """
+    rebooted_vms = {}
+    for vm in local_vms:
+        current_boot_id = get_vm_boot_id_via_console(vm=vm, username=vm.username, password=vm.password)
+        if initial_boot_id[vm.name] != current_boot_id:
+            rebooted_vms[vm.name] = {"initial": initial_boot_id[vm.name], "current": current_boot_id}
+    assert not rebooted_vms, f"Boot id changed for VMs:\n {rebooted_vms}"
+
+
+def assert_vms_are_stopped(vms: list[VirtualMachineForTests]) -> None:
+    not_stopped_vms = {}
+    for vm in vms:
+        vm_status = vm.printable_status
+        if vm_status != vm.Status.STOPPED:
+            not_stopped_vms[vm.name] = vm_status
+    assert not not_stopped_vms, f"Source VMs are not stopped: {not_stopped_vms}"
+
+
+def assert_vms_can_be_deleted(vms: list[VirtualMachineForTests]) -> None:
+    vms_failed_cleanup = {}
+    for vm in vms:
+        try:
+            cleanup_result = vm.clean_up()
+            if cleanup_result is not True:
+                vms_failed_cleanup[vm.name] = f"vm.clean_up() returned {cleanup_result}"
+        except Exception as cleanup_exception:
+            vms_failed_cleanup[vm.name] = str(cleanup_exception)
+    assert not vms_failed_cleanup, f"Failed to clean up source VMs: {vms_failed_cleanup}"