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]
+
+            # Default /labels excludes MCP labels — filter them out for comparison
+            model_expected_labels = [label for label in all_expected_labels if label.get("assetType") != "mcp_servers"]
+            mcp_expected_labels = [label for label in all_expected_labels if label.get("assetType") == "mcp_servers"]
+
+            # Verify model labels match default API response
+            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 MCP labels are only returned with assetType=mcp_servers
+            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

@@ -442,21 +442,25 @@ 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: str | 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"]