From e0ea7541f695679b259faba4ce22d440a47e4fb8 Mon Sep 17 00:00:00 2001 From: Emanuele Prella Date: Mon, 4 May 2026 15:52:03 +0200 Subject: [PATCH 1/4] [4.20] [Storage] CherryPicked: Move tests away from test_import_http.py and replace using datasource cloning (#4646) Manual backport of #4455 --------- Signed-off-by: Emanuele Prella --- tests/conftest.py | 5 - tests/storage/cdi_import/conftest.py | 74 ++-- tests/storage/cdi_import/test_import_http.py | 177 +-------- .../cdi_import/test_import_registry.py | 18 + tests/storage/cdi_import/utils.py | 102 ++++++ tests/storage/general/__init__.py | 0 tests/storage/general/conftest.py | 103 ++++++ .../storage/general/test_storage_behavior.py | 116 ++++++ utilities/pytest_matrix_utils.py | 35 ++ .../unittests/test_pytest_matrix_utils.py | 340 ++++++++++++++++++ 10 files changed, 748 insertions(+), 222 deletions(-) create mode 100644 tests/storage/cdi_import/utils.py create mode 100644 tests/storage/general/__init__.py create mode 100644 tests/storage/general/conftest.py create mode 100644 tests/storage/general/test_storage_behavior.py create mode 100644 utilities/unittests/test_pytest_matrix_utils.py diff --git a/tests/conftest.py b/tests/conftest.py index d47694c796..6e0afb826a 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -1113,11 +1113,6 @@ def _skip_access_mode_rwo(storage_class_matrix): pytest.skip(reason="Skipping when access_mode is RWO; possible reason: cannot migrate VMI with non-shared PVCs") -@pytest.fixture() -def skip_access_mode_rwo_scope_function(storage_class_matrix__function__): - _skip_access_mode_rwo(storage_class_matrix=storage_class_matrix__function__) - - @pytest.fixture(scope="class") def skip_access_mode_rwo_scope_class(storage_class_matrix__class__): _skip_access_mode_rwo(storage_class_matrix=storage_class_matrix__class__) diff --git a/tests/storage/cdi_import/conftest.py b/tests/storage/cdi_import/conftest.py index 5105d4b2de..591711123f 100644 --- a/tests/storage/cdi_import/conftest.py +++ b/tests/storage/cdi_import/conftest.py @@ -7,15 +7,14 @@ import pytest from ocp_resources.datavolume import DataVolume -from ocp_resources.persistent_volume_claim import PersistentVolumeClaim +from pytest_testconfig import py_config +from tests.storage.cdi_import.utils import wait_dv_and_get_importer, wait_for_multus_network_status from tests.storage.constants import ( - HPP_STORAGE_CLASSES, HTTP, QUAY_FEDORA_CONTAINER_IMAGE, ) from tests.storage.utils import ( - create_cirros_dv, create_pod_for_pvc, get_file_url, ) @@ -37,13 +36,8 @@ @pytest.fixture() -def skip_non_shared_storage(storage_class_name_scope_function): - if storage_class_name_scope_function in HPP_STORAGE_CLASSES: - pytest.skip("Skipping when storage is non-shared") - - -@pytest.fixture() -def bridge_on_node(): +def bridge_on_node(admin_client): + """Create a Linux Bridge network device and yield it.""" with network_device( interface_type=LINUX_BRIDGE, nncp_name=BRIDGE_NAME, @@ -53,7 +47,8 @@ def bridge_on_node(): @pytest.fixture() -def linux_nad(namespace, bridge_on_node): +def linux_nad(admin_client, namespace, bridge_on_node): + """Create a Linux Bridge Network Attachment Definition (NAD) and yield it.""" with network_nad( namespace=namespace, nad_type=LINUX_BRIDGE, @@ -63,25 +58,9 @@ def linux_nad(namespace, bridge_on_node): yield nad -@pytest.fixture() -def cirros_pvc( - data_volume_template_metadata, -): - return PersistentVolumeClaim( - name=data_volume_template_metadata["name"], - namespace=data_volume_template_metadata["namespace"], - ) - - -@pytest.fixture() -def pvc_original_timestamp( - cirros_pvc, -): - return cirros_pvc.instance.metadata.creationTimestamp - - @pytest.fixture() def dv_non_exist_url(namespace, storage_class_name_scope_module): + """Create a DV with a non-existent URL""" with create_dv( dv_name=f"cnv-876-{storage_class_name_scope_module}", namespace=namespace.name, @@ -99,6 +78,7 @@ def dv_from_http_import( storage_class_name_scope_module, images_internal_http_server, ): + """Create a DV from HTTP import with parameters from the test function.""" with create_dv( dv_name=f"{request.param.get('dv_name', 'http-dv')}-{storage_class_name_scope_module}", namespace=namespace.name, @@ -121,6 +101,7 @@ def running_pod_with_dv_pvc( storage_class_name_scope_module, dv_from_http_import, ): + """Create a running pod with DV's PVC.""" dv_from_http_import.wait_for_dv_success() with create_pod_for_pvc( pvc=dv_from_http_import.pvc, @@ -129,23 +110,9 @@ def running_pod_with_dv_pvc( yield pod -@pytest.fixture(scope="module") -def cirros_dv_unprivileged( - namespace, - storage_class_name_scope_module, - unprivileged_client, -): - yield from create_cirros_dv( - namespace=namespace.name, - name=f"cirros-dv-{storage_class_name_scope_module}", - storage_class=storage_class_name_scope_module, - client=unprivileged_client, - dv_size=DEFAULT_DV_SIZE, - ) - - @pytest.fixture() def created_blank_dv_list(unprivileged_client, namespace, storage_class_name_scope_module, number_of_dvs): + """Create a list of blank DVs.""" dvs_list = [] try: for dv_index in range(number_of_dvs): @@ -197,7 +164,8 @@ def created_vm_list(unprivileged_client, created_blank_dv_list, storage_class_na @pytest.fixture() -def dvs_and_vms_from_public_registry(namespace, storage_class_name_scope_function): +def dvs_and_vms_from_public_registry(unprivileged_client, namespace, storage_class_name_scope_function): + """Create DVs from public registry and VMs from those DVs.""" dvs = [] vms = [] try: @@ -233,3 +201,21 @@ def dvs_and_vms_from_public_registry(namespace, storage_class_name_scope_functio vm.clean_up() for dv in dvs: dv.clean_up() + + +@pytest.fixture() +def importer_pod_annotations(admin_client, namespace, linux_nad): + """Create a DV with multus annotation and yield the importer pod annotations.""" + with create_dv( + dv_name="dv-annotation", + namespace=namespace.name, + source=REGISTRY_STR, + url=QUAY_FEDORA_CONTAINER_IMAGE, + size=Images.Fedora.DEFAULT_DV_SIZE, + storage_class=py_config["default_storage_class"], + multus_annotation=linux_nad.name, + client=namespace.client, + ) as dv: + importer_pod = wait_dv_and_get_importer(dv=dv, admin_client=admin_client) + wait_for_multus_network_status(importer_pod=importer_pod) + yield importer_pod.instance.metadata.annotations diff --git a/tests/storage/cdi_import/test_import_http.py b/tests/storage/cdi_import/test_import_http.py index 4a4ba13f44..25da7e09a7 100644 --- a/tests/storage/cdi_import/test_import_http.py +++ b/tests/storage/cdi_import/test_import_http.py @@ -7,11 +7,12 @@ import pytest from kubernetes.dynamic.exceptions import UnprocessibleEntityError from ocp_resources.datavolume import DataVolume -from ocp_resources.resource import Resource from pytest_testconfig import config as py_config from timeout_sampler import TimeoutExpiredError, TimeoutSampler -from tests.os_params import FEDORA_LATEST, RHEL_LATEST +from tests.storage.cdi_import.utils import ( + wait_dv_and_get_importer, +) from tests.storage.constants import ( CIRROS_QCOW2_IMG, HTTP, @@ -22,30 +23,19 @@ from tests.storage.utils import ( assert_num_files_in_pod, assert_use_populator, - create_vm_from_dv, get_file_url, - get_importer_pod, wait_for_importer_container_message, ) -from utilities import console from utilities.constants import ( - OS_FLAVOR_RHEL, QUARANTINED, TIMEOUT_1MIN, TIMEOUT_5MIN, - TIMEOUT_5SEC, - TIMEOUT_12MIN, - TIMEOUT_20SEC, Images, ) -from utilities.infra import get_node_selector_dict from utilities.ssp import validate_os_info_vmi_vs_windows_os from utilities.storage import ( ErrorMsg, - create_dummy_first_consumer_pod, create_dv, - get_test_artifact_server_url, - sc_volume_binding_mode_is_wffc, ) from utilities.virt import running_vm @@ -59,78 +49,7 @@ TAR_IMG = "archive.tar" DEFAULT_DV_SIZE = Images.Cirros.DEFAULT_DV_SIZE SMALL_DV_SIZE = "200Mi" - - -def get_importer_pod_node(importer_pod): - for sample in TimeoutSampler( - wait_timeout=TIMEOUT_1MIN, - sleep=TIMEOUT_5SEC, - func=lambda: importer_pod.instance.get("spec", {}).get( - "nodeName", - ), - ): - if sample: - return sample - - -def wait_for_pvc_recreate(pvc, pvc_original_timestamp): - for sample in TimeoutSampler( - wait_timeout=TIMEOUT_20SEC, - sleep=1, - func=lambda: pvc.instance.metadata.creationTimestamp != pvc_original_timestamp, - ): - if sample: - break - - -def wait_dv_and_get_importer(dv, admin_client): - dv.wait_for_status( - status=DataVolume.Status.IMPORT_IN_PROGRESS, - timeout=TIMEOUT_1MIN, - stop_status=DataVolume.Status.SUCCEEDED, - ) - return get_importer_pod(dyn_client=admin_client, namespace=dv.namespace) - - -@pytest.fixture() -def dv_with_annotation(admin_client, namespace, linux_nad): - with create_dv( - dv_name="dv-annotation", - namespace=namespace.name, - url=f"{get_test_artifact_server_url()}{FEDORA_LATEST['image_path']}", - storage_class=py_config["default_storage_class"], - multus_annotation=linux_nad.name, - ) as dv: - return wait_dv_and_get_importer(dv=dv, admin_client=admin_client).instance.metadata.annotations - - -@pytest.mark.sno -@pytest.mark.parametrize( - "data_volume_multi_storage_scope_function", - [ - pytest.param( - { - "dv_name": "import-http-dv", - "source": HTTP, - "image": CIRROS_QCOW2_IMG, - "dv_size": DEFAULT_DV_SIZE, - }, - marks=pytest.mark.polarion("CNV-675"), - ), - ], - indirect=True, -) -def test_delete_pvc_after_successful_import( - data_volume_multi_storage_scope_function, -): - pvc = data_volume_multi_storage_scope_function.pvc - pvc_original_timestamp = pvc.instance.metadata.creationTimestamp - pvc.delete() - wait_for_pvc_recreate(pvc=pvc, pvc_original_timestamp=pvc_original_timestamp) - storage_class = data_volume_multi_storage_scope_function.storage_class - if sc_volume_binding_mode_is_wffc(sc=storage_class): - create_dummy_first_consumer_pod(pvc=pvc) - data_volume_multi_storage_scope_function.wait_for_dv_success() +LATEST_WINDOWS_OS_DICT = py_config.get("latest_windows_os_dict", {}) @pytest.mark.sno @@ -451,78 +370,6 @@ def test_blank_disk_import_validate_status(data_volume_multi_storage_scope_funct data_volume_multi_storage_scope_function.wait_for_dv_success(timeout=TIMEOUT_5MIN) -@pytest.mark.parametrize( - "dv_from_http_import", - [ - pytest.param( - { - "dv_name": "cnv-3065", - "file_name": Images.Cdi.QCOW2_IMG, - "source": HTTPS, - "size": "100Mi", - "configmap_name": INTERNAL_HTTP_CONFIGMAP_NAME, - }, - marks=pytest.mark.polarion("CNV-3065"), - ), - ], - indirect=True, -) -@pytest.mark.sno -def test_disk_falloc(internal_http_configmap, dv_from_http_import): - dv_from_http_import.wait_for_dv_success() - with create_vm_from_dv(dv=dv_from_http_import) as vm_dv: - with console.Console(vm=vm_dv) as vm_console: - LOGGER.info("Fill disk space.") - vm_console.sendline("dd if=/dev/zero of=file bs=1M") - vm_console.expect("dd: writing 'file': No space left on device", timeout=TIMEOUT_1MIN) - - -@pytest.mark.destructive -@pytest.mark.parametrize( - "data_volume_multi_storage_scope_function", - [ - pytest.param( - { - "dv_name": "cnv-3362", - "source": HTTP, - "image": RHEL_LATEST["image_path"], - "dv_size": "25Gi", - "access_modes": DataVolume.AccessMode.RWX, - "wait": False, - }, - marks=pytest.mark.polarion("CNV-3632"), - ), - ], - indirect=True, -) -def test_vm_from_dv_on_different_node( - admin_client, - skip_access_mode_rwo_scope_function, - skip_non_shared_storage, - schedulable_nodes, - data_volume_multi_storage_scope_function, -): - """ - Test that create and run VM from DataVolume (only use RWX access mode) on different node. - It applies to shared storage like Ceph or NFS. It cannot be tested on local storage like HPP. - """ - importer_pod = get_importer_pod( - dyn_client=admin_client, - namespace=data_volume_multi_storage_scope_function.namespace, - ) - importer_node_name = get_importer_pod_node(importer_pod=importer_pod) - nodes = list(filter(lambda node: importer_node_name != node.name, schedulable_nodes)) - data_volume_multi_storage_scope_function.wait_for_dv_success(timeout=TIMEOUT_12MIN) - with create_vm_from_dv( - dv=data_volume_multi_storage_scope_function, - vm_name="rhel-vm", - os_flavor=OS_FLAVOR_RHEL, - node_selector=get_node_selector_dict(node_selector=nodes[0].name), - memory_guest=Images.Rhel.DEFAULT_MEMORY_SIZE, - ) as vm_dv: - assert vm_dv.vmi.node.name != importer_node_name - - @pytest.mark.tier3 @pytest.mark.parametrize( "data_volume_multi_storage_scope_function," @@ -557,19 +404,3 @@ def test_successful_vm_from_imported_dv_windows( validate_os_info_vmi_vs_windows_os( vm=vm_instance_from_template_multi_storage_scope_function, ) - - -@pytest.mark.polarion("CNV-4724") -@pytest.mark.sno -def test_dv_api_version_after_import(cirros_dv_unprivileged): - assert ( - cirros_dv_unprivileged.api_version - == f"{cirros_dv_unprivileged.api_group}/{cirros_dv_unprivileged.ApiVersion.V1BETA1}" - ) - - -@pytest.mark.polarion("CNV-5509") -def test_importer_pod_annotation(dv_with_annotation, linux_nad): - # verify "k8s.v1.cni.cncf.io/networks" can pass to the importer pod - assert dv_with_annotation.get(f"{Resource.ApiGroup.K8S_V1_CNI_CNCF_IO}/networks") == linux_nad.name - assert '"interface": "net1"' in dv_with_annotation.get(f"{Resource.ApiGroup.K8S_V1_CNI_CNCF_IO}/network-status") diff --git a/tests/storage/cdi_import/test_import_registry.py b/tests/storage/cdi_import/test_import_registry.py index f89b882eca..d0432d852c 100644 --- a/tests/storage/cdi_import/test_import_registry.py +++ b/tests/storage/cdi_import/test_import_registry.py @@ -1,8 +1,10 @@ +import json import logging import pytest from kubernetes.client.rest import ApiException from ocp_resources.datavolume import DataVolume +from ocp_resources.resource import Resource from tests.storage.constants import QUAY_FEDORA_CONTAINER_IMAGE from tests.storage.utils import ( @@ -171,3 +173,19 @@ def test_public_registry_data_volume_archive(namespace, storage_class_name_scope storage_class=[*storage_class_name_scope_function][0], ): return + + +@pytest.mark.polarion("CNV-5509") +def test_importer_pod_annotation(importer_pod_annotations, linux_nad): + """Verify "k8s.v1.cni.cncf.io/networks" can be passed to the importer pod""" + networks_annotation = f"{Resource.ApiGroup.K8S_V1_CNI_CNCF_IO}/networks" + network_status_annotation = f"{Resource.ApiGroup.K8S_V1_CNI_CNCF_IO}/network-status" + + networks_value = importer_pod_annotations.get(networks_annotation) + assert networks_value == linux_nad.name, ( + f"DV annotation is not passed to the importer pod. Expected: {linux_nad.name}, Found: {networks_value}" + ) + + network_status_value = importer_pod_annotations.get(network_status_annotation) + interfaces = [entry["interface"] for entry in json.loads(network_status_value) if "interface" in entry] + assert "net1" in interfaces, f"Expected interface: net1, Found: {interfaces}" diff --git a/tests/storage/cdi_import/utils.py b/tests/storage/cdi_import/utils.py new file mode 100644 index 0000000000..51593ac056 --- /dev/null +++ b/tests/storage/cdi_import/utils.py @@ -0,0 +1,102 @@ +"""Helper utilities for CDI import tests.""" + +from __future__ import annotations + +from typing import TYPE_CHECKING + +from ocp_resources.datavolume import DataVolume +from ocp_resources.resource import Resource +from timeout_sampler import TimeoutExpiredError, TimeoutSampler + +from tests.storage.utils import get_importer_pod +from utilities.constants import TIMEOUT_1MIN, TIMEOUT_5SEC, TIMEOUT_20SEC + +if TYPE_CHECKING: + from ocp_resources.persistent_volume_claim import PersistentVolumeClaim + from ocp_resources.pod import Pod + from ocp_resources.resource import DynamicClient + + +def get_importer_pod_node(importer_pod: Pod) -> str: + """Get the node name where the importer pod is scheduled. + + Args: + importer_pod: The importer pod resource. + + Returns: + str: The node name where the pod is scheduled. + + Raises: + TimeoutExpiredError: If the importer pod is not scheduled within the timeout period. + """ + for sample in TimeoutSampler( + wait_timeout=TIMEOUT_1MIN, + sleep=TIMEOUT_5SEC, + func=lambda: importer_pod.instance.spec.nodeName, + ): + if sample: + return sample + raise TimeoutExpiredError("Importer pod was not scheduled within the timeout period.") + + +def wait_for_pvc_recreate(pvc: PersistentVolumeClaim, pvc_creation_timestamp: str) -> None: + """Wait for PVC to be recreated with a new timestamp. + + Args: + pvc: The PVC resource to monitor. + pvc_creation_timestamp: The original creation timestamp to compare against. + + Raises: + TimeoutExpiredError: If the PVC is not recreated within the timeout period. + """ + for sample in TimeoutSampler( + wait_timeout=TIMEOUT_20SEC, + sleep=1, + func=lambda: pvc.instance.metadata.creationTimestamp != pvc_creation_timestamp, + ): + if sample: + return + raise TimeoutExpiredError("PVC was not recreated within the timeout period.") + + +def wait_dv_and_get_importer(dv: DataVolume, admin_client: DynamicClient) -> Pod: + """Wait for DataVolume import to start and get the importer pod. + + Args: + dv: The DataVolume resource. + admin_client: The admin client for accessing cluster resources. + + Returns: + Pod: The importer pod resource. + """ + dv.wait_for_status( + status=DataVolume.Status.IMPORT_IN_PROGRESS, + timeout=TIMEOUT_1MIN, + stop_status=DataVolume.Status.SUCCEEDED, + ) + return get_importer_pod(dyn_client=admin_client, namespace=dv.namespace) + + +def wait_for_multus_network_status(importer_pod: Pod) -> None: + """Wait for Multus network-status annotation to be populated on the importer pod. + + Multus CNI populates the network-status annotation asynchronously after the pod starts. + This function waits for the annotation to appear before proceeding. + + Args: + importer_pod: The importer pod resource. + + Raises: + TimeoutExpiredError: If the network-status annotation is not populated within the timeout period. + """ + network_status_annotation = f"{Resource.ApiGroup.K8S_V1_CNI_CNCF_IO}/network-status" + for sample in TimeoutSampler( + wait_timeout=TIMEOUT_1MIN, + sleep=TIMEOUT_5SEC, + func=lambda: importer_pod.instance.metadata.annotations.get(network_status_annotation), + ): + if sample: + return + raise TimeoutExpiredError( + f"Multus {network_status_annotation} annotation was not populated within the timeout period." + ) diff --git a/tests/storage/general/__init__.py b/tests/storage/general/__init__.py new file mode 100644 index 0000000000..e69de29bb2 diff --git a/tests/storage/general/conftest.py b/tests/storage/general/conftest.py new file mode 100644 index 0000000000..f703f7b731 --- /dev/null +++ b/tests/storage/general/conftest.py @@ -0,0 +1,103 @@ +""" +Storage general tests fixtures +""" + +import logging + +import pytest +from ocp_resources.virtual_machine_cluster_instancetype import VirtualMachineClusterInstancetype +from ocp_resources.virtual_machine_cluster_preference import VirtualMachineClusterPreference + +from tests.storage.cdi_import.utils import get_importer_pod_node, wait_dv_and_get_importer +from tests.storage.constants import QUAY_FEDORA_CONTAINER_IMAGE +from utilities.constants import OS_FLAVOR_FEDORA, REGISTRY_STR, TIMEOUT_5MIN, TIMEOUT_12MIN, U1_SMALL, Images +from utilities.storage import create_dv, data_volume_template_with_source_ref_dict, get_dv_size_from_datasource +from utilities.virt import VirtualMachineForTests, running_vm + +LOGGER = logging.getLogger(__name__) + + +@pytest.fixture() +def fedora_data_volume(namespace, fedora_data_source_scope_module, storage_class_name_scope_function): + """ + Provides a DataVolume created from Fedora DataSource. + + The DataVolume is created and waits for success before yielding. + """ + with create_dv( + dv_name=f"fedora-dv-{storage_class_name_scope_function}", + namespace=namespace.name, + storage_class=storage_class_name_scope_function, + size=get_dv_size_from_datasource(fedora_data_source_scope_module), + client=namespace.client, + source_ref={ + "kind": fedora_data_source_scope_module.kind, + "name": fedora_data_source_scope_module.name, + "namespace": fedora_data_source_scope_module.namespace, + }, + ) as dv: + dv.wait_for_dv_success(timeout=TIMEOUT_5MIN) + yield dv + + +@pytest.fixture() +def fedora_vm_with_instance_type( + namespace, + unprivileged_client, + fedora_data_source_scope_module, + storage_class_name_scope_function, +): + """ + Provides a running Fedora VM with instance type and preference. + + The VM is created with U1_SMALL instance type and Fedora preference, + using a DataVolume template from the provided data source. + """ + with VirtualMachineForTests( + name="fedora-vm", + namespace=namespace.name, + client=unprivileged_client, + os_flavor=OS_FLAVOR_FEDORA, + vm_instance_type=VirtualMachineClusterInstancetype(name=U1_SMALL, client=unprivileged_client), + vm_preference=VirtualMachineClusterPreference(name=OS_FLAVOR_FEDORA, client=unprivileged_client), + data_volume_template=data_volume_template_with_source_ref_dict( + data_source=fedora_data_source_scope_module, + storage_class=storage_class_name_scope_function, + ), + ) as vm: + running_vm(vm=vm) + yield vm + + +@pytest.fixture() +def fedora_dv_rwx_with_importer_node( + admin_client, + unprivileged_client, + namespace, + storage_class_matrix_rwx_matrix__function__, +): + """ + Provides a DataVolume imported from Quay registry with importer pod node information. + + Returns: + Tuple of (DataVolume, importer_pod_node_name) where the DataVolume is ready + and importer_pod_node_name is the node where the import operation ran. + """ + storage_class_name = next(iter(storage_class_matrix_rwx_matrix__function__)) + with create_dv( + dv_name=f"fedora-dv-different-node-{storage_class_name}", + namespace=namespace.name, + source=REGISTRY_STR, + url=QUAY_FEDORA_CONTAINER_IMAGE, + size=Images.Fedora.DEFAULT_DV_SIZE, + storage_class=storage_class_name, + client=unprivileged_client, + ) as dv: + LOGGER.info(f"Getting importer pod for DataVolume {dv.name}") + importer_pod = wait_dv_and_get_importer(dv=dv, admin_client=admin_client) + importer_pod_node = get_importer_pod_node(importer_pod=importer_pod) + LOGGER.info(f"Importer pod {importer_pod.name} is running on node {importer_pod_node}") + + dv.wait_for_dv_success(timeout=TIMEOUT_12MIN) + + yield dv, importer_pod_node diff --git a/tests/storage/general/test_storage_behavior.py b/tests/storage/general/test_storage_behavior.py new file mode 100644 index 0000000000..5edf0ca182 --- /dev/null +++ b/tests/storage/general/test_storage_behavior.py @@ -0,0 +1,116 @@ +""" +General storage behavior tests +""" + +import logging + +import pytest + +from tests.storage.cdi_import.utils import wait_for_pvc_recreate +from utilities import console +from utilities.constants import OS_FLAVOR_FEDORA, TIMEOUT_1MIN, Images +from utilities.infra import get_node_selector_dict +from utilities.storage import create_dummy_first_consumer_pod, create_vm_from_dv, sc_volume_binding_mode_is_wffc + +pytestmark = [ + pytest.mark.post_upgrade, +] + +LOGGER = logging.getLogger(__name__) + + +@pytest.mark.sno +@pytest.mark.polarion("CNV-675") +def test_pvc_recreates_after_deletion(fedora_data_volume, namespace, storage_class_name_scope_function): + """ + Test that a PVC is automatically recreated by CDI after manual deletion. + + Preconditions: + - Fedora DataSource available + - Storage class available + - DataVolume created from Fedora DataSource + - PVC bound and DataVolume import completed + + Steps: + 1. Record the PVC original creation timestamp + 2. Delete the PVC + 3. Wait for PVC to be recreated with a new timestamp + 4. Create a dummy first consumer pod if storage class uses WaitForFirstConsumer binding mode + 5. Wait for DataVolume to reach Succeeded status + + Expected: + - PVC is recreated automatically + - DataVolume status is "Succeeded" + """ + pvc = fedora_data_volume.pvc + pvc_original_timestamp = pvc.instance.metadata.creationTimestamp + pvc.delete() + wait_for_pvc_recreate(pvc=pvc, pvc_creation_timestamp=pvc_original_timestamp) + if sc_volume_binding_mode_is_wffc(sc=storage_class_name_scope_function): + create_dummy_first_consumer_pod(pvc=pvc) + fedora_data_volume.wait_for_dv_success() + + +@pytest.mark.polarion("CNV-3065") +@pytest.mark.sno +def test_disk_falloc(fedora_vm_with_instance_type): + """ + Test that attempting to allocate more space than available on a disk fails with the expected error. + + Preconditions: + - VM with instance type and preference created and running with console access + + Steps: + 1. Connect to VM console + 2. Execute fallocate command to allocate a file larger than the available disk space + 3. Verify the error message + + Expected: + - fallocate command fails with "No space left on device" error + """ + allocation_size_bytes = 42949672960 # 40GiB in bytes, assuming DV is about 30Gi + with console.Console(vm=fedora_vm_with_instance_type) as vm_console: + LOGGER.info(f"Attempting to allocate {allocation_size_bytes} bytes to trigger disk full error") + vm_console.sendline(f"fallocate -l {allocation_size_bytes} test-file") + vm_console.expect("No space left on device", timeout=TIMEOUT_1MIN) + + +@pytest.mark.polarion("CNV-3632") +def test_vm_from_dv_on_different_node( + unprivileged_client, + schedulable_nodes, + fedora_dv_rwx_with_importer_node, +): + """ + Test that a VM created from a DataVolume runs on a different node than the import operation. + + Preconditions: + - Storage class with RWX access mode (shared storage like Ceph or NFS) + - Multiple schedulable nodes available + - DataVolume imported from Quay registry + + Steps: + 1. Get nodes excluding the importer pod node + 2. Create and start a VM from the DataVolume on a different node + 3. Verify the VM is running on a different node than the importer pod + + Expected: + - VM runs successfully on a node different from the import operation node + """ + dv, importer_pod_node = fedora_dv_rwx_with_importer_node + + nodes = [node for node in schedulable_nodes if node.name != importer_pod_node] + assert nodes, f"No available nodes different from importer pod node {importer_pod_node}" + + with create_vm_from_dv( + dv=dv, + vm_name="fedora-vm-different-node", + os_flavor=OS_FLAVOR_FEDORA, + node_selector=get_node_selector_dict(node_selector=nodes[0].name), + memory_guest=Images.Fedora.DEFAULT_MEMORY_SIZE, + ) as vm: + vmi_node_name = vm.vmi.node.name + assert vmi_node_name != importer_pod_node, ( + f"VM is running on the same node as importer pod. Expected different nodes. " + f"Importer pod node: {importer_pod_node}, VM node: {vmi_node_name}" + ) diff --git a/utilities/pytest_matrix_utils.py b/utilities/pytest_matrix_utils.py index 3d6eb768fe..7d685e906a 100644 --- a/utilities/pytest_matrix_utils.py +++ b/utilities/pytest_matrix_utils.py @@ -5,6 +5,10 @@ def foo_matrix(matrix): return matrix """ +from functools import cache + +from kubernetes.dynamic import DynamicClient +from ocp_resources.resource import get_client from ocp_resources.storage_class import StorageClass from utilities.infra import cache_admin_client @@ -71,3 +75,34 @@ def immediate_matrix(matrix): if storage_class[storage_class_name]["wffc"] is False: matrix_to_return.append(storage_class) return matrix_to_return + + +def rwx_matrix(matrix: list[dict[str, dict[str, str]]]) -> list[dict[str, dict[str, str]]]: + """Filter storage classes with ReadWriteMany access mode. + + Args: + matrix: List of storage class dictionaries. + + Returns: + List of storage classes with RWX access mode. + """ + matrix_to_return = [] + for storage_class in matrix: + storage_class_name = [*storage_class][0] + if storage_class[storage_class_name]["access_mode"] == "ReadWriteMany": + matrix_to_return.append(storage_class) + return matrix_to_return + + +@cache +def _cache_admin_client() -> DynamicClient: + """Get admin_client once and reuse it + + This usage of this function is limited to places where `client` cannot be passed as an argument. + + Returns: + DynamicClient: admin_client + + """ + + return get_client() diff --git a/utilities/unittests/test_pytest_matrix_utils.py b/utilities/unittests/test_pytest_matrix_utils.py new file mode 100644 index 0000000000..fee506d88e --- /dev/null +++ b/utilities/unittests/test_pytest_matrix_utils.py @@ -0,0 +1,340 @@ +# Generated using Claude cli + +"""Unit tests for pytest_matrix_utils module""" + +from inspect import signature +from unittest.mock import MagicMock, patch + +import pytest + +from utilities.pytest_matrix_utils import ( # noqa: E402 + hpp_matrix, + immediate_matrix, + online_resize_matrix, + rwx_matrix, + snapshot_matrix, + wffc_matrix, + without_snapshot_capability_matrix, +) + + +class TestSnapshotMatrix: + """Test cases for snapshot_matrix function""" + + def test_snapshot_matrix_with_snapshot_enabled(self): + """Test snapshot_matrix filters storage classes with snapshot enabled""" + matrix = [ + {"sc-with-snapshot": {"snapshot": True, "other": "value"}}, + {"sc-without-snapshot": {"snapshot": False, "other": "value"}}, + {"sc-with-snapshot-2": {"snapshot": True, "other": "value"}}, + ] + + result = snapshot_matrix(matrix) + + assert len(result) == 2 + assert {"sc-with-snapshot": {"snapshot": True, "other": "value"}} in result + assert {"sc-with-snapshot-2": {"snapshot": True, "other": "value"}} in result + assert {"sc-without-snapshot": {"snapshot": False, "other": "value"}} not in result + + def test_snapshot_matrix_empty_matrix(self): + """Test snapshot_matrix with empty matrix""" + matrix = [] + + result = snapshot_matrix(matrix) + + assert result == [] + + def test_snapshot_matrix_no_snapshot_enabled(self): + """Test snapshot_matrix with no snapshot enabled storage classes""" + matrix = [ + {"sc-without-snapshot-1": {"snapshot": False, "other": "value"}}, + {"sc-without-snapshot-2": {"snapshot": False, "other": "value"}}, + ] + + result = snapshot_matrix(matrix) + + assert result == [] + + +class TestWithoutSnapshotCapabilityMatrix: + """Test cases for without_snapshot_capability_matrix function""" + + def test_without_snapshot_capability_matrix(self): + """Test without_snapshot_capability_matrix filters storage classes without snapshot capability""" + matrix = [ + {"sc-with-snapshot": {"snapshot": True, "other": "value"}}, + {"sc-without-snapshot": {"snapshot": False, "other": "value"}}, + {"sc-without-snapshot-2": {"snapshot": False, "other": "value"}}, + ] + + result = without_snapshot_capability_matrix(matrix) + + assert len(result) == 2 + assert {"sc-without-snapshot": {"snapshot": False, "other": "value"}} in result + assert {"sc-without-snapshot-2": {"snapshot": False, "other": "value"}} in result + assert {"sc-with-snapshot": {"snapshot": True, "other": "value"}} not in result + + def test_without_snapshot_capability_matrix_empty_matrix(self): + """Test without_snapshot_capability_matrix with empty matrix""" + matrix = [] + + result = without_snapshot_capability_matrix(matrix) + + assert result == [] + + def test_without_snapshot_capability_matrix_all_have_snapshot(self): + """Test without_snapshot_capability_matrix with all storage classes having snapshot capability""" + matrix = [ + {"sc-with-snapshot-1": {"snapshot": True, "other": "value"}}, + {"sc-with-snapshot-2": {"snapshot": True, "other": "value"}}, + ] + + result = without_snapshot_capability_matrix(matrix) + + assert result == [] + + +class TestOnlineResizeMatrix: + """Test cases for online_resize_matrix function""" + + def test_online_resize_matrix_with_online_resize_enabled(self): + """Test online_resize_matrix filters storage classes with online resize enabled""" + matrix = [ + {"sc-with-resize": {"online_resize": True, "other": "value"}}, + {"sc-without-resize": {"online_resize": False, "other": "value"}}, + {"sc-with-resize-2": {"online_resize": True, "other": "value"}}, + ] + + result = online_resize_matrix(matrix) + + assert len(result) == 2 + assert {"sc-with-resize": {"online_resize": True, "other": "value"}} in result + assert {"sc-with-resize-2": {"online_resize": True, "other": "value"}} in result + assert {"sc-without-resize": {"online_resize": False, "other": "value"}} not in result + + def test_online_resize_matrix_empty_matrix(self): + """Test online_resize_matrix with empty matrix""" + matrix = [] + + result = online_resize_matrix(matrix) + + assert result == [] + + +class TestHppMatrix: + """Test cases for hpp_matrix function""" + + @patch("utilities.pytest_matrix_utils._cache_admin_client") + @patch("utilities.pytest_matrix_utils.StorageClass") + def test_hpp_matrix_with_hpp_provisioner(self, mock_storage_class, mock_cache_admin_client): + """Test hpp_matrix filters storage classes with HPP provisioner""" + mock_client = MagicMock() + mock_cache_admin_client.return_value = mock_client + + # Mock StorageClass instances + mock_sc_hpp = MagicMock() + mock_sc_hpp.instance.provisioner = "kubevirt.io.hostpath-provisioner" + + mock_sc_non_hpp = MagicMock() + mock_sc_non_hpp.instance.provisioner = "other.provisioner" + + # Configure StorageClass mock to return different instances + def storage_class_side_effect(client, name): + if name == "hpp-sc": + return mock_sc_hpp + if name == "non-hpp-sc": + return mock_sc_non_hpp + return MagicMock() + + mock_storage_class.side_effect = storage_class_side_effect + mock_storage_class.Provisioner.HOSTPATH_CSI = "kubevirt.io/hostpath-csi" + mock_storage_class.Provisioner.HOSTPATH = "kubevirt.io.hostpath-provisioner" + + matrix = [ + {"hpp-sc": {"other": "value"}}, + {"non-hpp-sc": {"other": "value"}}, + ] + + result = hpp_matrix(matrix) + + assert len(result) == 1 + assert {"hpp-sc": {"other": "value"}} in result + + @patch("utilities.pytest_matrix_utils._cache_admin_client") + @patch("utilities.pytest_matrix_utils.StorageClass") + def test_hpp_matrix_empty_matrix(self, mock_storage_class, mock_cache_admin_client): + """Test hpp_matrix with empty matrix""" + matrix = [] + + result = hpp_matrix(matrix) + + assert result == [] + + +class TestWffcMatrix: + """Test cases for wffc_matrix function""" + + def test_wffc_matrix_with_wffc_enabled(self): + """Test wffc_matrix filters storage classes with WFFC enabled""" + matrix = [ + {"sc-with-wffc": {"wffc": True, "other": "value"}}, + {"sc-without-wffc": {"wffc": False, "other": "value"}}, + {"sc-with-wffc-2": {"wffc": True, "other": "value"}}, + ] + + result = wffc_matrix(matrix) + + assert len(result) == 2 + assert {"sc-with-wffc": {"wffc": True, "other": "value"}} in result + assert {"sc-with-wffc-2": {"wffc": True, "other": "value"}} in result + assert {"sc-without-wffc": {"wffc": False, "other": "value"}} not in result + + def test_wffc_matrix_empty_matrix(self): + """Test wffc_matrix with empty matrix""" + matrix = [] + + result = wffc_matrix(matrix) + + assert result == [] + + def test_wffc_matrix_no_wffc_enabled(self): + """Test wffc_matrix with no WFFC enabled storage classes""" + matrix = [ + {"sc-without-wffc-1": {"wffc": False, "other": "value"}}, + {"sc-without-wffc-2": {"wffc": False, "other": "value"}}, + ] + + result = wffc_matrix(matrix) + + assert result == [] + + +class TestImmediateMatrix: + """Test cases for immediate_matrix function""" + + def test_immediate_matrix_with_immediate_enabled(self): + """Test immediate_matrix filters storage classes with immediate enabled (WFFC disabled)""" + matrix = [ + {"sc-with-immediate": {"wffc": False, "other": "value"}}, + {"sc-without-immediate": {"wffc": True, "other": "value"}}, + {"sc-with-immediate-2": {"wffc": False, "other": "value"}}, + ] + + result = immediate_matrix(matrix) + + assert len(result) == 2 + assert {"sc-with-immediate": {"wffc": False, "other": "value"}} in result + assert {"sc-with-immediate-2": {"wffc": False, "other": "value"}} in result + assert {"sc-without-immediate": {"wffc": True, "other": "value"}} not in result + + def test_immediate_matrix_empty_matrix(self): + """Test immediate_matrix with empty matrix""" + matrix = [] + + result = immediate_matrix(matrix) + + assert result == [] + + def test_immediate_matrix_no_immediate_enabled(self): + """Test immediate_matrix with no WFFC enabled storage classes""" + matrix = [ + {"sc-without-immediate-1": {"wffc": True, "other": "value"}}, + {"sc-without-immediate-2": {"wffc": True, "other": "value"}}, + ] + + result = immediate_matrix(matrix) + + assert result == [] + + +class TestRwxMatrix: + """Test cases for rwx_matrix function""" + + def test_rwx_matrix_with_rwx_enabled(self): + """Test rwx_matrix filters storage classes with ReadWriteMany access mode""" + matrix = [ + {"sc-with-rwx": {"access_mode": "ReadWriteMany", "other": "value"}}, + {"sc-with-rwo": {"access_mode": "ReadWriteOnce", "other": "value"}}, + {"sc-with-rwx-2": {"access_mode": "ReadWriteMany", "other": "value"}}, + ] + + result = rwx_matrix(matrix) + + assert len(result) == 2 + assert {"sc-with-rwx": {"access_mode": "ReadWriteMany", "other": "value"}} in result + assert {"sc-with-rwx-2": {"access_mode": "ReadWriteMany", "other": "value"}} in result + assert {"sc-with-rwo": {"access_mode": "ReadWriteOnce", "other": "value"}} not in result + + def test_rwx_matrix_empty_matrix(self): + """Test rwx_matrix with empty matrix""" + matrix = [] + + result = rwx_matrix(matrix) + + assert result == [] + + def test_rwx_matrix_no_rwx_enabled(self): + """Test rwx_matrix with no RWX storage classes""" + matrix = [ + {"sc-rwo-1": {"access_mode": "ReadWriteOnce", "other": "value"}}, + {"sc-rwo-2": {"access_mode": "ReadWriteOnce", "other": "value"}}, + ] + + result = rwx_matrix(matrix) + + assert result == [] + + def test_rwx_matrix_missing_access_mode_key(self): + """Test rwx_matrix fails fast when access_mode key is missing""" + matrix = [ + {"sc-no-key": {"other": "value"}}, + ] + + with pytest.raises(KeyError, match="access_mode"): + rwx_matrix(matrix) + + +class TestMatrixFunctionSignatures: + """Test that all matrix functions accept only matrix argument""" + + def test_snapshot_matrix_signature(self): + """Test snapshot_matrix function signature""" + + sig = signature(snapshot_matrix) + params = list(sig.parameters.keys()) + assert params == ["matrix"] + + def test_without_snapshot_capability_matrix_signature(self): + """Test without_snapshot_capability_matrix function signature""" + + sig = signature(without_snapshot_capability_matrix) + params = list(sig.parameters.keys()) + assert params == ["matrix"] + + def test_online_resize_matrix_signature(self): + """Test online_resize_matrix function signature""" + + sig = signature(online_resize_matrix) + params = list(sig.parameters.keys()) + assert params == ["matrix"] + + def test_hpp_matrix_signature(self): + """Test hpp_matrix function signature""" + + sig = signature(hpp_matrix) + params = list(sig.parameters.keys()) + assert params == ["matrix"] + + def test_wffc_matrix_signature(self): + """Test wffc_matrix function signature""" + + sig = signature(wffc_matrix) + params = list(sig.parameters.keys()) + assert params == ["matrix"] + + def test_rwx_matrix_signature(self): + """Test rwx_matrix function signature""" + + sig = signature(rwx_matrix) + params = list(sig.parameters.keys()) + assert params == ["matrix"] From e7f2516ed9899c2bd0184d3c0198659f1aeee312 Mon Sep 17 00:00:00 2001 From: Emanuele Prella Date: Mon, 15 Jun 2026 10:28:22 +0200 Subject: [PATCH 2/4] Remove test_pytest_matrix_utils.py Signed-off-by: Emanuele Prella --- .../unittests/test_pytest_matrix_utils.py | 340 ------------------ 1 file changed, 340 deletions(-) delete mode 100644 utilities/unittests/test_pytest_matrix_utils.py diff --git a/utilities/unittests/test_pytest_matrix_utils.py b/utilities/unittests/test_pytest_matrix_utils.py deleted file mode 100644 index fee506d88e..0000000000 --- a/utilities/unittests/test_pytest_matrix_utils.py +++ /dev/null @@ -1,340 +0,0 @@ -# Generated using Claude cli - -"""Unit tests for pytest_matrix_utils module""" - -from inspect import signature -from unittest.mock import MagicMock, patch - -import pytest - -from utilities.pytest_matrix_utils import ( # noqa: E402 - hpp_matrix, - immediate_matrix, - online_resize_matrix, - rwx_matrix, - snapshot_matrix, - wffc_matrix, - without_snapshot_capability_matrix, -) - - -class TestSnapshotMatrix: - """Test cases for snapshot_matrix function""" - - def test_snapshot_matrix_with_snapshot_enabled(self): - """Test snapshot_matrix filters storage classes with snapshot enabled""" - matrix = [ - {"sc-with-snapshot": {"snapshot": True, "other": "value"}}, - {"sc-without-snapshot": {"snapshot": False, "other": "value"}}, - {"sc-with-snapshot-2": {"snapshot": True, "other": "value"}}, - ] - - result = snapshot_matrix(matrix) - - assert len(result) == 2 - assert {"sc-with-snapshot": {"snapshot": True, "other": "value"}} in result - assert {"sc-with-snapshot-2": {"snapshot": True, "other": "value"}} in result - assert {"sc-without-snapshot": {"snapshot": False, "other": "value"}} not in result - - def test_snapshot_matrix_empty_matrix(self): - """Test snapshot_matrix with empty matrix""" - matrix = [] - - result = snapshot_matrix(matrix) - - assert result == [] - - def test_snapshot_matrix_no_snapshot_enabled(self): - """Test snapshot_matrix with no snapshot enabled storage classes""" - matrix = [ - {"sc-without-snapshot-1": {"snapshot": False, "other": "value"}}, - {"sc-without-snapshot-2": {"snapshot": False, "other": "value"}}, - ] - - result = snapshot_matrix(matrix) - - assert result == [] - - -class TestWithoutSnapshotCapabilityMatrix: - """Test cases for without_snapshot_capability_matrix function""" - - def test_without_snapshot_capability_matrix(self): - """Test without_snapshot_capability_matrix filters storage classes without snapshot capability""" - matrix = [ - {"sc-with-snapshot": {"snapshot": True, "other": "value"}}, - {"sc-without-snapshot": {"snapshot": False, "other": "value"}}, - {"sc-without-snapshot-2": {"snapshot": False, "other": "value"}}, - ] - - result = without_snapshot_capability_matrix(matrix) - - assert len(result) == 2 - assert {"sc-without-snapshot": {"snapshot": False, "other": "value"}} in result - assert {"sc-without-snapshot-2": {"snapshot": False, "other": "value"}} in result - assert {"sc-with-snapshot": {"snapshot": True, "other": "value"}} not in result - - def test_without_snapshot_capability_matrix_empty_matrix(self): - """Test without_snapshot_capability_matrix with empty matrix""" - matrix = [] - - result = without_snapshot_capability_matrix(matrix) - - assert result == [] - - def test_without_snapshot_capability_matrix_all_have_snapshot(self): - """Test without_snapshot_capability_matrix with all storage classes having snapshot capability""" - matrix = [ - {"sc-with-snapshot-1": {"snapshot": True, "other": "value"}}, - {"sc-with-snapshot-2": {"snapshot": True, "other": "value"}}, - ] - - result = without_snapshot_capability_matrix(matrix) - - assert result == [] - - -class TestOnlineResizeMatrix: - """Test cases for online_resize_matrix function""" - - def test_online_resize_matrix_with_online_resize_enabled(self): - """Test online_resize_matrix filters storage classes with online resize enabled""" - matrix = [ - {"sc-with-resize": {"online_resize": True, "other": "value"}}, - {"sc-without-resize": {"online_resize": False, "other": "value"}}, - {"sc-with-resize-2": {"online_resize": True, "other": "value"}}, - ] - - result = online_resize_matrix(matrix) - - assert len(result) == 2 - assert {"sc-with-resize": {"online_resize": True, "other": "value"}} in result - assert {"sc-with-resize-2": {"online_resize": True, "other": "value"}} in result - assert {"sc-without-resize": {"online_resize": False, "other": "value"}} not in result - - def test_online_resize_matrix_empty_matrix(self): - """Test online_resize_matrix with empty matrix""" - matrix = [] - - result = online_resize_matrix(matrix) - - assert result == [] - - -class TestHppMatrix: - """Test cases for hpp_matrix function""" - - @patch("utilities.pytest_matrix_utils._cache_admin_client") - @patch("utilities.pytest_matrix_utils.StorageClass") - def test_hpp_matrix_with_hpp_provisioner(self, mock_storage_class, mock_cache_admin_client): - """Test hpp_matrix filters storage classes with HPP provisioner""" - mock_client = MagicMock() - mock_cache_admin_client.return_value = mock_client - - # Mock StorageClass instances - mock_sc_hpp = MagicMock() - mock_sc_hpp.instance.provisioner = "kubevirt.io.hostpath-provisioner" - - mock_sc_non_hpp = MagicMock() - mock_sc_non_hpp.instance.provisioner = "other.provisioner" - - # Configure StorageClass mock to return different instances - def storage_class_side_effect(client, name): - if name == "hpp-sc": - return mock_sc_hpp - if name == "non-hpp-sc": - return mock_sc_non_hpp - return MagicMock() - - mock_storage_class.side_effect = storage_class_side_effect - mock_storage_class.Provisioner.HOSTPATH_CSI = "kubevirt.io/hostpath-csi" - mock_storage_class.Provisioner.HOSTPATH = "kubevirt.io.hostpath-provisioner" - - matrix = [ - {"hpp-sc": {"other": "value"}}, - {"non-hpp-sc": {"other": "value"}}, - ] - - result = hpp_matrix(matrix) - - assert len(result) == 1 - assert {"hpp-sc": {"other": "value"}} in result - - @patch("utilities.pytest_matrix_utils._cache_admin_client") - @patch("utilities.pytest_matrix_utils.StorageClass") - def test_hpp_matrix_empty_matrix(self, mock_storage_class, mock_cache_admin_client): - """Test hpp_matrix with empty matrix""" - matrix = [] - - result = hpp_matrix(matrix) - - assert result == [] - - -class TestWffcMatrix: - """Test cases for wffc_matrix function""" - - def test_wffc_matrix_with_wffc_enabled(self): - """Test wffc_matrix filters storage classes with WFFC enabled""" - matrix = [ - {"sc-with-wffc": {"wffc": True, "other": "value"}}, - {"sc-without-wffc": {"wffc": False, "other": "value"}}, - {"sc-with-wffc-2": {"wffc": True, "other": "value"}}, - ] - - result = wffc_matrix(matrix) - - assert len(result) == 2 - assert {"sc-with-wffc": {"wffc": True, "other": "value"}} in result - assert {"sc-with-wffc-2": {"wffc": True, "other": "value"}} in result - assert {"sc-without-wffc": {"wffc": False, "other": "value"}} not in result - - def test_wffc_matrix_empty_matrix(self): - """Test wffc_matrix with empty matrix""" - matrix = [] - - result = wffc_matrix(matrix) - - assert result == [] - - def test_wffc_matrix_no_wffc_enabled(self): - """Test wffc_matrix with no WFFC enabled storage classes""" - matrix = [ - {"sc-without-wffc-1": {"wffc": False, "other": "value"}}, - {"sc-without-wffc-2": {"wffc": False, "other": "value"}}, - ] - - result = wffc_matrix(matrix) - - assert result == [] - - -class TestImmediateMatrix: - """Test cases for immediate_matrix function""" - - def test_immediate_matrix_with_immediate_enabled(self): - """Test immediate_matrix filters storage classes with immediate enabled (WFFC disabled)""" - matrix = [ - {"sc-with-immediate": {"wffc": False, "other": "value"}}, - {"sc-without-immediate": {"wffc": True, "other": "value"}}, - {"sc-with-immediate-2": {"wffc": False, "other": "value"}}, - ] - - result = immediate_matrix(matrix) - - assert len(result) == 2 - assert {"sc-with-immediate": {"wffc": False, "other": "value"}} in result - assert {"sc-with-immediate-2": {"wffc": False, "other": "value"}} in result - assert {"sc-without-immediate": {"wffc": True, "other": "value"}} not in result - - def test_immediate_matrix_empty_matrix(self): - """Test immediate_matrix with empty matrix""" - matrix = [] - - result = immediate_matrix(matrix) - - assert result == [] - - def test_immediate_matrix_no_immediate_enabled(self): - """Test immediate_matrix with no WFFC enabled storage classes""" - matrix = [ - {"sc-without-immediate-1": {"wffc": True, "other": "value"}}, - {"sc-without-immediate-2": {"wffc": True, "other": "value"}}, - ] - - result = immediate_matrix(matrix) - - assert result == [] - - -class TestRwxMatrix: - """Test cases for rwx_matrix function""" - - def test_rwx_matrix_with_rwx_enabled(self): - """Test rwx_matrix filters storage classes with ReadWriteMany access mode""" - matrix = [ - {"sc-with-rwx": {"access_mode": "ReadWriteMany", "other": "value"}}, - {"sc-with-rwo": {"access_mode": "ReadWriteOnce", "other": "value"}}, - {"sc-with-rwx-2": {"access_mode": "ReadWriteMany", "other": "value"}}, - ] - - result = rwx_matrix(matrix) - - assert len(result) == 2 - assert {"sc-with-rwx": {"access_mode": "ReadWriteMany", "other": "value"}} in result - assert {"sc-with-rwx-2": {"access_mode": "ReadWriteMany", "other": "value"}} in result - assert {"sc-with-rwo": {"access_mode": "ReadWriteOnce", "other": "value"}} not in result - - def test_rwx_matrix_empty_matrix(self): - """Test rwx_matrix with empty matrix""" - matrix = [] - - result = rwx_matrix(matrix) - - assert result == [] - - def test_rwx_matrix_no_rwx_enabled(self): - """Test rwx_matrix with no RWX storage classes""" - matrix = [ - {"sc-rwo-1": {"access_mode": "ReadWriteOnce", "other": "value"}}, - {"sc-rwo-2": {"access_mode": "ReadWriteOnce", "other": "value"}}, - ] - - result = rwx_matrix(matrix) - - assert result == [] - - def test_rwx_matrix_missing_access_mode_key(self): - """Test rwx_matrix fails fast when access_mode key is missing""" - matrix = [ - {"sc-no-key": {"other": "value"}}, - ] - - with pytest.raises(KeyError, match="access_mode"): - rwx_matrix(matrix) - - -class TestMatrixFunctionSignatures: - """Test that all matrix functions accept only matrix argument""" - - def test_snapshot_matrix_signature(self): - """Test snapshot_matrix function signature""" - - sig = signature(snapshot_matrix) - params = list(sig.parameters.keys()) - assert params == ["matrix"] - - def test_without_snapshot_capability_matrix_signature(self): - """Test without_snapshot_capability_matrix function signature""" - - sig = signature(without_snapshot_capability_matrix) - params = list(sig.parameters.keys()) - assert params == ["matrix"] - - def test_online_resize_matrix_signature(self): - """Test online_resize_matrix function signature""" - - sig = signature(online_resize_matrix) - params = list(sig.parameters.keys()) - assert params == ["matrix"] - - def test_hpp_matrix_signature(self): - """Test hpp_matrix function signature""" - - sig = signature(hpp_matrix) - params = list(sig.parameters.keys()) - assert params == ["matrix"] - - def test_wffc_matrix_signature(self): - """Test wffc_matrix function signature""" - - sig = signature(wffc_matrix) - params = list(sig.parameters.keys()) - assert params == ["matrix"] - - def test_rwx_matrix_signature(self): - """Test rwx_matrix function signature""" - - sig = signature(rwx_matrix) - params = list(sig.parameters.keys()) - assert params == ["matrix"] From dd9b5dc24c6b329b051fe7f305fdcdb24b732573 Mon Sep 17 00:00:00 2001 From: Emanuele Prella Date: Mon, 15 Jun 2026 10:47:36 +0200 Subject: [PATCH 3/4] remove unused Signed-off-by: Emanuele Prella --- tests/storage/cdi_import/test_import_http.py | 1 - utilities/pytest_matrix_utils.py | 18 ------------------ 2 files changed, 19 deletions(-) diff --git a/tests/storage/cdi_import/test_import_http.py b/tests/storage/cdi_import/test_import_http.py index 25da7e09a7..51ccc9ffc1 100644 --- a/tests/storage/cdi_import/test_import_http.py +++ b/tests/storage/cdi_import/test_import_http.py @@ -49,7 +49,6 @@ TAR_IMG = "archive.tar" DEFAULT_DV_SIZE = Images.Cirros.DEFAULT_DV_SIZE SMALL_DV_SIZE = "200Mi" -LATEST_WINDOWS_OS_DICT = py_config.get("latest_windows_os_dict", {}) @pytest.mark.sno diff --git a/utilities/pytest_matrix_utils.py b/utilities/pytest_matrix_utils.py index 7d685e906a..22db3900fd 100644 --- a/utilities/pytest_matrix_utils.py +++ b/utilities/pytest_matrix_utils.py @@ -5,10 +5,6 @@ def foo_matrix(matrix): return matrix """ -from functools import cache - -from kubernetes.dynamic import DynamicClient -from ocp_resources.resource import get_client from ocp_resources.storage_class import StorageClass from utilities.infra import cache_admin_client @@ -92,17 +88,3 @@ def rwx_matrix(matrix: list[dict[str, dict[str, str]]]) -> list[dict[str, dict[s if storage_class[storage_class_name]["access_mode"] == "ReadWriteMany": matrix_to_return.append(storage_class) return matrix_to_return - - -@cache -def _cache_admin_client() -> DynamicClient: - """Get admin_client once and reuse it - - This usage of this function is limited to places where `client` cannot be passed as an argument. - - Returns: - DynamicClient: admin_client - - """ - - return get_client() From 174bc63eec39d933180754533d460c19ec54a542 Mon Sep 17 00:00:00 2001 From: Emanuele Prella Date: Tue, 16 Jun 2026 10:22:09 +0200 Subject: [PATCH 4/4] Remove admin_client Signed-off-by: Emanuele Prella --- tests/storage/cdi_import/conftest.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/tests/storage/cdi_import/conftest.py b/tests/storage/cdi_import/conftest.py index 591711123f..4f27cc27aa 100644 --- a/tests/storage/cdi_import/conftest.py +++ b/tests/storage/cdi_import/conftest.py @@ -36,7 +36,7 @@ @pytest.fixture() -def bridge_on_node(admin_client): +def bridge_on_node(): """Create a Linux Bridge network device and yield it.""" with network_device( interface_type=LINUX_BRIDGE, @@ -47,7 +47,7 @@ def bridge_on_node(admin_client): @pytest.fixture() -def linux_nad(admin_client, namespace, bridge_on_node): +def linux_nad(namespace, bridge_on_node): """Create a Linux Bridge Network Attachment Definition (NAD) and yield it.""" with network_nad( namespace=namespace,