test: label endpoint with asset type

tests/model_registry/model_catalog/conftest.py

@@ -369,15 +369,21 @@ def labels_configmap_patch(
     # Parse current data and add test label
     current_data = yaml.safe_load(sources_cm.instance.data["sources.yaml"])
 
-    new_label = {
-        "name": "test-dynamic",
-        "displayName": "Dynamic Test Label",
-        "description": "A label added during test execution",
-    }
+    new_labels = [
+        {
+            "name": "test-dynamic",
+            "displayName": "Dynamic Test Label",
+            "description": "A label added during test execution",
+        },
+        {
+            "name": "mcp-test-label",
+            "assetType": "mcp_servers",
+        },
+    ]
 
     if "labels" not in current_data:
         current_data["labels"] = []
-    current_data["labels"].append(new_label)
+    current_data["labels"].extend(new_labels)
 
     patches = {"data": {"sources.yaml": yaml.dump(current_data, default_flow_style=False)}}
 

tests/model_registry/model_catalog/metadata/test_labels_endpoint.py

@@ -56,15 +56,24 @@ def test_labels_endpoint_configmap_updates(
 
         def _check_updated_labels():
             # Get updated expected labels from ConfigMaps
-            expected_labels = get_labels_from_configmaps(admin_client=admin_client, namespace=model_registry_namespace)
-
-            # Get labels from API
-            api_labels = get_labels_from_api(
-                model_catalog_rest_url=model_catalog_rest_url[0], user_token=get_openshift_token()
+            all_expected_labels = get_labels_from_configmaps(
+                admin_client=admin_client, namespace=model_registry_namespace
             )
 
-            # Verify they match (including the new label)
-            verify_labels_match(expected_labels=expected_labels, api_labels=api_labels)
+            token = get_openshift_token()
+            url = model_catalog_rest_url[0]
+
+            # Split expected labels by asset type
+            mcp_expected_labels = [label for label in all_expected_labels if label.get("assetType") == "mcp_servers"]
+            model_expected_labels = [label for label in all_expected_labels if label not in mcp_expected_labels]
+
+            # Verify default /labels returns only model labels (no MCP cross-contamination)
+            api_labels = get_labels_from_api(model_catalog_rest_url=url, user_token=token)
+            verify_labels_match(expected_labels=model_expected_labels, api_labels=api_labels)
+
+            # Verify assetType=mcp_servers returns only MCP labels (no model cross-contamination)
+            mcp_api_labels = get_labels_from_api(model_catalog_rest_url=url, user_token=token, asset_type="mcp_servers")
+            verify_labels_match(expected_labels=mcp_expected_labels, api_labels=mcp_api_labels)
 
         sampler = TimeoutSampler(wait_timeout=60, sleep=5, func=_check_updated_labels)
         for _ in sampler:

tests/model_registry/model_catalog/metadata/utils.py

@@ -1,6 +1,6 @@
 import json
 from fnmatch import fnmatch
-from typing import Any
+from typing import Any, Literal
 
 import requests
 import yaml
@@ -442,46 +442,56 @@ def get_labels_from_configmaps(admin_client: DynamicClient, namespace: str) -> l
     return labels
 
 
-def get_labels_from_api(model_catalog_rest_url: str, user_token: str) -> list[dict[str, Any]]:
+def get_labels_from_api(
+    model_catalog_rest_url: str, user_token: str, asset_type: Literal["models", "mcp_servers"] | None = None
+) -> list[dict[str, Any]]:
     """
     Get labels from the API endpoint.
 
     Args:
         model_catalog_rest_url: Base URL for model catalog API
         user_token: Authentication token
+        asset_type: Filter by asset type ('models' or 'mcp_servers')
 
     Returns:
         List of label dictionaries from API response
     """
 
     url = f"{model_catalog_rest_url}labels"
     headers = get_rest_headers(token=user_token)
-    response = execute_get_command(url=url, headers=headers)
+    params: dict[str, str] | None = {"assetType": asset_type} if asset_type is not None else None
+    response = execute_get_command(url=url, headers=headers, params=params)
     return response["items"]
 
 
+def _label_key(label: dict[str, Any]) -> tuple[str | None, str | None, str | None]:
+    """Extract comparable key from a label dict."""
+    return (label.get("name"), label.get("displayName"), label.get("description"))
+
+
 def verify_labels_match(expected_labels: list[dict[str, Any]], api_labels: list[dict[str, Any]]) -> None:
     """
-    Verify that all expected labels are present in the API response.
+    Verify that expected labels and API labels match exactly (bidirectional).
 
     Args:
         expected_labels: Labels expected from ConfigMaps
         api_labels: Labels returned by API
 
     Raises:
-        AssertionError: If any expected label is not found in API response
+        AssertionError: If there are missing or unexpected labels
     """
     LOGGER.info(f"Verifying {len(expected_labels)} expected labels against {len(api_labels)} API labels")
 
-    for expected_label in expected_labels:
-        found = False
-        for api_label in api_labels:
-            if (
-                expected_label.get("name") == api_label.get("name")
-                and expected_label.get("displayName") == api_label.get("displayName")
-                and expected_label.get("description") == api_label.get("description")
-            ):
-                found = True
-                break
-
-        assert found, f"Expected label not found in API response: {expected_label}"
+    expected_keys = {_label_key(label) for label in expected_labels}
+    api_keys = {_label_key(label) for label in api_labels}
+
+    missing = expected_keys - api_keys
+    unexpected = api_keys - expected_keys
+
+    errors = []
+    if missing:
+        errors.append(f"Missing labels not found in API response: {missing}")
+    if unexpected:
+        errors.append(f"Unexpected labels in API response: {unexpected}")
+
+    assert not errors, "\n".join(errors)