Add exclude filter to verify_boot_sources_reimported

tests/install_upgrade_operators/hco_enablement_golden_image_updates/conftest.py

@@ -3,6 +3,7 @@
 from ocp_resources.pod import Pod
 from ocp_utilities.infra import get_pods_by_name_prefix
 
+from tests.install_upgrade_operators.constants import CUSTOM_DATASOURCE_NAME
 from tests.install_upgrade_operators.hco_enablement_golden_image_updates.utils import (
     CUSTOM_TEMPLATE,
     HCO_CR_DATA_IMPORT_SCHEDULE_KEY,
@@ -15,6 +16,7 @@
     COMMON_TEMPLATES_KEY_NAME,
     SSP_CR_COMMON_TEMPLATES_LIST_KEY_NAME,
 )
+from utilities.hco import disable_common_boot_image_import_hco_spec
 from utilities.ssp import get_ssp_resource
 
 
@@ -113,3 +115,20 @@ def ssp_spec_templates_scope_function(ssp_resource_scope_function):
 @pytest.fixture(scope="session")
 def common_templates_scope_session(hyperconverged_status_scope_session):
     return hyperconverged_status_scope_session[SSP_CR_COMMON_TEMPLATES_LIST_KEY_NAME]
+
+
+@pytest.fixture()
+def disabled_boot_image_import_excluding_custom_datasource(
+    admin_client,
+    hyperconverged_resource_scope_function,
+    golden_images_namespace,
+    golden_images_data_import_crons_scope_function,
+):
+    """Disable common boot image import, skipping verification of the custom DataSource."""
+    yield from disable_common_boot_image_import_hco_spec(
+        admin_client=admin_client,
+        hco_resource=hyperconverged_resource_scope_function,
+        golden_images_namespace=golden_images_namespace,
+        golden_images_data_import_crons=golden_images_data_import_crons_scope_function,
+        exclude_data_source_names={CUSTOM_DATASOURCE_NAME},
+    )

tests/install_upgrade_operators/hco_enablement_golden_image_updates/test_update_hco_cr.py

@@ -10,7 +10,6 @@
     get_template_dict_by_name,
     get_templates_by_type_from_hco_status,
 )
