test: add artifact type filtering validation

Add comprehensive test suite for model catalog artifact filtering by type:
- Validate artifact filtering by model-artifact and metrics-artifact types
- Test error handling for invalid artifact types
- Test multiple artifact type filtering (currently failing - RHOAIENG-36938)
- Refactor randomly_picked_model fixture to support header type selection
- Add dynamic pagination helper for artifact fetching
- Rename get_models_from_api to get_models_from_catalog_api for clarity

Author: fege

tests/model_registry/model_catalog/conftest.py

@@ -119,22 +119,43 @@ def user_token_for_api_calls(
         raise RuntimeError(f"Unknown user type: {user}")
 
 
-@pytest.fixture(scope="class")
-def randomly_picked_model(
-    model_catalog_rest_url: list[str], user_token_for_api_calls: str, request: pytest.FixtureRequest
+@pytest.fixture(scope="function")
+def randomly_picked_model_from_catalog_api_by_source(
+    model_catalog_rest_url: list[str],
+    user_token_for_api_calls: str,
+    model_registry_rest_headers: dict[str, str],
+    request: pytest.FixtureRequest,
 ) -> dict[Any, Any]:
-    """Pick a random model"""
+    """Pick a random model from a specific catalog (function-scoped for test isolation)
+
+    Supports parameterized headers via 'header_type':
+    - 'user_token': Uses user_token_for_api_calls (default for user-specific tests)
+    - 'registry': Uses model_registry_rest_headers (for catalog/registry tests)
+
+    Accepts 'catalog_id' or 'source' (alias) to specify the catalog.
+    """
     param = getattr(request, "param", {})
-    source = param.get("source", REDHAT_AI_CATALOG_ID)
-    LOGGER.info(f"Picking random model from {source}")
-    url = f"{model_catalog_rest_url[0]}models?source={source}&pageSize=100"
-    result = execute_get_command(
-        url=url,
-        headers=get_rest_headers(token=user_token_for_api_calls),
-    )["items"]
-    assert result, f"Expected Default models to be present. Actual: {result}"
-    LOGGER.info(f"{len(result)} models found")
-    return random.choice(seq=result)
+    # Support both 'catalog_id' and 'source' for backward compatibility
+    catalog_id = param.get("catalog_id") or param.get("source", REDHAT_AI_CATALOG_ID)
+    header_type = param.get("header_type", "user_token")
+
+    # Select headers based on header_type
+    if header_type == "registry":
+        headers = model_registry_rest_headers
+    else:
+        headers = get_rest_headers(token=user_token_for_api_calls)
+
+    LOGGER.info(f"Picking random model from catalog: {catalog_id} with header_type: {header_type}")
+
+    models_response = execute_get_command(
+        url=f"{model_catalog_rest_url[0]}models?source={catalog_id}&pageSize=100",
+        headers=headers,
+    )
+    models = models_response.get("items", [])
+    assert models, f"No models found for catalog: {catalog_id}"
+    LOGGER.info(f"{len(models)} models found in catalog {catalog_id}")
+
+    return random.choice(seq=models)
 
 
 @pytest.fixture(scope="class")

tests/model_registry/model_catalog/constants.py

@@ -36,3 +36,5 @@
 
 REDHAT_AI_FILTER: str = "Red+Hat+AI"
 REDHAT_AI_VALIDATED_FILTER = "Red+Hat+AI+Validated"
+MODEL_ARTIFACT_TYPE: str = "model-artifact"
+METRICS_ARTIFACT_TYPE: str = "metrics-artifact"

tests/model_registry/model_catalog/test_default_model_catalog.py

@@ -188,41 +188,41 @@ def test_model_catalog_default_catalog_sources(
     def test_model_default_catalog_get_models_by_source(
         self: Self,
         model_catalog_rest_url: list[str],
-        randomly_picked_model: dict[Any, Any],
+        randomly_picked_model_from_catalog_api_by_source: dict[Any, Any],
     ):
         """
         Validate a specific user can access models api for model catalog associated with a default source
         """
-        LOGGER.info(f"picked model: {randomly_picked_model}")
-        assert randomly_picked_model
+        LOGGER.info(f"picked model: {randomly_picked_model_from_catalog_api_by_source}")
+        assert randomly_picked_model_from_catalog_api_by_source
 
     def test_model_default_catalog_get_model_by_name(
         self: Self,
         model_catalog_rest_url: list[str],
         user_token_for_api_calls: str,
-        randomly_picked_model: dict[Any, Any],
+        randomly_picked_model_from_catalog_api_by_source: dict[Any, Any],
     ):
         """
         Validate a specific user can access get Model by name associated with a default source
         """
-        model_name = randomly_picked_model["name"]
+        model_name = randomly_picked_model_from_catalog_api_by_source["name"]
         result = execute_get_command(
             url=f"{model_catalog_rest_url[0]}sources/{REDHAT_AI_CATALOG_ID}/models/{model_name}",
             headers=get_rest_headers(token=user_token_for_api_calls),
         )
-        differences = list(diff(randomly_picked_model, result))
+        differences = list(diff(randomly_picked_model_from_catalog_api_by_source, result))
         assert not differences, f"Expected no differences in model information for {model_name}: {differences}"
 
     def test_model_default_catalog_get_model_artifact(
         self: Self,
         model_catalog_rest_url: list[str],
         user_token_for_api_calls: str,
-        randomly_picked_model: dict[Any, Any],
+        randomly_picked_model_from_catalog_api_by_source: dict[Any, Any],
     ):
         """
         Validate a specific user can access get Model artifacts for model associated with default source
         """
-        model_name = randomly_picked_model["name"]
+        model_name = randomly_picked_model_from_catalog_api_by_source["name"]
         result = execute_get_command(
             url=f"{model_catalog_rest_url[0]}sources/{REDHAT_AI_CATALOG_ID}/models/{model_name}/artifacts",
             headers=get_rest_headers(token=user_token_for_api_calls),

tests/model_registry/model_catalog/test_model_search.py

@@ -8,10 +8,14 @@
     REDHAT_AI_VALIDATED_FILTER,
     REDHAT_AI_CATALOG_ID,
     VALIDATED_CATALOG_ID,
+    MODEL_ARTIFACT_TYPE,
+    METRICS_ARTIFACT_TYPE,
 )
 from tests.model_registry.model_catalog.utils import (
-    get_models_from_api,
+    get_models_from_catalog_api,
+    fetch_all_artifacts_with_dynamic_paging,
 )
+from kubernetes.dynamic.exceptions import ResourceNotFoundError
 
 LOGGER = get_logger(name=__name__)
 pytestmark = [
@@ -28,20 +32,20 @@ def test_search_model_catalog_source_label(
         RHOAIENG-33656: Validate search model catalog by source label
         """
 
-        redhat_ai_filter_moldels_size = get_models_from_api(
+        redhat_ai_filter_moldels_size = get_models_from_catalog_api(
             model_catalog_rest_url=model_catalog_rest_url,
             model_registry_rest_headers=model_registry_rest_headers,
             source_label=REDHAT_AI_FILTER,
         )["size"]
-        redhat_ai_validated_filter_models_size = get_models_from_api(
+        redhat_ai_validated_filter_models_size = get_models_from_catalog_api(
             model_catalog_rest_url=model_catalog_rest_url,
             model_registry_rest_headers=model_registry_rest_headers,
             source_label=REDHAT_AI_VALIDATED_FILTER,
         )["size"]
-        no_filtered_models_size = get_models_from_api(
+        no_filtered_models_size = get_models_from_catalog_api(
             model_catalog_rest_url=model_catalog_rest_url, model_registry_rest_headers=model_registry_rest_headers
         )["size"]
-        both_filtered_models_size = get_models_from_api(
+        both_filtered_models_size = get_models_from_catalog_api(
             model_catalog_rest_url=model_catalog_rest_url,
             model_registry_rest_headers=model_registry_rest_headers,
             source_label=f"{REDHAT_AI_VALIDATED_FILTER},{REDHAT_AI_FILTER}",
@@ -58,13 +62,13 @@ def test_search_model_catalog_invalid_source_label(
         Validate search model catalog by invalid source label
         """
 
-        null_size = get_models_from_api(
+        null_size = get_models_from_catalog_api(
             model_catalog_rest_url=model_catalog_rest_url,
             model_registry_rest_headers=model_registry_rest_headers,
             source_label="null",
         )["size"]
 
-        invalid_size = get_models_from_api(
+        invalid_size = get_models_from_catalog_api(
             model_catalog_rest_url=model_catalog_rest_url,
             model_registry_rest_headers=model_registry_rest_headers,
             source_label="invalid",
@@ -75,33 +79,35 @@ def test_search_model_catalog_invalid_source_label(
         )
 
     @pytest.mark.parametrize(
-        "randomly_picked_model,source_filter",
+        "randomly_picked_model_from_catalog_api_by_source,source_filter",
         [
             pytest.param(
-                {"source": VALIDATED_CATALOG_ID},
+                {"source": VALIDATED_CATALOG_ID, "header_type": "registry"},
                 REDHAT_AI_VALIDATED_FILTER,
                 id="test_search_model_catalog_redhat_ai_validated",
             ),
             pytest.param(
-                {"source": REDHAT_AI_CATALOG_ID}, REDHAT_AI_FILTER, id="test_search_model_catalog_redhat_ai_default"
+                {"source": REDHAT_AI_CATALOG_ID, "header_type": "registry"},
+                REDHAT_AI_FILTER,
+                id="test_search_model_catalog_redhat_ai_default",
             ),
         ],
-        indirect=["randomly_picked_model"],
+        indirect=["randomly_picked_model_from_catalog_api_by_source"],
     )
     def test_search_model_catalog_match(
         self: Self,
         model_catalog_rest_url: list[str],
         model_registry_rest_headers: dict[str, str],
-        randomly_picked_model: dict[Any, Any],
+        randomly_picked_model_from_catalog_api_by_source: dict[Any, Any],
         source_filter: str,
     ):
         """
         RHOAIENG-33656: Validate search model catalog by match
         """
-        random_model = randomly_picked_model
+        random_model = randomly_picked_model_from_catalog_api_by_source
         random_model_name = random_model["name"]
         LOGGER.info(f"random_model_name: {random_model_name}")
-        result = get_models_from_api(
+        result = get_models_from_catalog_api(
             model_catalog_rest_url=model_catalog_rest_url,
             model_registry_rest_headers=model_registry_rest_headers,
             source_label=source_filter,
@@ -113,3 +119,180 @@ def test_search_model_catalog_match(
         differences = list(diff(random_model, result["items"][0]))
         assert not differences, f"Expected no differences in model information for {random_model_name}: {differences}"
         LOGGER.info("Model information matches")
+
+
+# All the tests in this class are failing for RHOAIENG-36938, there are two problems:
+# 1. The filter parameter is setup to use artifact_type instead of artifactType
+# 2. The filter with multiple artifact types is not working as expected
+class TestSearchModelArtifact:
+    @pytest.mark.parametrize(
+        "randomly_picked_model_from_catalog_api_by_source, artifact_type",
+        [
+            pytest.param(
+                {"catalog_id": REDHAT_AI_CATALOG_ID, "header_type": "registry"},
+                MODEL_ARTIFACT_TYPE,
+                id="redhat_ai_model_artifact",
+            ),
+            pytest.param(
+                {"catalog_id": REDHAT_AI_CATALOG_ID, "header_type": "registry"},
+                METRICS_ARTIFACT_TYPE,
+                id="redhat_ai_metrics_artifact",
+            ),
+            pytest.param(
+                {"catalog_id": VALIDATED_CATALOG_ID, "header_type": "registry"},
+                MODEL_ARTIFACT_TYPE,
+                id="validated_model_artifact",
+            ),
+            pytest.param(
+                {"catalog_id": VALIDATED_CATALOG_ID, "header_type": "registry"},
+                METRICS_ARTIFACT_TYPE,
+                id="validated_metrics_artifact",
+            ),
+        ],
+        indirect=["randomly_picked_model_from_catalog_api_by_source"],
+    )
+    def test_validate_model_artifacts_by_artifact_type(
+        self: Self,
+        model_catalog_rest_url: list[str],
+        model_registry_rest_headers: dict[str, str],
+        randomly_picked_model_from_catalog_api_by_source: dict[Any, Any],
+        artifact_type: str,
+    ):
+        """
+        RHOAIENG-33659: Validates that the model artifacts returned by the artifactType filter
+        match the complete set of artifacts for a random model.
+        """
+        model_name = randomly_picked_model_from_catalog_api_by_source.get("name")
+        catalog_id = randomly_picked_model_from_catalog_api_by_source.get("source_id")
+        assert model_name and catalog_id, "Model name or catalog ID not found in random model"
+        LOGGER.info(f"Testing model '{model_name}' from catalog '{catalog_id}' for artifact type '{artifact_type}'")
+
+        # Fetch all artifacts with dynamic page size adjustment
+        all_model_artifacts = fetch_all_artifacts_with_dynamic_paging(
+            url_with_pagesize=f"{model_catalog_rest_url[0]}sources/{catalog_id}/models/{model_name}/artifacts?pageSize",
+            headers=model_registry_rest_headers,
+            page_size=100,
+        )["items"]
+
+        # Fetch filtered artifacts by type with dynamic page size adjustment
+        artifact_type_artifacts = fetch_all_artifacts_with_dynamic_paging(
+            url_with_pagesize=(
+                f"{model_catalog_rest_url[0]}sources/{catalog_id}/models/{model_name}/artifacts?"
+                f"artifactType={artifact_type}&pageSize"
+            ),
+            headers=model_registry_rest_headers,
+            page_size=50,
+        )["items"]
+
+        # Create lookup for validation
+        all_artifacts_by_id = {artifact["id"]: artifact for artifact in all_model_artifacts}
+
+        # Verify all filtered artifacts exist
+        for artifact in artifact_type_artifacts:
+            artifact_id = artifact["id"]
+            assert artifact_id in all_artifacts_by_id, (
+                f"Filtered artifact {artifact_id} not found in complete artifact list for {model_name}"
+            )
+
+            differences = list(diff(artifact, all_artifacts_by_id[artifact_id]))
+            assert not differences, f"Artifact {artifact_id} mismatch for {model_name}: {differences}"
+
+        # Verify the filter didn't miss any artifacts of the type
+        artifacts_of_type_in_all = [
+            artifact for artifact in all_model_artifacts if artifact.get("artifactType") == artifact_type
+        ]
+        assert len(artifact_type_artifacts) == len(artifacts_of_type_in_all), (
+            f"Filter returned {len(artifact_type_artifacts)} {artifact_type} artifacts, "
+            f"but found {len(artifacts_of_type_in_all)} in complete list for {model_name}"
+        )
+
+        LOGGER.info(f"Validated {len(artifact_type_artifacts)} {artifact_type} artifacts for {model_name}")
+
+    @pytest.mark.parametrize(
+        "randomly_picked_model_from_catalog_api_by_source",
+        [
+            pytest.param(
+                {"header_type": "registry"},
+                id="invalid_artifact_type",
+            ),
+        ],
+        indirect=["randomly_picked_model_from_catalog_api_by_source"],
+    )
+    def test_error_handled_for_invalid_artifact_type(
+        self: Self,
+        model_catalog_rest_url: list[str],
+        model_registry_rest_headers: dict[str, str],
+        randomly_picked_model_from_catalog_api_by_source: dict[Any, Any],
+    ):
+        """
+        RHOAIENG-33659: Validates that the API returns the correct error when an invalid artifactType
+        is provided regardless of catalog or model.
+        """
+        model_name = randomly_picked_model_from_catalog_api_by_source.get("name")
+        catalog_id = randomly_picked_model_from_catalog_api_by_source.get("source_id")
+        assert model_name and catalog_id, "Model name or catalog ID not found in random model"
+
+        invalid_artifact_type = "invalid"
+
+        with pytest.raises(ResourceNotFoundError, match=f"unsupported catalog artifact type: {invalid_artifact_type}"):
+            fetch_all_artifacts_with_dynamic_paging(
+                url_with_pagesize=(
+                    f"{model_catalog_rest_url[0]}sources/{catalog_id}/models/{model_name}/artifacts?"
+                    f"artifactType={invalid_artifact_type}&pageSize"
+                ),
+                headers=model_registry_rest_headers,
+                page_size=1,
+            )
+
+        LOGGER.info(f"Successfully validated that invalid artifact type '{invalid_artifact_type}' raises an error")
+
+    @pytest.mark.parametrize(
+        "randomly_picked_model_from_catalog_api_by_source",
+        [
+            pytest.param(
+                {"catalog_id": REDHAT_AI_CATALOG_ID, "header_type": "registry"},
+                id="redhat_ai_catalog",
+            ),
+            pytest.param(
+                {"catalog_id": VALIDATED_CATALOG_ID, "header_type": "registry"},
+                id="validated_catalog",
+            ),
+        ],
+        indirect=["randomly_picked_model_from_catalog_api_by_source"],
+    )
+    def test_multiple_artifact_type_filtering(
+        self: Self,
+        model_catalog_rest_url: list[str],
+        model_registry_rest_headers: dict[str, str],
+        randomly_picked_model_from_catalog_api_by_source: dict[Any, Any],
+    ):
+        """
+        RHOAIENG-33659: Validates that the API returns all artifacts of a random model
+        when filtering by multiple artifact types.
+        """
+        model_name = randomly_picked_model_from_catalog_api_by_source.get("name")
+        catalog_id = randomly_picked_model_from_catalog_api_by_source.get("source_id")
+        assert model_name and catalog_id, "Model name or catalog ID not found in random model"
+        LOGGER.info(f"Testing model '{model_name}' from catalog '{catalog_id}' for multiple artifact types")
+        artifact_types = f"{METRICS_ARTIFACT_TYPE},{MODEL_ARTIFACT_TYPE}"
+        # Fetch all artifacts with dynamic page size adjustment
+        all_model_artifacts = fetch_all_artifacts_with_dynamic_paging(
+            url_with_pagesize=f"{model_catalog_rest_url[0]}sources/{catalog_id}/models/{model_name}/artifacts?pageSize",
+            headers=model_registry_rest_headers,
+            page_size=100,
+        )["items"]
+
+        # Fetch filtered artifacts by type with dynamic page size adjustment
+        artifact_type_artifacts = fetch_all_artifacts_with_dynamic_paging(
+            url_with_pagesize=(
+                f"{model_catalog_rest_url[0]}sources/{catalog_id}/models/{model_name}/artifacts?"
+                f"artifactType={artifact_types}&pageSize"
+            ),
+            headers=model_registry_rest_headers,
+            page_size=100,
+        )["items"]
+
+        assert len(artifact_type_artifacts) == len(all_model_artifacts), (
+            f"Filter returned {len(artifact_type_artifacts)} artifacts, "
+            f"but found {len(all_model_artifacts)} in complete list for {model_name}"
+        )