test: include and exclude filter for mcp servers (#1246)

Signed-off-by: fege <fmosca@redhat.com>
Co-authored-by: Debarati Basu-Nag <dbasunag@redhat.com>

Author: fege

tests/model_registry/mcp_servers/config/conftest.py

@@ -12,8 +12,11 @@
     MCP_CATALOG_INVALID_SOURCE,
     MCP_CATALOG_SOURCE,
     MCP_CATALOG_SOURCE2,
+    MCP_CATALOG_SOURCE_ID,
+    MCP_CATALOG_SOURCE_NAME,
     MCP_SERVERS_YAML,
     MCP_SERVERS_YAML2,
+    MCP_SERVERS_YAML_CATALOG_PATH,
 )
 from tests.model_registry.utils import (
     wait_for_mcp_catalog_api,
@@ -106,3 +109,64 @@ def mcp_invalid_yaml_configmap_patch(
     wait_for_model_catalog_pod_ready_after_deletion(
         client=admin_client, model_registry_namespace=model_registry_namespace
     )
+
+
+@pytest.fixture(scope="class")
+def mcp_included_excluded_configmap_patch(
+    request: pytest.FixtureRequest,
+    admin_client: DynamicClient,
+    model_registry_namespace: str,
+    mcp_catalog_rest_urls: list[str],
+    model_registry_rest_headers: dict[str, str],
+) -> Generator[None]:
+    """
+    Class-scoped fixture that patches the ConfigMap with an MCP source
+    including includedServers/excludedServers filters.
+
+    Parametrized via request.param with a dict containing:
+      - "includedServers": list[str] (optional)
+      - "excludedServers": list[str] (optional)
+    """
+    filter_params = request.param
+
+    source_config: dict = {
+        "name": MCP_CATALOG_SOURCE_NAME,
+        "id": MCP_CATALOG_SOURCE_ID,
+        "type": "yaml",
+        "enabled": True,
+        "properties": {"yamlCatalogPath": MCP_SERVERS_YAML_CATALOG_PATH},
+        "labels": [MCP_CATALOG_SOURCE_NAME],
+    }
+    if "includedServers" in filter_params:
+        source_config["includedServers"] = filter_params["includedServers"]
+    if "excludedServers" in filter_params:
+        source_config["excludedServers"] = filter_params["excludedServers"]
+
+    catalog_config_map = ConfigMap(
+        name=DEFAULT_CUSTOM_MODEL_CATALOG,
+        client=admin_client,
+        namespace=model_registry_namespace,
+    )
+
+    current_data = yaml.safe_load(catalog_config_map.instance.data.get("sources.yaml", "{}") or "{}")
+    if "mcp_catalogs" not in current_data:
+        current_data["mcp_catalogs"] = []
+    current_data["mcp_catalogs"].append(source_config)
+
+    patches = {
+        "data": {
+            "sources.yaml": yaml.dump(current_data, default_flow_style=False),
+            "mcp-servers.yaml": MCP_SERVERS_YAML,
+        }
+    }
+
+    with ResourceEditor(patches={catalog_config_map: patches}):
+        wait_for_model_catalog_pod_ready_after_deletion(
+            client=admin_client, model_registry_namespace=model_registry_namespace
+        )
+        wait_for_mcp_catalog_api(url=mcp_catalog_rest_urls[0], headers=model_registry_rest_headers)
+        yield
+
+    wait_for_model_catalog_pod_ready_after_deletion(
+        client=admin_client, model_registry_namespace=model_registry_namespace
+    )

tests/model_registry/mcp_servers/config/test_included_excluded_servers.py

@@ -0,0 +1,38 @@
+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.tier2
+@pytest.mark.parametrize(
+    "mcp_included_excluded_configmap_patch",
+    [
+        pytest.param(
+            {"includedServers": ["weather-*", "file-*"], "excludedServers": ["file-*"]},
+            id="combined_include_exclude",
+        ),
+    ],
+    indirect=True,
+)
+@pytest.mark.usefixtures("mcp_included_excluded_configmap_patch")
+class TestMCPServerIncludedExcludedFiltering:
+    """Tests for includedServers/excludedServers glob pattern filtering (TC-LOAD-003, TC-LOAD-004, TC-LOAD-005)."""
+
+    def test_included_and_excluded_servers(
+        self: Self,
+        mcp_catalog_rest_urls: list[str],
+        model_registry_rest_headers: dict[str, str],
+    ):
+        """Verify includedServers are loaded and excludedServers are not (TC-LOAD-003, TC-LOAD-004, TC-LOAD-005)."""
+        response = execute_get_command(
+            url=f"{mcp_catalog_rest_urls[0]}mcp_servers",
+            headers=model_registry_rest_headers,
+        )
+        server_names = {server["name"] for server in response.get("items", [])}
+        assert "weather-api" in server_names, f"Expected 'weather-api' in {server_names}"
+        assert "file-manager" not in server_names, f"Expected 'file-manager' not in {server_names}"