Add tests for data sources with instance types

Signed-off-by: Roni Kishner<rkishner@redhat.com>

tests/global_config.py

@@ -173,6 +173,15 @@ def _get_default_storage_class(sc_list):
     {"rhel9": {"template_os": "rhel9.0"}},
 ]
 
+data_import_cron_matrix = [
+    {"centos-stream9": {"instance_type": "u1.medium", "preference": "centos.stream9"}},
+    {"centos-stream10": {"instance_type": "u1.medium", "preference": "centos.stream10"}},
+    {"fedora": {"instance_type": "u1.medium", "preference": "fedora"}},
+    {"rhel8": {"instance_type": "u1.medium", "preference": "rhel.8"}},
+    {"rhel9": {"instance_type": "u1.medium", "preference": "rhel.9"}},
+    {"rhel10-beta": {"instance_type": "u1.medium", "preference": "rhel.10"}},
+]
+
 IMAGE_NAME_STR = "image_name"
 IMAGE_PATH_STR = "image_path"
 DV_SIZE_STR = "dv_size"

tests/infrastructure/golden_images/update_boot_source/test_boot_sources_vm.py

@@ -0,0 +1,68 @@
+import logging
+
+import pytest
+from ocp_resources.data_source import DataSource
+
+from tests.infrastructure.golden_images.utils import assert_os_version_mismatch_in_vm
+from utilities.constants import TIMEOUT_5SEC
+from utilities.infra import validate_os_info_vmi_vs_linux_os
+from utilities.storage import data_volume_template_with_source_ref_dict
+from utilities.virt import VirtualMachineForTests, running_vm
+
+LOGGER = logging.getLogger(__name__)
+
+
+@pytest.fixture()
+def boot_source_preference_from_data_source_dict(
+    request, data_source_from_data_import_cron, data_import_cron_matrix__function__
+):
+    preference_name = data_import_cron_matrix__function__[[*data_import_cron_matrix__function__][0]]["preference"]
+    if "fedora" in data_source_from_data_import_cron.name:
+        preference_name = f"fedora{request.getfixturevalue('latest_fedora_release_version')}"
+    return preference_name
+
+
+@pytest.fixture()
+def data_source_from_data_import_cron(
+    golden_images_namespace,
+    data_import_cron_matrix__function__,
+):
+    data_source = DataSource(name=[*data_import_cron_matrix__function__][0], namespace=golden_images_namespace.name)
+    data_source.wait_for_condition(
+        condition=data_source.Condition.READY, status=data_source.Condition.Status.TRUE, timeout=TIMEOUT_5SEC
+    )
+    return data_source
+
+
+@pytest.fixture()
+def auto_update_boot_source_instance_type_vm(
+    unprivileged_client,
+    namespace,
+    data_source_from_data_import_cron,
+):
+    LOGGER.info(f"Create a VM using {data_source_from_data_import_cron.name} dataSource")
+    with VirtualMachineForTests(
+        client=unprivileged_client,
+        name=f"{data_source_from_data_import_cron.name}-data-source-vm",
+        namespace=namespace.name,
+        vm_instance_type_infer=True,
+        vm_preference_infer=True,
+        data_volume_template=data_volume_template_with_source_ref_dict(
+            data_source=data_source_from_data_import_cron,
+        ),
+    ) as vm:
+        running_vm(vm=vm)
+        yield vm
+
+
+@pytest.mark.polarion("CNV-11774")
+def test_instance_type_vm_from_auto_update_boot_source(
+    auto_update_boot_source_instance_type_vm,
+    boot_source_preference_from_data_source_dict,
+):
+    LOGGER.info(f"Verify {auto_update_boot_source_instance_type_vm.name} OS version and virtctl info")
+    assert_os_version_mismatch_in_vm(
+        vm=auto_update_boot_source_instance_type_vm,
+        expected_os=boot_source_preference_from_data_source_dict,
+    )
+    validate_os_info_vmi_vs_linux_os(vm=auto_update_boot_source_instance_type_vm)

tests/infrastructure/golden_images/update_boot_source/test_ssp_common_templates_boot_sources.py

@@ -1,5 +1,4 @@
 import logging
-import re
 
 import pytest
 from ocp_resources.data_source import DataSource
