Skip to content

[4.19][Storage] Move tests away from test_import_http.py and replace using datasource cloning #5202

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
rnetser merged 5 commits into from
Jun 21, 2026
Merged

[4.19][Storage] Move tests away from test_import_http.py and replace using datasource cloning #5202

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
e0ea754
[4.20] [Storage] CherryPicked: Move tests away from test_import_http.…
ema-aka-young May 4, 2026
e7f2516
Remove test_pytest_matrix_utils.py
ema-aka-young Jun 15, 2026
dd9b5dc
remove unused
ema-aka-young Jun 15, 2026
174bc63
Remove admin_client
ema-aka-young Jun 16, 2026
e11b154
Merge branch 'cnv-4.19' into CNV-83361-replace-http-import-cnv-4.20-c…
ema-aka-young Jun 17, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 0 additions & 5 deletions tests/conftest.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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()
Comment thread
ema-aka-young marked this conversation as resolved.
Comment thread
ema-aka-young marked this conversation as resolved.
def skip_access_mode_rwo_scope_function(storage_class_matrix__function__):
Comment thread
ema-aka-young marked this conversation as resolved.
Comment thread
ema-aka-young marked this conversation as resolved.
Comment thread
ema-aka-young marked this conversation as resolved.
_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__)
Expand Down
70 changes: 28 additions & 42 deletions tests/storage/cdi_import/conftest.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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,
)
Expand All @@ -36,14 +35,9 @@
DEFAULT_DV_SIZE = Images.Cirros.DEFAULT_DV_SIZE


@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():
"""Create a Linux Bridge network device and yield it."""
with network_device(
Comment thread
ema-aka-young marked this conversation as resolved.
interface_type=LINUX_BRIDGE,
nncp_name=BRIDGE_NAME,
Expand All @@ -54,6 +48,7 @@ def bridge_on_node():

@pytest.fixture()
def linux_nad(namespace, bridge_on_node):
"""Create a Linux Bridge Network Attachment Definition (NAD) and yield it."""
with network_nad(
Comment thread
ema-aka-young marked this conversation as resolved.
namespace=namespace,
nad_type=LINUX_BRIDGE,
Expand All @@ -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,
Expand All @@ -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,
Expand All @@ -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,
Expand All @@ -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):
Expand Down Expand Up @@ -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:
Expand Down Expand Up @@ -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
176 changes: 3 additions & 173 deletions tests/storage/cdi_import/test_import_http.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand All @@ -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

Expand All @@ -61,78 +51,6 @@
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()


@pytest.mark.sno
@pytest.mark.polarion("CNV-876")
def test_invalid_url(dv_non_exist_url):
Expand Down Expand Up @@ -451,78 +369,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,"
Expand Down Expand Up @@ -557,19 +403,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")
Loading