Add tests for labels endpoint

tests/model_registry/model_catalog/conftest.py

@@ -18,7 +18,7 @@
     REDHAT_AI_CATALOG_ID,
 )
 from tests.model_registry.model_catalog.utils import get_models_from_catalog_api
-from tests.model_registry.constants import CUSTOM_CATALOG_ID1
+from tests.model_registry.constants import CUSTOM_CATALOG_ID1, DEFAULT_CUSTOM_MODEL_CATALOG
 from tests.model_registry.utils import (
     get_rest_headers,
     is_model_catalog_ready,
@@ -225,3 +225,27 @@ def models_from_filter_query(
     LOGGER.info(f"Filter query '{filter_query}' returned {len(model_names)} models: {', '.join(model_names)}")
 
     return model_names
+
+
+@pytest.fixture()
+def labels_configmap_patch(admin_client: DynamicClient, model_registry_namespace: str) -> dict[str, Any]:
+    # Get the editable ConfigMap
+    sources_cm = ConfigMap(name=DEFAULT_CUSTOM_MODEL_CATALOG, client=admin_client, namespace=model_registry_namespace)
+
+    # 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",
+    }
+
+    if "labels" not in current_data:
+        current_data["labels"] = []
+    current_data["labels"].append(new_label)
+
+    patches = {"data": {"sources.yaml": yaml.dump(current_data, default_flow_style=False)}}
+
+    with ResourceEditor(patches={sources_cm: patches}):
+        yield patches

tests/model_registry/model_catalog/test_labels_endpoint.py

@@ -0,0 +1,71 @@
+import pytest
+from typing import Any
+from kubernetes.dynamic import DynamicClient
+from simple_logger.logger import get_logger
+
+
+from utilities.infra import get_openshift_token
+from timeout_sampler import TimeoutSampler
+
+from tests.model_registry.model_catalog.utils import (
+    get_labels_from_configmaps,
+    get_labels_from_api,
+    verify_labels_match,
+)
+
+LOGGER = get_logger(name=__name__)
+
+
+class TestLabelsEndpoint:
+    """Test class for the model catalog labels endpoint."""
+
+    @pytest.mark.smoke
+    def test_labels_endpoint_default_data(
+        self,
+        admin_client: DynamicClient,
+        model_registry_namespace: str,
+        model_catalog_rest_url: list[str],
+    ):
+        """
+        Smoke test: Validate default labels from ConfigMaps are returned by the endpoint.
+        """
+        LOGGER.info("Testing labels endpoint with default data")
+
+        # Get 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()
+        )
+
+        # Verify they match
+        verify_labels_match(expected_labels=expected_labels, api_labels=api_labels)
+
+    @pytest.mark.sanity
+    def test_labels_endpoint_configmap_updates(
+        self,
+        admin_client: DynamicClient,
+        model_registry_namespace: str,
+        model_catalog_rest_url: list[str],
+        labels_configmap_patch: dict[str, Any],
+    ):
+        """
+        Sanity test: Edit the editable ConfigMap and verify changes are reflected in API.
+        """
+        _ = labels_configmap_patch
+
+        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()
+            )
+
+            # Verify they match (including the new label)
+            verify_labels_match(expected_labels=expected_labels, api_labels=api_labels)
+
+        sampler = TimeoutSampler(wait_timeout=60, sleep=5, func=_check_updated_labels)
+        next(iter(sampler))

tests/model_registry/model_catalog/utils.py

@@ -1,4 +1,4 @@
-from typing import Any, Tuple, List
+from typing import Any, Tuple, List, Dict
 import yaml
 
 from kubernetes.dynamic import DynamicClient
@@ -22,7 +22,9 @@
     CATALOG_CONTAINER,
     PERFORMANCE_DATA_DIR,
 )
+from tests.model_registry.constants import DEFAULT_CUSTOM_MODEL_CATALOG, DEFAULT_MODEL_CATALOG_CM
 from tests.model_registry.utils import execute_get_command
+from tests.model_registry.utils import get_rest_headers
 
 LOGGER = get_logger(name=__name__)
 
@@ -1121,3 +1123,75 @@ def validate_model_artifacts_match_criteria_or(
 
     LOGGER.error(f"Model {model_name} failed all OR validations")
     return False
+
+
+def get_labels_from_configmaps(admin_client: DynamicClient, namespace: str) -> List[Dict[str, Any]]:
+    """
+    Get all labels from both model catalog ConfigMaps.
+
+    Args:
+        admin_client: Kubernetes client
+        namespace: Namespace containing the ConfigMaps
+
+    Returns:
+        List of all label dictionaries from both ConfigMaps
+    """
+    labels = []
+
+    # Get labels from default ConfigMap
+    default_cm = ConfigMap(name=DEFAULT_MODEL_CATALOG_CM, client=admin_client, namespace=namespace)
+    default_data = yaml.safe_load(default_cm.instance.data["sources.yaml"])
+    if "labels" in default_data:
+        labels.extend(default_data["labels"])
+
+    # Get labels from sources ConfigMap
+    sources_cm = ConfigMap(name=DEFAULT_CUSTOM_MODEL_CATALOG, client=admin_client, namespace=namespace)
+    sources_data = yaml.safe_load(sources_cm.instance.data["sources.yaml"])
+    if "labels" in sources_data:
+        labels.extend(sources_data["labels"])
+
+    return labels
+
+
+def get_labels_from_api(model_catalog_rest_url: str, user_token: str) -> 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
+
+    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)
+    return response["items"]
+
+
+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.
+
+    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
+    """
+    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}"