@@ -13,8 +12,9 @@
 )
 from tests.infrastructure.golden_images.utils import (
     assert_missing_golden_image_pvc,
+    assert_os_version_mismatch_in_vm,
 )
-from utilities.constants import OS_FLAVOR_RHEL, TIMEOUT_5MIN, TIMEOUT_5SEC, Images
+from utilities.constants import TIMEOUT_5MIN, TIMEOUT_5SEC, Images
 from utilities.infra import (
     cleanup_artifactory_secret_and_config_map,
     get_artifactory_config_map,
@@ -27,16 +27,6 @@
 RHEL9_NAME = "rhel9"
 
 
-def assert_os_version_mismatch_in_vm(vm, expected_os):
-    expected_os_params = re.match(r"(?P<os_name>[a-z]+)(-stream)?(?P<os_ver>[0-9]+)", expected_os).groupdict()
-    vm_os = vm.ssh_exec.os.release_str.lower()
-    os_name = "redhat" if expected_os_params["os_name"] == OS_FLAVOR_RHEL else vm.os_flavor
-    expected_name_in_vm_os = "red hat" if expected_os_params["os_name"] == OS_FLAVOR_RHEL else os_name
-    assert re.match(rf"({expected_name_in_vm_os}).*({expected_os_params['os_ver']}).*", vm_os), (
-        f"Wrong VM OS, expected: {expected_os_params}, actual: {vm_os}"
-    )
-
-
 @pytest.fixture()
 def boot_source_os_from_data_source_dict(auto_update_data_source_matrix__function__):
     return auto_update_data_source_matrix__function__[[*auto_update_data_source_matrix__function__][0]]["template_os"]

tests/infrastructure/golden_images/utils.py

@@ -1,8 +1,11 @@
 import logging
+import re
 
+import pytest
 from timeout_sampler import TimeoutExpiredError, TimeoutSampler
 
-from utilities.constants import TIMEOUT_2MIN, TIMEOUT_5SEC
+from utilities.constants import OS_FLAVOR_RHEL, TIMEOUT_2MIN, TIMEOUT_5SEC
+from utilities.virt import VirtualMachineForTests
 
 LOGGER = logging.getLogger(__name__)
 
@@ -32,3 +35,17 @@ def _verify_missing_pvc_in_vm_conditions(_vm, _expected_message):
             f"conditions: {vm.instance.status.conditions}"
         )
         raise
+
+
+def assert_os_version_mismatch_in_vm(vm: VirtualMachineForTests, expected_os: str) -> None:
+    vm_os = vm.ssh_exec.os.release_str.lower()
+    match = re.match(r"(?P<os_name>[a-z]+)(?:[.-]stream|\.)?(?P<os_ver>[0-9]+)?", expected_os)
+    if not match:
+        pytest.fail(f"Did not find matching os_name and os_ver for {expected_os}")
+    expected_os_name = match.group("os_name")
+    expected_os_ver = match.group("os_ver")
+    if expected_os_name == OS_FLAVOR_RHEL:
+        expected_os_name = "red hat"
+    assert re.match(rf"({expected_os_name}).*({expected_os_ver}).*", vm_os), (
+        f"Wrong VM OS, expected name: {expected_os_name}, ver: {expected_os_ver}, actual: {vm_os}"
+    )

utilities/infra.py

@@ -45,6 +45,7 @@
 from ocp_resources.resource import Resource, ResourceEditor, get_client
 from ocp_resources.secret import Secret
 from ocp_resources.subscription import Subscription
+from ocp_resources.virtual_machine import VirtualMachine
 from ocp_utilities.exceptions import NodeNotReadyError, NodeUnschedulableError
 from ocp_utilities.infra import (
     assert_nodes_in_healthy_condition,
@@ -1516,11 +1517,12 @@ def get_linux_os_info(ssh_exec):
     }
 
 
-def validate_os_info_vmi_vs_linux_os(vm):
+def validate_os_info_vmi_vs_linux_os(vm: VirtualMachine) -> None:
     vmi_info = utilities.virt.get_guest_os_info(vmi=vm.vmi)
     linux_info = get_linux_os_info(ssh_exec=vm.ssh_exec)["os"]
 
     assert vmi_info == linux_info, f"Data mismatch! VMI: {vmi_info}\nOS: {linux_info}"
+    return None
 
 
 def get_nodes_cpu_architecture(nodes: list[Node]) -> str: