chore: ensure tests can run against live cluster

AI-assisted: Claude Code (Opus 4.6)
Signed-off-by: Debarati Basu-Nag <dbasunag@redhat.com>

Author: dbasunag

catalog/clients/python/tests/mcp_servers/test_filtering.py

@@ -8,10 +8,14 @@
 3. Run: pytest --e2e tests/mcp_servers/test_filtering.py
 """
 
+import os
+
 import pytest
 
 from model_catalog import CatalogAPIClient, CatalogAPIError, CatalogNotFoundError
 
+_KIND_CLUSTER = os.getenv("KIND_CLUSTER", "True").lower() in ("true", "1", "yes")
+
 
 class TestMCPServerFiltering:
     """Test suite for MCP server filterQuery functionality."""
@@ -36,6 +40,7 @@ class TestMCPServerFiltering:
                 1,
                 {"calculator"},
                 id="boolean_flags",
+                marks=pytest.mark.skipif(not _KIND_CLUSTER, reason="Non-KinD environments have additional MCP sources"),
             ),
             pytest.param(
                 "license='MIT' AND (provider='Math Community' OR provider='Weather Community')",

catalog/clients/python/tests/mcp_servers/test_named_queries.py

@@ -35,14 +35,16 @@ def test_named_query_execution(
         self,
         api_client: CatalogAPIClient,
         suppress_ssl_warnings: None,
+        kind_cluster: bool,
         named_query: str,
         expected_security_indicators: dict[str, bool],
     ):
         """TC-API-011: Test executing a named query filters servers by security indicators."""
         response = api_client.get_mcp_servers(named_query=named_query)
         items = response.get("items", [])
-        assert len(items) == 1, f"Expected 1 server matching '{named_query}', got {len(items)}"
-        assert items[0]["name"] == "calculator"
+        if kind_cluster:
+            assert len(items) == 1, f"Expected 1 server matching '{named_query}', got {len(items)}"
+            assert items[0]["name"] == "calculator"
 
         security_indicators = items[0].get("securityIndicators", {})
         for prop_name, expected_value in expected_security_indicators.items():

catalog/clients/python/tests/mcp_servers/test_servers.py

@@ -23,13 +23,15 @@ def test_mcp_servers_loaded(
         api_client: CatalogAPIClient,
         suppress_ssl_warnings: None,
         test_mcp_catalog_data: dict,
+            kind_cluster: bool,
     ):
         """Test that all expected MCP servers are loaded with required fields."""
         response = api_client.get_mcp_servers()
         items = response.get("items", [])
         actual_names = {server["name"] for server in items}
-        expected_names = {server["name"] for server in test_mcp_catalog_data["mcp_servers"]}
-        assert actual_names == expected_names
+        if kind_cluster:
+            expected_names = {server["name"] for server in test_mcp_catalog_data["mcp_servers"]}
+            assert actual_names == expected_names
 
         for server in items:
             missing = MCP_SERVER_REQUIRED_FIELDS - server.keys()
@@ -40,12 +42,16 @@ def test_mcp_server_providers(
         api_client: CatalogAPIClient,
         suppress_ssl_warnings: None,
         test_mcp_catalog_data: dict,
+        kind_cluster: bool,
     ):
         """Test that MCP server providers match expected values."""
         response = api_client.get_mcp_servers()
-        actual_providers = {s["name"]: s["provider"] for s in response["items"]}
-        expected_providers = {s["name"]: s["provider"] for s in test_mcp_catalog_data["mcp_servers"]}
-        assert actual_providers == expected_providers
+        actual_providers = {server["name"]: server["provider"] for server in response["items"]}
+        expected_providers = {server["name"]: server["provider"] for server in test_mcp_catalog_data["mcp_servers"]}
+        if kind_cluster:
+            assert actual_providers == expected_providers
+        else:
+            assert set(expected_providers.items()).issubset(set(actual_providers.items()))
 
     def test_mcp_server_get_by_id(
         self,
@@ -67,13 +73,16 @@ def test_mcp_server_pagination(
         api_client: CatalogAPIClient,
         suppress_ssl_warnings: None,
         test_mcp_catalog_data: dict,
+        kind_cluster: bool,
     ):
         """Test pagination by iterating all servers with pageSize=1."""
         expected_names = {server["name"] for server in test_mcp_catalog_data["mcp_servers"]}
         collected_names: set[str] = set()
         next_token = None
 
-        for _ in range(len(expected_names) + 1):
+        # Use a safe upper bound: on non-KinD environments there may be more servers
+        max_pages = len(expected_names) + 1 if kind_cluster else 100
+        for _ in range(max_pages):
             response = api_client.get_mcp_servers(page_size=1, next_page_token=next_token)
             items = response.get("items", [])
             assert len(items) == 1, f"Expected exactly 1 item per page, got {len(items)}"
@@ -86,7 +95,10 @@ def test_mcp_server_pagination(
             if not next_token:
                 break
 
-        assert collected_names == expected_names
+        if kind_cluster:
+            assert collected_names == expected_names
+        else:
+            assert expected_names.issubset(collected_names)
 
     def test_filter_by_name(
         self,

catalog/clients/python/tests/models/test_artifacts.py

@@ -447,6 +447,7 @@ def test_artifacts_sorting_by_id(
         self,
         api_client: CatalogAPIClient,
         model_with_artifacts: tuple[str, str],
+            suppress_ssl_warnings: None,
         order_by: str,
         sort_order: str,
     ) -> None:
@@ -475,6 +476,7 @@ def test_artifacts_sorting_by_custom_property(
         self,
         api_client: CatalogAPIClient,
         model_with_artifacts: tuple[str, str],
+            suppress_ssl_warnings: None,
         order_by: str,
         sort_order: str,
     ) -> None:
@@ -496,6 +498,7 @@ def test_sorting_by_non_existing_property_falls_back_to_id(
         self,
         api_client: CatalogAPIClient,
         model_with_artifacts: tuple[str, str],
+            suppress_ssl_warnings: None
     ) -> None:
         """Test that sorting by a non-existing property falls back to ID ASC."""
         source_id, model_name = model_with_artifacts

catalog/clients/python/tests/models/test_filter_query.py

@@ -31,6 +31,8 @@ def test_search_models_by_filter_query(
         pattern matching, equality on JSON array property (tasks),
         and AND logic combining all conditions.
         """
+        if not kind_cluster:
+            pytest.skip(reason="Specific configuration is not applied on a live cluster")
         sizes = ("7B", "13B")
         filter_query = "size IN ('7B','13B') AND name ILIKE '%test%' AND tasks = 'text-generation'"