[4.19] Storage: replace artifactory with DataSource in test_golden_image

tests/storage/conftest.py

@@ -49,6 +49,7 @@
     CDI_UPLOADPROXY,
     CNV_TEST_SERVICE_ACCOUNT,
     CNV_TESTS_CONTAINER,
+    OS_FLAVOR_FEDORA,
     OS_FLAVOR_RHEL,
     RHEL10_PREFERENCE,
     SECURITY_CONTEXT,
@@ -562,3 +563,13 @@ def rhel10_data_source_scope_session(golden_images_namespace):
         client=golden_images_namespace.client,
         ensure_exists=True,
     )
+
+
+@pytest.fixture(scope="module")
+def fedora_data_source_scope_module(golden_images_namespace):
+    return DataSource(
+        namespace=golden_images_namespace.name,
+        name=OS_FLAVOR_FEDORA,
+        client=golden_images_namespace.client,
+        ensure_exists=True,
+    )

tests/storage/golden_image/test_golden_image.py

@@ -6,25 +6,34 @@
 from ocp_resources.persistent_volume_claim import PersistentVolumeClaim
 from pytest_testconfig import config as py_config
 
-from tests.os_params import RHEL_LATEST, RHEL_LATEST_LABELS
+from tests.os_params import FEDORA_LATEST_LABELS
 from utilities.constants import PVC, TIMEOUT_20MIN
-from utilities.storage import ErrorMsg, create_dv, get_test_artifact_server_url
-from utilities.virt import wait_for_ssh_connectivity
+from utilities.storage import ErrorMsg, create_dv, create_dv_with_source_ref, get_dv_size_from_datasource
+from utilities.virt import vm_instance_from_template, wait_for_ssh_connectivity
 
 pytestmark = pytest.mark.post_upgrade
 
 
 LOGGER = logging.getLogger(__name__)
-LATEST_RHEL_IMAGE = RHEL_LATEST["image_path"]
-RHEL_IMAGE_SIZE = RHEL_LATEST["dv_size"]
 
 
-DV_PARAM = {
-    "dv_name": "golden-image-dv",
-    "image": LATEST_RHEL_IMAGE,
-    "dv_size": RHEL_IMAGE_SIZE,
-    "storage_class": py_config["default_storage_class"],
-}
+@pytest.fixture(scope="module")
+def golden_image_dv_from_fedora_datasource_scope_module(
+    admin_client,
+    golden_images_namespace,
+    fedora_data_source_scope_module,
+):
+    size = get_dv_size_from_datasource(data_source=fedora_data_source_scope_module)
+    with create_dv_with_source_ref(
+        client=admin_client,
+        dv_name=f"golden-image-fedora-{py_config['default_storage_class']}",
+        namespace=golden_images_namespace.name,
+        size=size,
+        storage_class=py_config["default_storage_class"],
+        data_source=fedora_data_source_scope_module,
+    ) as dv:
+        dv.wait_for_dv_success()
+        yield dv
 
 
 @pytest.fixture
@@ -34,132 +43,127 @@ def dv_created_by_unprivileged_user_with_rolebinding(
     golden_images_edit_rolebinding,
     unprivileged_client,
     storage_class_name_scope_function,
+    fedora_data_source_scope_module,
 ):
