Merge branch 'main' into QE_refactoring-tags

Author: dbasunag

tests/model_registry/mcp_servers/constants.py

@@ -1,3 +1,5 @@
+CALCULATOR_SERVER_NAME: str = "calculator"
+CALCULATOR_PROVIDER: str = "Math Community"
 MCP_CATALOG_SOURCE_ID: str = "test_mcp_servers"
 MCP_CATALOG_SOURCE_NAME: str = "Test MCP Servers"
 MCP_CATALOG_API_PATH: str = "/api/mcp_catalog/v1alpha1/"

tests/model_registry/mcp_servers/search/__init__.py

@@ -0,0 +1,103 @@
+from typing import Self
+
+import pytest
+from simple_logger.logger import get_logger
+
+from tests.model_registry.mcp_servers.constants import CALCULATOR_PROVIDER, CALCULATOR_SERVER_NAME
+from tests.model_registry.utils import execute_get_command
+
+LOGGER = get_logger(name=__name__)
+
+
+@pytest.mark.usefixtures("mcp_servers_configmap_patch")
+class TestMCPServerFiltering:
+    """RHOAIENG-51584: Tests for MCP server filterQuery functionality."""
+
+    @pytest.mark.parametrize(
+        "filter_query, expected_count, expected_name, field_check",
+        [
+            pytest.param(
+                f"provider='{CALCULATOR_PROVIDER}'",
+                1,
+                CALCULATOR_SERVER_NAME,
+                ("provider", CALCULATOR_PROVIDER),
+                id="by_provider",
+            ),
+            pytest.param("tags='math'", 1, CALCULATOR_SERVER_NAME, None, id="by_tags"),
+        ],
+    )
+    def test_filter_by_field(
+        self: Self,
+        mcp_catalog_rest_urls: list[str],
+        model_registry_rest_headers: dict[str, str],
+        filter_query: str,
+        expected_count: int,
+        expected_name: str,
+        field_check: tuple[str, str] | None,
+    ):
+        """TC-API-003, TC-API-005: Test filtering MCP servers by provider and tags."""
+        response = execute_get_command(
+            url=f"{mcp_catalog_rest_urls[0]}mcp_servers",
+            headers=model_registry_rest_headers,
+            params={"filterQuery": filter_query},
+        )
+        items = response.get("items", [])
+        assert len(items) == expected_count, (
+            f"Expected {expected_count} server(s) for '{filter_query}', got {len(items)}"
+        )
+        assert items[0]["name"] == expected_name
+        if field_check:
+            assert items[0][field_check[0]] == field_check[1]
+
+    def test_filter_options(
+        self: Self,
+        mcp_catalog_rest_urls: list[str],
+        model_registry_rest_headers: dict[str, str],
+    ):
+        """TC-API-026: Test that filter_options endpoint returns available filter fields."""
+        url = f"{mcp_catalog_rest_urls[0]}mcp_servers/filter_options"
+        LOGGER.info(f"Requesting filter_options from: {url}")
+
+        response = execute_get_command(
+            url=url,
+            headers=model_registry_rest_headers,
+        )
+        LOGGER.info(f"filter_options full response: {response}")
+
+        filters = response["filters"]
+
+        expected_filter_fields = {"description", "provider", "license", "tags"}
+        actual_filter_fields = set(filters.keys())
+        assert expected_filter_fields == actual_filter_fields
+
+    def test_pagination_with_filters(
+        self: Self,
+        mcp_catalog_rest_urls: list[str],
+        model_registry_rest_headers: dict[str, str],
+    ):
+        """TC-API-032: Test that pagination works correctly with filterQuery."""
+        base_url = f"{mcp_catalog_rest_urls[0]}mcp_servers"
+        filter_query = "license='MIT'"
+
+        # First page
+        response = execute_get_command(
+            url=base_url,
+            headers=model_registry_rest_headers,
+            params={"filterQuery": filter_query, "pageSize": "1"},
+        )
+        first_page_items = response.get("items", [])
+        assert len(first_page_items) == 1, f"Expected 1 item on first page, got {len(first_page_items)}"
+        next_page_token = response.get("nextPageToken")
+        assert next_page_token, "Expected nextPageToken for second page"
+
+        # Second page
+        response = execute_get_command(
+            url=base_url,
+            headers=model_registry_rest_headers,
+            params={"filterQuery": filter_query, "pageSize": "1", "nextPageToken": next_page_token},
+        )
+        second_page_items = response.get("items", [])
+        assert len(second_page_items) == 1, f"Expected 1 item on second page, got {len(second_page_items)}"
+
+        collected_names = {first_page_items[0]["name"], second_page_items[0]["name"]}
+        assert collected_names == {"weather-api", CALCULATOR_SERVER_NAME}

tests/model_registry/mcp_servers/search/test_filtering.py

@@ -0,0 +1,37 @@
+from typing import Self
+
+import pytest
+from simple_logger.logger import get_logger
+
+from tests.model_registry.utils import execute_get_command
+
+LOGGER = get_logger(name=__name__)
+
+
+@pytest.mark.usefixtures("mcp_servers_configmap_patch")
+class TestMCPServerOrdering:
+    """RHOAIENG-51584: Tests for MCP server ordering functionality."""
+
+    @pytest.mark.xfail(reason="RHOAIENG-52448: sortOrder/orderBy not working, fix in PR #2367")
+    @pytest.mark.parametrize(
+        "sort_order",
+        [
+            pytest.param("ASC", id="ascending"),
+            pytest.param("DESC", id="descending"),
+        ],
+    )
+    def test_ordering_by_name(
+        self: Self,
+        mcp_catalog_rest_urls: list[str],
+        model_registry_rest_headers: dict[str, str],
+        sort_order: str,
+    ):
+        """TC-API-014: Test ordering MCP servers by name ASC/DESC."""
+        response = execute_get_command(
+            url=f"{mcp_catalog_rest_urls[0]}mcp_servers",
+            headers=model_registry_rest_headers,
+            params={"orderBy": "name", "sortOrder": sort_order},
+        )
+        actual_names = [s["name"] for s in response.get("items", [])]
+        expected_names = sorted(actual_names, reverse=(sort_order == "DESC"))
+        assert actual_names == expected_names