Fix race condition in Kueue tests by waiting for CRDs to be available (opendatahub-io#1017)

* add function with retry logic to make sure crd are available

* refactor ensure_kueue_unmanaged_in_dsc fixture

* add comments and improve docstring

* fix precommit issues

* add false positive fix for dsc readiness check

* remove .get() calls for dsc

Author: threcc

tests/model_serving/model_server/conftest.py

@@ -1,3 +1,4 @@
+from contextlib import ExitStack
 from typing import Any, Generator, Dict
 
 import pytest
@@ -23,6 +24,7 @@
     LocalQueue,
     ClusterQueue,
     ResourceFlavor,
+    wait_for_kueue_crds_available,
 )
 from pytest_testconfig import config as py_config
 from simple_logger.logger import get_logger
@@ -38,6 +40,10 @@
 from utilities.constants import (
     ModelAndFormat,
 )
+from utilities.data_science_cluster_utils import (
+    get_dsc_ready_condition,
+    wait_for_dsc_reconciliation,
+)
 from utilities.inference_utils import create_isvc
 from utilities.infra import (
     s3_endpoint_secret,
@@ -418,6 +424,7 @@ def gpu_model_car_inference_service(
 
 # Kueue Fixtures
 def _is_kueue_operator_installed(admin_client: DynamicClient) -> bool:
+    """Check if the Kueue operator is installed and ready."""
     try:
         csvs = list(
             ClusterServiceVersion.get(
@@ -438,35 +445,39 @@ def _is_kueue_operator_installed(admin_client: DynamicClient) -> bool:
 def ensure_kueue_unmanaged_in_dsc(
     admin_client: DynamicClient, dsc_resource: DataScienceCluster
 ) -> Generator[None, Any, None]:
+    """Set DSC Kueue to Unmanaged and wait for CRDs to be available."""
     try:
         if not _is_kueue_operator_installed(admin_client):
             pytest.skip("Kueue operator is not installed, skipping Kueue tests")
 
-        dsc_resource.get()
+        # Check current Kueue state
         kueue_management_state = dsc_resource.instance.spec.components[DscComponents.KUEUE].managementState
 
-        if kueue_management_state == DscComponents.ManagementState.UNMANAGED:
-            LOGGER.info("Kueue is already Unmanaged in DSC, proceeding with tests")
-            yield
-        else:
-            LOGGER.info(f"Kueue management state is {kueue_management_state}, updating to Unmanaged")
-            dsc_dict = {
-                "spec": {
-                    "components": {DscComponents.KUEUE: {"managementState": DscComponents.ManagementState.UNMANAGED}}
+        with ExitStack() as stack:
+            # Only patch if Kueue is not already Unmanaged
+            if kueue_management_state != DscComponents.ManagementState.UNMANAGED:
+                LOGGER.info(f"Patching Kueue from {kueue_management_state} to Unmanaged")
+                # Read timestamp BEFORE applying patch
+                ready_condition = get_dsc_ready_condition(dsc=dsc_resource)
+                pre_patch_time = ready_condition.get("lastTransitionTime") if ready_condition else None
+
+                dsc_dict = {
+                    "spec": {
+                        "components": {
+                            DscComponents.KUEUE: {"managementState": DscComponents.ManagementState.UNMANAGED}
+                        }
+                    }
                 }
-            }
+                stack.enter_context(cm=ResourceEditor(patches={dsc_resource: dsc_dict}))
+
+                # Wait for DSC to reconcile the patch
+                wait_for_dsc_reconciliation(dsc=dsc_resource, baseline_time=pre_patch_time)
+            else:
+                LOGGER.info("Kueue already Unmanaged, no patch needed")
 
-            with ResourceEditor(patches={dsc_resource: dsc_dict}):
-                LOGGER.info("Updated Kueue to Unmanaged, waiting for DSC to be ready")
-                dsc_resource.wait_for_condition(condition="Ready", status="True", timeout=300)
-                LOGGER.info("DSC is ready, proceeding with tests")
-                yield
-
-            LOGGER.info(f"Restoring Kueue management state to {kueue_management_state}")
-            restore_dict = {"spec": {"components": {DscComponents.KUEUE: {"managementState": kueue_management_state}}}}
-            with ResourceEditor(patches={dsc_resource: restore_dict}):
-                dsc_resource.wait_for_condition(condition="Ready", status="True", timeout=300)
-                LOGGER.info("Restored Kueue management state")
+            # Always wait for Kueue CRDs and controller pods (regardless of patch)
+            wait_for_kueue_crds_available(client=admin_client)
+            yield
 
     except (AttributeError, KeyError) as e:
         pytest.skip(f"Kueue component not found in DSC: {e}")

utilities/data_science_cluster_utils.py

@@ -4,6 +4,7 @@
 from ocp_resources.data_science_cluster import DataScienceCluster
 from ocp_resources.resource import ResourceEditor
 from simple_logger.logger import get_logger
+from timeout_sampler import retry
 
 from utilities.constants import DscComponents
 
@@ -52,3 +53,61 @@ def update_components_in_dsc(
 
     else:
         yield dsc
+
+
+def get_dsc_ready_condition(dsc: DataScienceCluster) -> dict[str, Any] | None:
+    """Get DSC Ready condition.
+
+    Args:
+        dsc: DataScienceCluster resource
+
+    Returns:
+        The Ready condition dict (with 'status', 'lastTransitionTime', etc.), or None if not found
+    """
+    return next(
+        (
+            condition
+            for condition in dsc.instance.status.conditions or []
+            if condition.type == DataScienceCluster.Condition.READY
+        ),
+        None,
+    )
+
+
+@retry(wait_timeout=300, sleep=5)
+def wait_for_dsc_reconciliation(dsc: DataScienceCluster, baseline_time: str | None) -> bool:
+    """Wait for DSC to reconcile after a ResourceEditor patch.
+
+    This function prevents false positives where DSC reports Ready=True immediately
+    after a patch, before actual reconciliation begins. It waits for:
+    1. lastTransitionTime to change (reconciliation started)
+    2. Ready=True condition (reconciliation completed)
+
+    Args:
+        dsc: DataScienceCluster resource
+        baseline_time: The Ready condition lastTransitionTime before the patch, or None if not found
+
+    Returns:
+        True when DSC has reconciled and is Ready
+    """
+    ready_condition = get_dsc_ready_condition(dsc=dsc)
+    current_time = ready_condition.get("lastTransitionTime") if ready_condition else None
+    dsc_reconciling = current_time != baseline_time
+    dsc_ready = ready_condition and ready_condition.get("status") == DataScienceCluster.Condition.Status.TRUE
+
+    # Still waiting for reconciliation to start (timestamp unchanged)
+    if not dsc_reconciling:
+        LOGGER.info(f"Waiting for DSC reconciliation to start (baseline: {baseline_time or 'None'})...")
+        return False
+
+    # Timestamp changed but DSC is not Ready yet
+    if not dsc_ready:
+        LOGGER.info(f"DSC reconciliation in progress (timestamp: {current_time or 'None'}), waiting for Ready=True...")
+        return False
+
+    # DSC Reconciled: timestamp changed AND Ready=True
+    LOGGER.info(
+        f"DSC reconciliation complete: timestamp changed from {baseline_time or 'None'} "
+        f"to {current_time or 'None'} and Ready=True"
+    )
+    return True

utilities/kueue_utils.py

@@ -3,6 +3,11 @@
 from kubernetes.dynamic import DynamicClient
 from ocp_resources.resource import NamespacedResource, Resource, MissingRequiredArgumentError
 from ocp_resources.pod import Pod
+from simple_logger.logger import get_logger
+from timeout_sampler import retry
+from utilities.constants import Timeout
+
+LOGGER = get_logger(name=__name__)
 
 
 class ResourceFlavor(Resource):
@@ -170,3 +175,46 @@ def check_gated_pods_and_running_pods(
             ):
                 gated_pods += 1
     return running_pods, gated_pods
+
+
+@retry(
+    wait_timeout=Timeout.TIMEOUT_4MIN,
+    sleep=5,
+)
+def wait_for_kueue_crds_available(client: DynamicClient) -> bool:
+    """Wait for Kueue CRDs and controller to be fully available.
+
+    This function waits for:
+    1. Kueue CRDs to be registered in the API server
+    2. kueue-controller-manager pods to be Ready (needed for webhooks/admission control)
+
+    Raises:
+        TimeoutExpiredError: If CRDs or controller are not available within the timeout period.
+
+    Returns:
+        True when CRDs are available and controller is ready.
+    """
+    # Check if CRDs are registered (raises exception if not, then will @retry)
+    list(ResourceFlavor.get(client=client))
+
+    # Check kueue-controller-manager pods exist and are ready
+    pods = list(
+        Pod.get(
+            label_selector="control-plane=controller-manager,app.kubernetes.io/name=kueue",
+            namespace="openshift-kueue-operator",
+            client=client,
+        )
+    )
+    all_pods_ready = pods and all(
+        any(
+            condition.type == Pod.Condition.READY and condition.status == Pod.Condition.Status.TRUE
+            for condition in pod.instance.status.conditions or []
+        )
+        for pod in pods
+    )
+    if not all_pods_ready:
+        LOGGER.info("Kueue controller pods not ready yet, retrying...")
+        return False
+
+    LOGGER.info(f"Kueue is ready: CRDs available and {len(pods)} controller pod(s) running")
+    return True