-    with create_dv(
+    size = get_dv_size_from_datasource(data_source=fedora_data_source_scope_module)
+    with create_dv_with_source_ref(
         client=unprivileged_client,
         dv_name=f"{request.param['dv_name']}-{storage_class_name_scope_function}",
         namespace=golden_images_namespace.name,
-        url=f"{get_test_artifact_server_url()}{LATEST_RHEL_IMAGE}",
-        size=RHEL_IMAGE_SIZE,
+        size=size,
         storage_class=storage_class_name_scope_function,
+        data_source=fedora_data_source_scope_module,
     ) as dv:
         yield dv
 
 
+@pytest.fixture()
+def fedora_vm_from_datasource_multi_storage_scope_function(
+    request,
+    unprivileged_client,
+    namespace,
+    fedora_data_source_scope_module,
+):
+    with vm_instance_from_template(
+        request=request,
+        unprivileged_client=unprivileged_client,
+        namespace=namespace,
+        data_source=fedora_data_source_scope_module,
+    ) as vm:
+        yield vm
+
+
 @pytest.mark.sno
 @pytest.mark.polarion("CNV-4755")
 def test_regular_user_cant_create_dv_in_ns(
     golden_images_namespace,
     unprivileged_client,
+    fedora_data_source_scope_module,
 ):
     LOGGER.info("Try as a regular user, to create a DV in golden image NS and receive the proper error")
+    size = get_dv_size_from_datasource(data_source=fedora_data_source_scope_module)
     with pytest.raises(
         ApiException,
         match=ErrorMsg.CANNOT_CREATE_RESOURCE,
     ):
-        with create_dv(
+        with create_dv_with_source_ref(
             dv_name="cnv-4755",
             namespace=golden_images_namespace.name,
-            url=f"{get_test_artifact_server_url()}{LATEST_RHEL_IMAGE}",
-            size=RHEL_IMAGE_SIZE,
+            size=size,
             storage_class=py_config["default_storage_class"],
             client=unprivileged_client,
+            data_source=fedora_data_source_scope_module,
         ):
             return
 
 
 @pytest.mark.sno
-@pytest.mark.parametrize(
-    "golden_image_data_volume_scope_module",
-    [
-        pytest.param(DV_PARAM, marks=pytest.mark.polarion("CNV-4756")),
-    ],
-    indirect=True,
-)
+@pytest.mark.polarion("CNV-4756")
 def test_regular_user_cant_delete_dv_from_cloned_dv(
     golden_images_namespace,
     unprivileged_client,
-    golden_image_data_volume_scope_module,
+    golden_image_dv_from_fedora_datasource_scope_module,
 ):
     LOGGER.info("Try as a regular user, to delete a dv from golden image NS and receive the proper error")
     with pytest.raises(
         ApiException,
         match=ErrorMsg.CANNOT_DELETE_RESOURCE,
     ):
         DataVolume(
-            name=golden_image_data_volume_scope_module.name,
-            namespace=golden_image_data_volume_scope_module.namespace,
+            name=golden_image_dv_from_fedora_datasource_scope_module.name,
+            namespace=golden_image_dv_from_fedora_datasource_scope_module.namespace,
             client=unprivileged_client,
         ).delete()
 
 
 @pytest.mark.sno
 @pytest.mark.gating
 @pytest.mark.parametrize(
-    "golden_image_data_volume_multi_storage_scope_function,"
-    "golden_image_vm_instance_from_template_multi_storage_scope_function",
+    "fedora_vm_from_datasource_multi_storage_scope_function",
     [
         pytest.param(
             {
-                "dv_name": "cnv-4757",
-                "image": LATEST_RHEL_IMAGE,
-                "dv_size": RHEL_IMAGE_SIZE,
-            },
-            {
-                "vm_name": "rhel-vm",
-                "template_labels": RHEL_LATEST_LABELS,
+                "vm_name": "fedora-vm",
+                "template_labels": FEDORA_LATEST_LABELS,
             },
             marks=pytest.mark.polarion("CNV-4757"),
         ),
     ],
     indirect=True,
 )
 def test_regular_user_can_create_vm_from_cloned_dv(
-    golden_image_data_volume_multi_storage_scope_function,
-    golden_image_vm_instance_from_template_multi_storage_scope_function,
+    fedora_vm_from_datasource_multi_storage_scope_function,
 ):
-    wait_for_ssh_connectivity(vm=golden_image_vm_instance_from_template_multi_storage_scope_function)
+    wait_for_ssh_connectivity(vm=fedora_vm_from_datasource_multi_storage_scope_function)
 
 
 @pytest.mark.sno
-@pytest.mark.parametrize(
-    "golden_image_data_volume_scope_module",
-    [
-        pytest.param(DV_PARAM, marks=pytest.mark.polarion("CNV-4758")),
-    ],
-    indirect=True,
-)
+@pytest.mark.polarion("CNV-4758")
 def test_regular_user_can_list_all_pvc_in_ns(
     golden_images_namespace,
     unprivileged_client,
-    golden_image_data_volume_scope_module,
+    golden_image_dv_from_fedora_datasource_scope_module,
 ):
     LOGGER.info("Make sure regular user have permissions to view PVC's in golden image NS")
     assert list(
         PersistentVolumeClaim.get(
             dyn_client=unprivileged_client,
             namespace=golden_images_namespace.name,
-            field_selector=f"metadata.name=={golden_image_data_volume_scope_module.name}",
+            field_selector=f"metadata.name=={golden_image_dv_from_fedora_datasource_scope_module.name}",
         )
     )
 
 
 @pytest.mark.sno
-@pytest.mark.parametrize(
-    "golden_image_data_volume_scope_module",
-    [
-        pytest.param(DV_PARAM, marks=pytest.mark.polarion("CNV-4760")),
-    ],
-    indirect=True,
-)
+@pytest.mark.polarion("CNV-4760")
 def test_regular_user_cant_clone_dv_in_ns(
     unprivileged_client,
-    golden_image_data_volume_scope_module,
+    golden_image_dv_from_fedora_datasource_scope_module,
 ):
     LOGGER.info("Try to clone a DV in the golden image NS and fail with the proper message")
 
-    storage_class = golden_image_data_volume_scope_module.storage_class
-    golden_images_namespace = golden_image_data_volume_scope_module.namespace
+    storage_class = golden_image_dv_from_fedora_datasource_scope_module.storage_class
+    golden_images_namespace = golden_image_dv_from_fedora_datasource_scope_module.namespace
 
     with pytest.raises(
         ApiException,
@@ -169,8 +173,8 @@ def test_regular_user_cant_clone_dv_in_ns(
             dv_name=f"cnv-4760-{storage_class}",
             namespace=golden_images_namespace,
             source=PVC,
-            size=golden_image_data_volume_scope_module.size,
-            source_pvc=golden_image_data_volume_scope_module.pvc.name,
+            size=golden_image_dv_from_fedora_datasource_scope_module.size,
+            source_pvc=golden_image_dv_from_fedora_datasource_scope_module.pvc.name,
             source_namespace=golden_images_namespace,
             client=unprivileged_client,
             storage_class=storage_class,

utilities/storage.py

@@ -170,6 +170,43 @@ def create_dv(
     )
 
 
+@contextmanager
+def create_dv_with_source_ref(
+    dv_name,
+    namespace,
+    storage_class,
+    size,
+    data_source,
+    client=None,
+    consume_wffc=True,
+    api_name="storage",
+):
+    # openshift-python-wrapper doesn't support source_ref parameter.
+    dv = DataVolume(
+        source="blank",
+        name=dv_name,
+        namespace=namespace,
+        size=size,
+        storage_class=storage_class,
+        client=client,
+        api_name=api_name,
+    )
+    dv.to_dict()
+    if "source" in dv.res["spec"]:
+        del dv.res["spec"]["source"]
+    if "contentType" in dv.res["spec"]:
+        del dv.res["spec"]["contentType"]
+    dv.res["spec"]["sourceRef"] = {
+        "kind": data_source.kind,
+        "name": data_source.name,
+        "namespace": data_source.namespace,
+    }
+    with dv:
+        if sc_volume_binding_mode_is_wffc(sc=storage_class) and consume_wffc:
+            create_dummy_first_consumer_pod(dv=dv)
+        yield dv
+
+
 def data_volume(
     namespace,
     storage_class_matrix=None,