test: add source label search validation (#730)

Add comprehensive tests for model catalog search functionality including:
- Search by source label (Red Hat AI and Red Hat AI Validated)
- Invalid source label handling
- Combined source label queries
- Model matching with source filters

Refactor existing fixtures to support parameterized source selection
for better test reusability across different catalog sources.

Author: fege

tests/model_registry/model_catalog/conftest.py

@@ -147,10 +147,14 @@ def user_token_for_api_calls(
 
 
 @pytest.fixture(scope="class")
-def randomly_picked_model_from_default_catalog(
-    model_catalog_rest_url: list[str], user_token_for_api_calls: str
+def randomly_picked_model(
+    model_catalog_rest_url: list[str], user_token_for_api_calls: str, request: pytest.FixtureRequest
 ) -> dict[Any, Any]:
-    url = f"{model_catalog_rest_url[0]}models?source={REDHAT_AI_CATALOG_ID}&pageSize=100"
+    """Pick a random model"""
+    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),

tests/model_registry/model_catalog/constants.py

@@ -27,3 +27,7 @@
 }
 REDHAT_AI_CATALOG_ID: str = next(iter(DEFAULT_CATALOGS))
 DEFAULT_CATALOG_FILE: str = DEFAULT_CATALOGS[REDHAT_AI_CATALOG_ID]["properties"]["yamlCatalogPath"]
+VALIDATED_CATALOG_ID: str = tuple(DEFAULT_CATALOGS.keys())[1]
+
+REDHATI_AI_FILTER: str = "Red+Hat+AI"
+REDHATI_AI_VALIDATED_FILTER = "Red+Hat+AI+Validated"

tests/model_registry/model_catalog/test_default_model_catalog.py

@@ -156,41 +156,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_from_default_catalog: dict[Any, Any],
+        randomly_picked_model: 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_from_default_catalog}")
-        assert randomly_picked_model_from_default_catalog
+        LOGGER.info(f"picked model: {randomly_picked_model}")
+        assert randomly_picked_model
 
     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_from_default_catalog: dict[Any, Any],
+        randomly_picked_model: dict[Any, Any],
     ):
         """
         Validate a specific user can access get Model by name associated with a default source
         """
-        model_name = randomly_picked_model_from_default_catalog["name"]
+        model_name = randomly_picked_model["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_from_default_catalog, result))
+        differences = list(diff(randomly_picked_model, 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_from_default_catalog: dict[Any, Any],
+        randomly_picked_model: dict[Any, Any],
     ):
         """
         Validate a specific user can access get Model artifacts for model associated with default source
         """
-        model_name = randomly_picked_model_from_default_catalog["name"]
+        model_name = randomly_picked_model["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

@@ -0,0 +1,123 @@
+import pytest
+from dictdiffer import diff
+
+from simple_logger.logger import get_logger
+from typing import Self, Any
+from tests.model_registry.model_catalog.constants import (
+    REDHATI_AI_FILTER,
+    REDHATI_AI_VALIDATED_FILTER,
+    REDHAT_AI_CATALOG_ID,
+    VALIDATED_CATALOG_ID,
+)
+from tests.model_registry.model_catalog.utils import (
+    execute_get_command,
+)
+
+LOGGER = get_logger(name=__name__)
+pytestmark = [
+    pytest.mark.usefixtures("updated_dsc_component_state_scope_session", "model_registry_namespace", "test_idp_user")
+]
+
+
+class TestSearchModelCatalog:
+    @pytest.mark.smoke
+    def test_search_model_catalog_source_label(
+        self: Self, model_catalog_rest_url: list[str], model_registry_rest_headers: str
+    ):
+        """
+        RHOAIENG-33656: Validate search model catalog by source label
+        """
+
+        result = execute_get_command(
+            url=f"{model_catalog_rest_url[0]}models?sourceLabel={REDHATI_AI_FILTER}&pageSize=100",
+            headers=model_registry_rest_headers,
+        )
+        redhai_ai_filter_moldels_size = result["size"]
+
+        result = execute_get_command(
+            url=f"{model_catalog_rest_url[0]}models?sourceLabel={REDHATI_AI_VALIDATED_FILTER}&pageSize=100",
+            headers=model_registry_rest_headers,
+        )
+        redhai_ai_validated_filter_models_size = result["size"]
+
+        result = execute_get_command(
+            url=f"{model_catalog_rest_url[0]}models?pageSize=100", headers=model_registry_rest_headers
+        )
+        no_filtered_models_size = result["size"]
+
+        result = execute_get_command(
+            url=(
+                f"{model_catalog_rest_url[0]}models?"
+                f"sourceLabel={REDHATI_AI_VALIDATED_FILTER},{REDHATI_AI_FILTER}&pageSize=100"
+            ),
+            headers=model_registry_rest_headers,
+        )
+        both_filtered_models_size = result["size"]
+
+        assert no_filtered_models_size == both_filtered_models_size
+        assert redhai_ai_filter_moldels_size + redhai_ai_validated_filter_models_size == both_filtered_models_size
+
+    def test_search_model_catalog_invalid_source_label(
+        self: Self, model_catalog_rest_url: list[str], model_registry_rest_headers: str
+    ):
+        """
+        RHOAIENG-33656:
+        Validate search model catalog by invalid source label
+        """
+
+        result = execute_get_command(
+            url=f"{model_catalog_rest_url[0]}/models?sourceLabel=null&pageSize=100", headers=model_registry_rest_headers
+        )
+        null_size = result["size"]
+
+        result = execute_get_command(
+            url=f"{model_catalog_rest_url[0]}/models?sourceLabel=invalid&pageSize=100",
+            headers=model_registry_rest_headers,
+        )
+        invalid_size = result["size"]
+
+        assert null_size == invalid_size == 0, (
+            "Expected 0 models for null and invalid source label found {null_size} and {invalid_size}"
+        )
+
+    @pytest.mark.parametrize(
+        "randomly_picked_model,source_filter",
+        [
+            pytest.param(
+                {"source": VALIDATED_CATALOG_ID},
+                REDHATI_AI_VALIDATED_FILTER,
+                id="test_search_model_catalog_redhat_ai_validated",
+            ),
+            pytest.param(
+                {"source": REDHAT_AI_CATALOG_ID}, REDHATI_AI_FILTER, id="test_search_model_catalog_redhat_ai_default"
+            ),
+        ],
+        indirect=["randomly_picked_model"],
+    )
+    def test_search_model_catalog_match(
+        self: Self,
+        model_catalog_rest_url: list[str],
+        model_registry_rest_headers: str,
+        randomly_picked_model: dict[Any, Any],
+        source_filter: str,
+    ):
+        """
+        RHOAIENG-33656: Validate search model catalog by match
+        """
+        random_model = randomly_picked_model
+        random_model_name = random_model["name"]
+        LOGGER.info(f"random_model_name: {random_model_name}")
+        result = execute_get_command(
+            url=(
+                f"{model_catalog_rest_url[0]}/models?"
+                f"sourceLabel={source_filter}&"
+                f"filterQuery=name='{random_model_name}'&pageSize=100"
+            ),
+            headers=model_registry_rest_headers,
+        )
+        assert random_model_name == result["items"][0]["name"]
+        assert result["size"] == 1
+
+        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")