test: add custom properties sorting validation (#915)

fege · web-flow · commit 1901208d8e6e · 2025-12-09T08:57:28.000+01:00
Tests ensure items with custom properties appear first sorted by value,
followed by items without the property sorted by ID.
diff --git a/tests/model_registry/model_catalog/conftest.py b/tests/model_registry/model_catalog/conftest.py
@@ -203,6 +203,8 @@ def randomly_picked_model_from_catalog_api_by_source(
         headers = model_registry_rest_headers
     else:
         headers = get_rest_headers(token=user_token_for_api_calls)
+    wait_for_model_catalog_api(url=f"{model_catalog_rest_url[0]}", headers=headers)
+
     if not model_name:
         LOGGER.info(f"Picking random model from catalog: {catalog_id} with header_type: {header_type}")
         models_response = execute_get_command(
diff --git a/tests/model_registry/model_catalog/test_sorting_functionality.py b/tests/model_registry/model_catalog/test_sorting_functionality.py
@@ -1,12 +1,14 @@
 import pytest
 from typing import Self
-
+import random
+from ocp_resources.config_map import ConfigMap
 from simple_logger.logger import get_logger
 from tests.model_registry.model_catalog.utils import (
     get_models_from_catalog_api,
     get_sources_with_sorting,
     get_artifacts_with_sorting,
     validate_items_sorted_correctly,
+    verify_custom_properties_sorted,
 )
 from tests.model_registry.model_catalog.constants import VALIDATED_CATALOG_ID
 
@@ -45,6 +47,7 @@ class TestModelsSorting:
     )
     def test_models_sorting_works_correctly(
         self: Self,
+        enabled_model_catalog_config_map: ConfigMap,
         order_by: str,
         sort_order: str,
         model_catalog_rest_url: list[str],
@@ -79,6 +82,7 @@ class TestSourcesSorting:
     )
     def test_sources_sorting_works_correctly(
         self: Self,
+        enabled_model_catalog_config_map: ConfigMap,
         order_by: str,
         sort_order: str,
         model_catalog_rest_url: list[str],
@@ -101,6 +105,7 @@ def test_sources_sorting_works_correctly(
     @pytest.mark.parametrize("unsupported_field", ["CREATE_TIME", "LAST_UPDATE_TIME"])
     def test_sources_rejects_unsupported_fields(
         self: Self,
+        enabled_model_catalog_config_map: ConfigMap,
         unsupported_field: str,
         model_catalog_rest_url: list[str],
         model_registry_rest_headers: dict[str, str],
@@ -148,6 +153,7 @@ class TestArtifactsSorting:
     )
     def test_artifacts_sorting_works_correctly(
         self: Self,
+        enabled_model_catalog_config_map: ConfigMap,
         order_by: str,
         sort_order: str,
         model_catalog_rest_url: list[str],
@@ -170,3 +176,166 @@ def test_artifacts_sorting_works_correctly(
         )
 
         assert validate_items_sorted_correctly(response["items"], order_by, sort_order)
+
+
+@pytest.mark.downstream_only
+class TestCustomPropertiesSorting:
+    """Test sorting functionality for custom properties"""
+
+    MODEL_NAMEs_CUSTOM_PROPERTIES: list[str] = [
+        "RedHatAI/Llama-3.1-Nemotron-70B-Instruct-HF",
+        "RedHatAI/phi-4-quantized.w8a8",
+        "RedHatAI/Qwen2.5-7B-Instruct-quantized.w4a16",
+    ]
+
+    @pytest.mark.parametrize(
+        "order_by,sort_order,randomly_picked_model_from_catalog_api_by_source,expect_pure_fallback",
+        [
+            (
+                "e2e_p90.double_value",
+                "ASC",
+                {
+                    "catalog_id": VALIDATED_CATALOG_ID,
+                    "header_type": "registry",
+                    "model_name": random.choice(MODEL_NAMEs_CUSTOM_PROPERTIES),
+                },
+                False,
+            ),
+            (
+                "e2e_p90.double_value",
+                "DESC",
+                {
+                    "catalog_id": VALIDATED_CATALOG_ID,
+                    "header_type": "registry",
+                    "model_name": random.choice(MODEL_NAMEs_CUSTOM_PROPERTIES),
+                },
+                False,
+            ),
+            (
+                "hardware_count.int_value",
+                "ASC",
+                {
+                    "catalog_id": VALIDATED_CATALOG_ID,
+                    "header_type": "registry",
+                    "model_name": random.choice(MODEL_NAMEs_CUSTOM_PROPERTIES),
+                },
+                False,
+            ),
+            (
+                "hardware_count.int_value",
+                "DESC",
+                {
+                    "catalog_id": VALIDATED_CATALOG_ID,
+                    "header_type": "registry",
+                    "model_name": random.choice(MODEL_NAMEs_CUSTOM_PROPERTIES),
+                },
+                False,
+            ),
+            (
+                "hardware_type.string_value",
+                "ASC",
+                {
+                    "catalog_id": VALIDATED_CATALOG_ID,
+                    "header_type": "registry",
+                    "model_name": random.choice(MODEL_NAMEs_CUSTOM_PROPERTIES),
+                },
+                False,
+            ),
+            (
+                "hardware_type.string_value",
+                "DESC",
+                {
+                    "catalog_id": VALIDATED_CATALOG_ID,
+                    "header_type": "registry",
+                    "model_name": random.choice(MODEL_NAMEs_CUSTOM_PROPERTIES),
+                },
+                False,
+            ),
+            (
+                "non_existing_property.double_value",
+                "ASC",
+                {
+                    "catalog_id": VALIDATED_CATALOG_ID,
+                    "header_type": "registry",
+                    "model_name": random.choice(MODEL_NAMEs_CUSTOM_PROPERTIES),
+                },
+                True,
+            ),
+            (
+                "non_existing_property.double_value",
+                "DESC",
+                {
+                    "catalog_id": VALIDATED_CATALOG_ID,
+                    "header_type": "registry",
+                    "model_name": random.choice(MODEL_NAMEs_CUSTOM_PROPERTIES),
+                },
+                True,
+            ),
+        ],
+        indirect=["randomly_picked_model_from_catalog_api_by_source"],
+    )
+    def test_custom_properties_sorting_works_correctly(
+        self: Self,
+        enabled_model_catalog_config_map: ConfigMap,
+        order_by: str,
+        sort_order: str,
+        model_catalog_rest_url: list[str],
+        model_registry_rest_headers: dict[str, str],
+        randomly_picked_model_from_catalog_api_by_source: tuple[dict, str, str],
+        expect_pure_fallback: bool,
+    ):
+        """
+        RHOAIENG-38010: Test custom properties endpoint sorts correctly by supported fields
+
+        This test validates two scenarios:
+        1. expect_pure_fallback=False: Tests custom property sorting where at least some artifacts
+           have the property. Items with the property are sorted by the property value (ASC/DESC),
+           followed by items without the property sorted by ID ASC (fallback behavior).
+
+        2. expect_pure_fallback=True: Tests pure fallback behavior where NO artifacts have the
+           property. All items are sorted by ID ASC, regardless of the requested sortOrder.
+        """
+        _, model_name, _ = randomly_picked_model_from_catalog_api_by_source
+        LOGGER.info(
+            f"Testing custom properties sorting for {model_name}: "
+            f"orderBy={order_by}, sortOrder={sort_order}, expect_pure_fallback={expect_pure_fallback}"
+        )
+
+        response = get_artifacts_with_sorting(
+            model_catalog_rest_url=model_catalog_rest_url,
+            model_registry_rest_headers=model_registry_rest_headers,
+            source_id=VALIDATED_CATALOG_ID,
+            model_name=model_name,
+            order_by=order_by,
+            sort_order=sort_order,
+        )
+
+        # Verify how many artifacts have the custom property
+        property_name = order_by.rsplit(".", 1)[0]
+        artifacts_with_property = sum(
+            1 for item in response["items"] if property_name in item.get("customProperties", {})
+        )
+
+        if expect_pure_fallback:
+            # When property doesn't exist, sorting always falls back to ID ASC regardless of sortOrder
+            assert artifacts_with_property == 0, (
+                f"Expected no artifacts to have property {property_name} for pure fallback test, "
+                f"but found {artifacts_with_property} artifacts with it"
+            )
+            is_sorted = validate_items_sorted_correctly(items=response["items"], field="ID", order="ASC")
+            assert is_sorted, f"Pure fallback to ID ASC sorting failed for non-existing property {order_by}"
+        else:
+            # This ensures we're testing actual custom property sorting (not silent fallback)
+            assert artifacts_with_property > 0, (
+                f"Cannot test custom property sorting: no artifacts have property {property_name}. "
+                f"This would result in silent fallback to ID sorting."
+            )
+            LOGGER.info(f"{artifacts_with_property}/{len(response['items'])} artifacts have property {property_name}")
+
+            # verify_custom_properties_sorted validates:
+            # - Items WITH property come first, sorted by property value (respecting sortOrder)
+            # - Items WITHOUT property come after, sorted by ID ASC (fallback)
+            is_sorted = verify_custom_properties_sorted(
+                items=response["items"], property_field=order_by, sort_order=sort_order
+            )
+            assert is_sorted, f"Custom properties are not sorted correctly for {model_name}"
diff --git a/tests/model_registry/model_catalog/utils.py b/tests/model_registry/model_catalog/utils.py
@@ -1009,6 +1009,117 @@ def validate_items_sorted_correctly(items: list[dict], field: str, order: str) -
         raise ValueError(f"Invalid sort order: {order}")
 
 
+def verify_custom_properties_sorted(items: list[dict], property_field: str, sort_order: str) -> bool:
+    """
+    Verify if a list of items is sorted by a specific custom property value.
+
+    Expected sorting behavior:
+    1. Items WITH the custom property appear first, sorted by property value (ASC or DESC)
+    2. Items WITHOUT the custom property appear after, sorted by ID in ASC order
+
+    Args:
+        items: List of artifact items (with 'id' and 'customProperties')
+        property_field: Property field to sort by (e.g., "e2e_p90.double_value")
+        sort_order: "ASC" or "DESC" (applies only to items with the property)
+
+    Returns:
+        True if sorted correctly, False otherwise
+
+    Raises:
+        ValueError: If there are not enough items to verify sorting
+    """
+    property_name, value_type = property_field.rsplit(".", 1)
+    # Separate items into two groups
+    items_with_property, items_without_property = _split_items_by_custom_property(
+        items=items, property_name=property_name, value_type=value_type, property_field=property_field
+    )
+
+    LOGGER.info(
+        f"Total items: {len(items)}, with '{property_name}': {len(items_with_property)}, "
+        f"without: {len(items_without_property)}"
+    )
+
+    # Verify items with property come first
+    if items_with_property and items_without_property and items_with_property[-1][0] > items_without_property[0][0]:
+        LOGGER.error(
+            f"Items without property '{property_name}' appear before items with property. "
+            f"Last with property at index {items_with_property[-1][0]}, "
+            f"first without at index {items_without_property[0][0]}"
+        )
+        return False
+
+    if items_with_property and not _verify_items_with_property_sorted(
+        items=items_with_property, property_name=property_name, value_type=value_type, sort_order=sort_order
+    ):
+        return False
+
+    if items_without_property and not _verify_items_without_property_sorted(items=items_without_property):
+        return False
+
+    LOGGER.info(f"All items sorted correctly by '{property_name}' ({sort_order})")
+    return True
+
+
+def _split_items_by_custom_property(
+    items: list[dict],
+    property_name: str,
+    value_type: str,
+    property_field: str,
+) -> tuple[list[dict], list[dict]]:
+    """
+    Split items into two lists based on the presence of a custom property.
+    """
+    items_with_property = []
+    items_without_property = []
+
+    for idx, item in enumerate(items):
+        custom_props = item.get("customProperties", {})
+        if property_name in custom_props:
+            prop = custom_props[property_name]
+            value = prop.get(value_type) if isinstance(prop, dict) else None
+
+            # Only add to items_with_property if value is not None
+            if value is not None:
+                LOGGER.info(f"  [{idx}] ID: {item['id']}, {property_field}: {value}")
+                items_with_property.append((idx, item))
+            else:
+                LOGGER.info(f"  [{idx}] ID: {item['id']} (has {property_name} but value is None)")
+                items_without_property.append((idx, item))
+        else:
+            LOGGER.info(f"  [{idx}] ID: {item['id']} (no {property_name})")
+            items_without_property.append((idx, item))
+    return items_with_property, items_without_property
+
+
+def _verify_items_with_property_sorted(items: list[dict], property_name: str, value_type: str, sort_order: str) -> bool:
+    """
+    Verify if items with property are sorted correctly by property value
+    """
+    values = []
+    for _, item in items:
+        prop = item["customProperties"][property_name]
+        value = prop.get(value_type)
+        values.append(value)
+
+    expected_values = sorted(values, reverse=(sort_order == "DESC"))
+    if values != expected_values:
+        LOGGER.error(f"Items with property not sorted {sort_order}: {values}")
+        return False
+    return True
+
+
+def _verify_items_without_property_sorted(items: list[dict]) -> bool:
+    """
+    Verify if items without property are sorted correctly by ID in ASC order
+    """
+    ids = [int(item["id"]) for _, item in items]
+    expected_ids = sorted(ids)
+    if ids != expected_ids:
+        LOGGER.error(f"Items without property not sorted by ID ASC: {ids}")
+        return False
+    return True
+
+
 def _validate_single_criterion(
     artifact_name: str, custom_properties: dict[str, Any], validation: dict[str, Any]
 ) -> tuple[bool, str]:
