test: add new tests for embedding models

Signed-off-by: Debarati Basu-Nag <dbasunag@redhat.com>

Author: dbasunag

tests/model_registry/model_catalog/search/conftest.py

@@ -0,0 +1,18 @@
+from typing import Any
+
+import pytest
+
+from tests.model_registry.model_catalog.utils import get_models_from_catalog_api
+
+
+@pytest.fixture(scope="class")
+def embedding_models_response(
+    model_catalog_rest_url: list[str],
+    model_registry_rest_headers: dict[str, str],
+) -> dict[str, Any]:
+    """Fetch models filtered by tasks='text-embedding' via filterQuery"""
+    return get_models_from_catalog_api(
+        model_catalog_rest_url=model_catalog_rest_url,
+        model_registry_rest_headers=model_registry_rest_headers,
+        additional_params="&filterQuery=tasks='text-embedding'",
+    )

tests/model_registry/model_catalog/search/test_model_search.py

@@ -7,6 +7,7 @@
 from simple_logger.logger import get_logger
 
 from tests.model_registry.model_catalog.constants import (
+    OTHER_MODELS_CATALOG_ID,
     REDHAT_AI_CATALOG_ID,
     REDHAT_AI_CATALOG_NAME,
     REDHAT_AI_VALIDATED_UNESCAPED_CATALOG_NAME,
@@ -438,3 +439,48 @@ def test_filter_query_advanced_model_search(
         LOGGER.info(
             f"Advanced {logic_type.upper()} filter validation completed for {len(models_from_filter_query)} models"
         )
+
+
+@pytest.mark.install
+@pytest.mark.post_upgrade
+class TestEmbeddingModelSearch:
+    def test_filter_query_by_text_embedding_task(
+        self: Self,
+        embedding_models_response: dict[str, Any],
+    ):
+        """
+        Validate filterQuery with tasks='text-embedding' returns models
+        """
+        number_of_models = embedding_models_response.get("size", 0)
+        LOGGER.info(f"Found number of embedding models: {number_of_models}")
+        assert number_of_models > 0, "Expected at least one model with tasks='text-embedding'"
+
+    def test_embedding_models_source_id(
+        self: Self,
+        embedding_models_response: dict[str, Any],
+    ):
+        """
+        Validate all embedding models belong to the Other Models source
+        """
+        mismatched_models = [
+            f"{model['name']} (source_id={model['source_id']})"
+            for model in embedding_models_response.get("items", [])
+            if model["source_id"] != OTHER_MODELS_CATALOG_ID
+        ]
+        assert not mismatched_models, (
+            f"Models with unexpected source_id (expected '{OTHER_MODELS_CATALOG_ID}'): {mismatched_models}"
+        )
+
+    def test_embedding_models_have_text_embedding_task(
+        self: Self,
+        embedding_models_response: dict[str, Any],
+    ):
+        """
+        Validate all returned models have 'text-embedding' in their tasks
+        """
+        models_missing_task = [
+            model["name"]
+            for model in embedding_models_response.get("items", [])
+            if "text-embedding" not in model.get("tasks", [])
+        ]
+        assert not models_missing_task, f"Models missing 'text-embedding' task: {models_missing_task}"