change: Split async job test and remove older test

Author: fege

tests/model_registry/async_job/conftest.py

@@ -51,7 +51,7 @@ def to_dict(self) -> None:
             self.res["spec"]["template"]["spec"]["volumes"] = self.volumes
 
 
-@pytest.fixture(scope="function")
+@pytest.fixture(scope="class")
 def s3_secret_for_async_job(
     admin_client: DynamicClient,
     service_account: ServiceAccount,
@@ -79,7 +79,7 @@ def s3_secret_for_async_job(
         yield secret
 
 
-@pytest.fixture(scope="function")
+@pytest.fixture(scope="class")
 def oci_secret_for_async_job(
     admin_client: DynamicClient,
     service_account: ServiceAccount,
@@ -112,7 +112,7 @@ def oci_secret_for_async_job(
         yield secret
 
 
-@pytest.fixture(scope="function")
+@pytest.fixture(scope="class")
 def model_sync_async_job(
     admin_client: DynamicClient,
     sa_token: str,
@@ -357,7 +357,7 @@ def oci_registry_host(oci_registry_route: Route) -> str:
     return oci_registry_route.instance.spec.host
 
 
-@pytest.fixture(scope="function")
+@pytest.fixture(scope="class")
 def create_test_data_in_minio_from_image(
     minio_service: Service,
     admin_client: DynamicClient,

tests/model_registry/async_job/test_async_upload_e2e.py

@@ -4,7 +4,6 @@
 import pytest
 from kubernetes.dynamic import DynamicClient
 from ocp_resources.job import Job
-from ocp_resources.route import Route
 from model_registry.types import ArtifactState, RegisteredModelState
 from tests.model_registry.async_job.constants import (
     ASYNC_UPLOAD_JOB_NAME,
@@ -60,32 +59,36 @@
     indirect=True,
 )
 class TestAsyncUploadE2E:
-    """Test for async upload job with real MinIO, OCI registry, and Model Registry"""
+    """RHOAIENG-29344: Test for async upload job with real MinIO, OCI registry, and Model Registry"""
 
-    def test_async_upload_job(
+    @pytest.mark.dependency(name="job_creation_and_pod_spawning")
+    def test_job_creation_and_pod_spawning(
         self: Self,
         admin_client: DynamicClient,
         model_sync_async_job: Job,
-        model_registry_client: list[ModelRegistryClient],
-        oci_registry_route: Route,
     ) -> None:
         """
-        RHOAIENG-29344: Test for async upload job execution using model from KSERVE_MINIO_IMAGE
-        Steps:
-            Create a model from KSERVE_MINIO_IMAGE
-            Create a job to upload the model to OCI registry
-            Verify the job completed successfully
-            Verify the model is uploaded to OCI registry
-            Verify the model is registered in model registry
+        Verify job creation and pod spawning
         """
-        LOGGER.info("Starting async upload job test with KSERVE_MINIO_IMAGE")
+        LOGGER.info("Verifying job creation and pod spawning")
+
         # Wait for job to create a pod
         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 OCI registry contains the uploaded artifact
-        registry_host = oci_registry_route.instance.spec.host
-        registry_url = f"http://{registry_host}"
+    @pytest.mark.dependency(name="oci_registry_verification", depends=["job_creation_and_pod_spawning"])
+    def test_oci_registry_verification(
+        self: Self,
+        oci_registry_host: str,
+    ) -> None:
+        """
+        Verify OCI registry upload
+        - Model manifest exists in OCI registry
+        - Manifest has correct structure and layers
+        """
+        LOGGER.info("Verifying OCI registry upload")
+
+        registry_url = f"http://{oci_registry_host}"
 
         LOGGER.info(f"Verifying artifact in OCI registry: {registry_url}/v2/{REPO_NAME}/manifests/{TAG}")
 
@@ -101,8 +104,20 @@ def test_async_upload_job(
         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
+    @pytest.mark.dependency(name="model_registry_verification", depends=["oci_registry_verification"])
+    def test_model_registry_verification(
+        self: Self,
+        model_registry_client: list[ModelRegistryClient],
+        oci_registry_host: str,
+    ) -> None:
+        """
+        Verify model registration
+        - Model is registered in model registry
+        - Model artifact has correct attributes
+        """
         LOGGER.info("Verifying model registry model and artifact")
+
+        # Verify model registry metadata was updated
         client = model_registry_client[0]
         model = client.get_registered_model(name=MODEL_NAME)
         assert model.state == RegisteredModelState.LIVE
@@ -112,7 +127,7 @@ def test_async_upload_job(
         # Validate model artifact attributes
         assert model_artifact.name == MODEL_NAME
         assert model_artifact.state == ArtifactState.LIVE
-        assert model_artifact.uri == f"oci://{registry_host}/{REPO_NAME}"
+        assert model_artifact.uri == f"oci://{oci_registry_host}/{REPO_NAME}"
         assert model_artifact.storage_key == MODEL_DATA["model_storage_key"]
         assert model_artifact.storage_path == MODEL_DATA["model_storage_path"]
 

tests/model_registry/async_job/test_basic_oci_registry.py

@@ -1,6 +1,4 @@
-import hashlib
 import requests
-import json
 from kubernetes.dynamic import DynamicClient
 from ocp_resources.job import Job
 from ocp_resources.pod import Pod
@@ -33,60 +31,6 @@ def get_latest_job_pod(admin_client: DynamicClient, job: Job) -> Pod:
     return latest_pod
 
 
-def push_blob_to_oci_registry(registry_url: str, data: bytes, repo: str = "test/simple-artifact") -> str:
-    """
-    Push a blob to an OCI registry.
-    https://specs.opencontainers.org/distribution-spec/?v=v1.0.0#pushing-blobs
-    POST to /v2/<repo>/blobs/uploads/ in order to initiate the upload
-    The response will contain a Location header that contains the upload URL
-    PUT to the Location URL with the data to be uploaded
-    """
-
-    blob_digest = f"sha256:{hashlib.sha256(data).hexdigest()}"
-
-    LOGGER.info(f"Pushing blob with digest: {blob_digest}")
-
-    upload_response = requests.post(f"{registry_url}/v2/{repo}/blobs/uploads/", timeout=10)
-    LOGGER.info(f"Blob upload initiation: {upload_response.status_code}")
-    assert upload_response.status_code == 202, f"Failed to initiate blob upload: {upload_response.status_code}"
-
-    upload_location = upload_response.headers.get("Location")
-    LOGGER.info(f"Upload location: {upload_location}")
-    base_url = f"{registry_url}{upload_location}"
-    upload_url = f"{base_url}?digest={blob_digest}"
-    response = requests.put(url=upload_url, data=data, headers={"Content-Type": "application/octet-stream"}, timeout=10)
-    assert response.status_code == 201, f"Failed to upload blob: {response.status_code}"
-    return blob_digest
-
-
-def create_manifest(blob_digest: str, config_json: str, config_digest: str, data: bytes) -> bytes:
-    """Create a manifest for an OCI registry."""
-
-    manifest = {
-        "schemaVersion": 2,
-        "mediaType": "application/vnd.oci.image.manifest.v1+json",
-        "config": {
-            "mediaType": "application/vnd.oci.image.config.v1+json",
-            "size": len(config_json),
-            "digest": config_digest,
-        },
-        "layers": [{"mediaType": "application/vnd.oci.image.layer.v1.tar", "size": len(data), "digest": blob_digest}],
-    }
-
-    return json.dumps(manifest, separators=(",", ":")).encode("utf-8")
-
-
-def push_manifest_to_oci_registry(registry_url: str, manifest: bytes, repo: str, tag: str) -> None:
-    """Push a manifest to an OCI registry."""
-    response = requests.put(
-        f"{registry_url}/v2/{repo}/manifests/{tag}",
-        data=manifest,
-        headers={"Content-Type": "application/vnd.oci.image.manifest.v1+json"},
-        timeout=10,
-    )
-    assert response.status_code == 201, f"Failed to push manifest: {response.status_code}"
-
-
 def pull_manifest_from_oci_registry(registry_url: str, repo: str, tag: str) -> dict:
     """Pull a manifest from an OCI registry."""
     response = requests.get(

tests/model_registry/async_job/utils.py

@@ -308,7 +308,7 @@ def model_registry_pod(admin_client: DynamicClient, model_registry_namespace: st
     return mr_pod[0]
 
 
-@pytest.fixture(scope="function")
+@pytest.fixture(scope="module")
 def sa_namespace(request: pytest.FixtureRequest, admin_client: DynamicClient) -> Generator[Namespace, None, None]:
     """
     Creates a namespace
@@ -321,7 +321,7 @@ def sa_namespace(request: pytest.FixtureRequest, admin_client: DynamicClient) ->
         yield ns
 
 
-@pytest.fixture(scope="function")
+@pytest.fixture(scope="class")
 def service_account(admin_client: DynamicClient, sa_namespace: Namespace) -> Generator[ServiceAccount, None, None]:
     """
     Creates a ServiceAccount.
@@ -332,7 +332,7 @@ def service_account(admin_client: DynamicClient, sa_namespace: Namespace) -> Gen
         yield sa
 
 
-@pytest.fixture(scope="function")
+@pytest.fixture(scope="class")
 def sa_token(service_account: ServiceAccount) -> str:
     """
     Retrieves a short-lived token for the ServiceAccount using 'oc create token'.
@@ -376,7 +376,7 @@ def sa_token(service_account: ServiceAccount) -> str:
         raise
 
 
-@pytest.fixture(scope="function")
+@pytest.fixture(scope="class")
 def mr_access_role(
     admin_client: DynamicClient,
     model_registry_namespace: str,
@@ -412,7 +412,7 @@ def mr_access_role(
         yield role
 
 
-@pytest.fixture(scope="function")
+@pytest.fixture(scope="class")
 def mr_access_role_binding(
     admin_client: DynamicClient,
     model_registry_namespace: str,