change: use wait_for_condition

Author: fege

tests/model_registry/async_job/conftest.py

@@ -230,6 +230,7 @@ def model_sync_async_job(
         ],
         teardown=teardown_resources,
     ) as job:
+        job.wait_for_condition(condition="Complete", status="True")
         yield job
 
 

tests/model_registry/async_job/test_async_upload_e2e.py

@@ -6,15 +6,12 @@
 from ocp_resources.job import Job
 from ocp_resources.route import Route
 from model_registry.types import ArtifactState, RegisteredModelState
-from timeout_sampler import TimeoutSampler, TimeoutExpiredError
 from tests.model_registry.async_job.constants import (
-    ASYNC_UPLOAD_IMAGE,
     ASYNC_UPLOAD_JOB_NAME,
 )
 from tests.model_registry.async_job.utils import (
     get_latest_job_pod,
     pull_manifest_from_oci_registry,
-    check_job_completion,
 )
 from tests.model_registry.constants import MODEL_DICT
 from utilities.constants import MinIo, OCIRegistry
@@ -65,10 +62,6 @@
 class TestAsyncUploadE2E:
     """Test for async upload job with real MinIO, OCI registry, and Model Registry"""
 
-    @pytest.mark.skipif(
-        ASYNC_UPLOAD_IMAGE.startswith("PLACEHOLDER"),
-        reason="Downstream image not yet available - job will fail to start",
-    )
     def test_async_upload_job(
         self: Self,
         admin_client: DynamicClient,
@@ -82,31 +75,6 @@ def test_async_upload_job(
         job_pod = get_latest_job_pod(admin_client=admin_client, job=model_sync_async_job)
         assert job_pod.name.startswith(ASYNC_UPLOAD_JOB_NAME)
 
-        # Verify job is created and configured correctly
-        assert model_sync_async_job.exists
-        LOGGER.info("Job created successfully")
-
-        # Wait for job to complete
-        try:
-            for sample in TimeoutSampler(
-                wait_timeout=300,  # 5 minutes
-                sleep=10,  # Check every 10 seconds
-                func=lambda: check_job_completion(admin_client=admin_client, job=model_sync_async_job),
-            ):
-                if sample:  # Job completed successfully
-                    break
-        except TimeoutExpiredError:
-            # Timeout reached - get final logs for debugging
-            LOGGER.error("Job timed out after 300 seconds")
-            try:
-                current_pod = get_latest_job_pod(admin_client=admin_client, job=model_sync_async_job)
-                pod_logs = current_pod.log()
-                LOGGER.error(f"Final pod logs: {pod_logs}")
-            except Exception as e:
-                LOGGER.error(f"Could not retrieve final pod logs: {e}")
-
-            pytest.fail("Async upload job did not complete within 300 seconds")
-
         # Verify OCI registry contains the uploaded artifact
         registry_host = oci_registry_route.instance.spec.host
         registry_url = f"http://{registry_host}"
@@ -123,18 +91,15 @@ def test_async_upload_job(
         # Verify the manifest has the expected structure
         assert "manifests" in manifest, "Manifest should contain manifests section"
         assert len(manifest["manifests"]) > 0, "Manifest should have at least one manifest"
-
         LOGGER.info(f"Manifest contains {len(manifest['manifests'])} layer(s)")
 
         # Verify model registry metadata was updated
         LOGGER.info("Verifying model registry model and artifact")
         client = model_registry_client[0]
         model = client.get_registered_model(name=MODEL_NAME)
-        LOGGER.info(f"Model: {model}")
         assert model.state == RegisteredModelState.LIVE
 
         model_artifact = client.get_model_artifact(name=MODEL_NAME, version=MODEL_DATA["model_version"])
-        LOGGER.info(f"Model artifact: {model_artifact}")
 
         # Validate model artifact attributes
         assert model_artifact.name == MODEL_NAME

tests/model_registry/async_job/utils.py

@@ -6,49 +6,12 @@
 from ocp_resources.pod import Pod
 from ocp_resources.service import Service
 from utilities.constants import MinIo
-
 from utilities.general import b64_encoded_string
 from simple_logger.logger import get_logger
 
 LOGGER = get_logger(name=__name__)
 
 
-def check_job_completion(admin_client: DynamicClient, job: Job) -> bool:
-    """
-    Check if job has completed successfully.
-
-    Args:
-        admin_client: Kubernetes dynamic client
-        job: Job object to check
-
-    Returns:
-        bool: True if job completed successfully, False otherwise
-    """
-    job_status = job.instance.status
-
-    if hasattr(job_status, "succeeded") and job_status.succeeded:
-        LOGGER.info("Job completed successfully")
-        return True
-
-    # Log current pod status for debugging
-    try:
-        current_pod = get_latest_job_pod(admin_client=admin_client, job=job)
-        pod_status = current_pod.instance.status
-        if hasattr(pod_status, "phase"):
-            if pod_status.phase == "Succeeded":
-                LOGGER.info(f"Pod {current_pod.name} succeeded")
-            elif pod_status.phase == "Failed":
-                LOGGER.warning(f"Pod {current_pod.name} failed, job may retry with new pod")
-            elif pod_status.phase == "Running":
-                LOGGER.info(f"Pod {current_pod.name} is running...")
-            else:
-                LOGGER.info(f"Pod {current_pod.name} in phase: {pod_status.phase}")
-    except Exception as e:
-        LOGGER.warning(f"Could not get pod status: {e}")
-
-    return False
-
-
 def get_latest_job_pod(admin_client: DynamicClient, job: Job) -> Pod:
     """Get the latest (most recently created) Pod created by a Job"""
     pods = list(