-from utilities.constants.pytest import QUARANTINED
 from utilities.hco import (
     update_hco_templates_spec,
     wait_for_auto_boot_config_stabilization,
@@ -88,18 +87,13 @@ def test_add_custom_data_import_cron_template(
             ssp_spec_templates_scope_function=ssp_spec_templates_scope_function,
         )
 
-    @pytest.mark.xfail(
-        reason=f"{QUARANTINED}: Teardown AssertionError in verify_boot_sources_reimported after re-enabling"
-        " enableCommonBootImageImport spec, CNV-88015",
-        run=False,
-    )
     @pytest.mark.dependency(name="test_add_custom_data_import_cron_template_disable_spec")
     @pytest.mark.polarion("CNV-7914")
+    @pytest.mark.usefixtures("disabled_boot_image_import_excluding_custom_datasource")
     def test_add_custom_data_import_cron_template_disable_spec(
         self,
         admin_client,
         hco_namespace,
-        disabled_common_boot_image_import_hco_spec_scope_function,
         hyperconverged_status_templates_scope_function,
         ssp_spec_templates_scope_function,
         image_stream_names,

utilities/hco.py

@@ -1,6 +1,8 @@
 import json
 import logging
+from collections.abc import Collection, Iterator
 from contextlib import contextmanager
+from typing import TYPE_CHECKING
 
 from kubernetes.dynamic.exceptions import NotFoundError, ResourceNotFoundError
 from ocp_resources.cdi import CDI
@@ -37,6 +39,10 @@
 )
 from utilities.storage import verify_boot_sources_reimported
 
+if TYPE_CHECKING:
+    from kubernetes.dynamic import DynamicClient
+    from ocp_resources.data_import_cron import DataImportCron
+
 LOGGER = logging.getLogger(__name__)
 
 DEFAULT_HCO_PROGRESSING_CONDITIONS = {
@@ -348,11 +354,12 @@ def wait_for_hco_version(client, hco_ns_name, cnv_version):
 
 
 def disable_common_boot_image_import_hco_spec(
-    admin_client,
-    hco_resource,
-    golden_images_namespace,
-    golden_images_data_import_crons,
-):
+    admin_client: DynamicClient,
+    hco_resource: HyperConverged,
+    golden_images_namespace: Namespace,
+    golden_images_data_import_crons: list[DataImportCron],
+    exclude_data_source_names: Collection[str] | None = None,
+) -> Iterator[None]:
     if hco_resource.instance.spec[ENABLE_COMMON_BOOT_IMAGE_IMPORT]:
         update_common_boot_image_import_spec(
             hco_resource=hco_resource,
@@ -365,12 +372,18 @@ def disable_common_boot_image_import_hco_spec(
             hco_resource=hco_resource,
             admin_client=admin_client,
             namespace=golden_images_namespace,
+            exclude_data_source_names=exclude_data_source_names,
         )
     else:
         yield
 
 
-def enable_common_boot_image_import_spec_wait_for_data_import_cron(hco_resource, admin_client, namespace):
+def enable_common_boot_image_import_spec_wait_for_data_import_cron(
+    hco_resource: HyperConverged,
+    admin_client: DynamicClient,
+    namespace: Namespace,
+    exclude_data_source_names: Collection[str] | None = None,
+) -> None:
     hco_namespace = Namespace(name=hco_resource.namespace)
     update_common_boot_image_import_spec(
         hco_resource=hco_resource,
@@ -383,6 +396,7 @@ def enable_common_boot_image_import_spec_wait_for_data_import_cron(hco_resource,
         admin_client=admin_client,
         namespace=namespace.name,
         consecutive_checks_count=1,
+        exclude_data_source_names=exclude_data_source_names,
     )
 
 

utilities/storage.py

@@ -2,7 +2,7 @@
 import math
 import os
 import shlex
-from collections.abc import Generator
+from collections.abc import Collection, Generator
 from contextlib import contextmanager
 from typing import Any
 
@@ -1118,9 +1118,12 @@ def get_data_sources_managed_by_data_import_cron(client: DynamicClient, namespac
 
 
 def verify_boot_sources_reimported(
-    admin_client: DynamicClient, namespace: str, consecutive_checks_count: int = 6
+    admin_client: DynamicClient,
+    namespace: str,
+    consecutive_checks_count: int = 6,
+    exclude_data_source_names: Collection[str] | None = None,
 ) -> bool:
-    """Verify all DataImportCron-managed DataSources reach Ready=True.
+    """Verify DataImportCron-managed DataSources reach Ready=True.
 
     Checks DataSources sequentially each with its own timeout. Stops on the first
     DataSource that does not become ready.
@@ -1129,12 +1132,18 @@ def verify_boot_sources_reimported(
         admin_client: Cluster admin client.
         namespace: Namespace containing the DataImportCron-managed DataSources.
         consecutive_checks_count: Consecutive Ready=True polls required for stability.
+        exclude_data_source_names: DataSources whose name is in this collection
+            are skipped (e.g. custom DIC templates without valid sources).
+            When None, all DIC-managed DataSources are verified.
 
     Returns:
-        True if all DataSources reached Ready=True otherwise false
+        True if all non-excluded DIC-managed DataSources reached Ready=True, otherwise False
     """
     try:
         for data_source in get_data_sources_managed_by_data_import_cron(client=admin_client, namespace=namespace):
+            if exclude_data_source_names is not None and data_source.name in exclude_data_source_names:
+                LOGGER.info(f"Skipping DataSource {data_source.name}: excluded from verification")
+                continue
             LOGGER.info(f"Waiting for DataSource {data_source.name} consistent ready status")
             utilities.infra.wait_for_consistent_resource_conditions(
                 dynamic_client=admin_client,

utilities/unittests/test_hco.py

@@ -766,7 +766,41 @@ def test_disable_when_enabled(self, mock_update_spec, mock_wait_deleted, mock_en
         except StopIteration:
             pass
 
-        mock_enable_spec.assert_called_once()
+        mock_enable_spec.assert_called_once_with(
+            hco_resource=mock_hco,
+            admin_client=mock_admin_client,
+            namespace=mock_namespace,
+            exclude_data_source_names=None,
+        )
+
+    @patch("utilities.hco.enable_common_boot_image_import_spec_wait_for_data_import_cron")
+    @patch("utilities.hco.wait_for_deleted_data_import_crons")
+    @patch("utilities.hco.update_common_boot_image_import_spec")
+    def test_disable_propagates_exclude_data_source_names(self, mock_update_spec, mock_wait_deleted, mock_enable_spec):
+        """Test that exclude_data_source_names is forwarded to the teardown call"""
+        mock_admin_client = MagicMock()
+        mock_hco = MagicMock()
+        mock_hco.instance.spec = {"enableCommonBootImageImport": True}
+        mock_namespace = MagicMock()
+        mock_dics = [MagicMock()]
+        exclude_names = {"custom-datasource"}
+
+        gen = disable_common_boot_image_import_hco_spec(
+            mock_admin_client, mock_hco, mock_namespace, mock_dics, exclude_data_source_names=exclude_names
+        )
+        next(gen)
+
+        try:
+            next(gen)
+        except StopIteration:
+            pass
+
+        mock_enable_spec.assert_called_once_with(
+            hco_resource=mock_hco,
+            admin_client=mock_admin_client,
+            namespace=mock_namespace,
+            exclude_data_source_names=exclude_names,
+        )
 
     @patch("utilities.hco.enable_common_boot_image_import_spec_wait_for_data_import_cron")
     @patch("utilities.hco.wait_for_deleted_data_import_crons")
@@ -828,6 +862,40 @@ def test_enable_spec(
             admin_client=mock_admin_client,
             namespace=mock_namespace.name,
             consecutive_checks_count=1,
+            exclude_data_source_names=None,
+        )
+
+    @patch("utilities.hco.wait_for_hco_conditions")
+    @patch("utilities.hco.wait_for_ssp_conditions")
+    @patch("utilities.hco.wait_for_at_least_one_auto_update_data_import_cron")
+    @patch("utilities.hco.update_common_boot_image_import_spec")
+    @patch("utilities.hco.Namespace")
+    @patch("utilities.hco.verify_boot_sources_reimported", return_value=True)
+    def test_enable_spec_propagates_exclude_data_source_names(
+        self,
+        mock_verify_boot,
+        mock_namespace_class,
+        mock_update_spec,
+        mock_wait_dic,
+        mock_wait_ssp,
+        mock_wait_hco,
+    ):
+        """Test that exclude_data_source_names is forwarded to verify_boot_sources_reimported"""
+        mock_hco = MagicMock()
+        mock_hco.namespace = "openshift-cnv"
+        mock_admin_client = MagicMock()
+        mock_namespace = MagicMock()
+        exclude_names = {"custom-datasource"}
+
+        enable_common_boot_image_import_spec_wait_for_data_import_cron(
+            mock_hco, mock_admin_client, mock_namespace, exclude_data_source_names=exclude_names
+        )
+
+        mock_verify_boot.assert_called_once_with(
+            admin_client=mock_admin_client,
+            namespace=mock_namespace.name,
+            consecutive_checks_count=1,
+            exclude_data_source_names=exclude_names,
         )