test: remove schema validation tests moved to upstream (#1070)

Removes schema structure validation tests that are moved by
kubeflow/model-registry#2149.
Retains functional tests that verify data correctness against database state and metadata files.

Author: fege

tests/model_registry/model_catalog/metadata/test_custom_properties.py

@@ -7,7 +7,6 @@
 from tests.model_registry.model_catalog.constants import VALIDATED_CATALOG_ID, REDHAT_AI_CATALOG_ID
 from tests.model_registry.model_catalog.metadata.utils import (
     extract_custom_property_values,
-    validate_custom_properties_structure,
     validate_custom_properties_match_metadata,
     get_metadata_from_catalog_pod,
 )
@@ -27,21 +26,6 @@
 class TestCustomProperties:
     """Test suite for validating custom properties in model catalog API"""
 
-    @pytest.mark.parametrize(
-        "randomly_picked_model_from_catalog_api_by_source", [{"source": VALIDATED_CATALOG_ID}], indirect=True
-    )
-    def test_custom_properties_structure_is_valid(
-        self,
-        randomly_picked_model_from_catalog_api_by_source: tuple[dict[Any, Any], str, str],
-    ):
-        """Test that custom properties follow the expected MetadataStringValue structure."""
-        model_data, model_name, catalog_id = randomly_picked_model_from_catalog_api_by_source
-
-        LOGGER.info(f"Testing custom properties structure for model '{model_name}' from catalog '{catalog_id}'")
-
-        custom_props = model_data.get("customProperties", {})
-        assert validate_custom_properties_structure(custom_props)
-
     @pytest.mark.parametrize(
         "randomly_picked_model_from_catalog_api_by_source", [{"source": VALIDATED_CATALOG_ID}], indirect=True
     )

tests/model_registry/model_catalog/metadata/test_filter_options_endpoint.py

@@ -2,7 +2,6 @@
 from typing import Self
 from simple_logger.logger import get_logger
 from tests.model_registry.model_catalog.metadata.utils import (
-    validate_filter_options_structure,
     compare_filter_options_with_database,
 )
 from tests.model_registry.model_catalog.utils import (
@@ -30,62 +29,6 @@ class TestFilterOptionsEndpoint:
     RHOAIENG-36696
     """
 
-    @pytest.mark.parametrize(
-        "user_token_for_api_calls,",
-        [
-            pytest.param(
-                {},
-                id="test_filter_options_admin_user",
-            ),
-            pytest.param(
-                {"user_type": "test"},
-                id="test_filter_options_non_admin_user",
-            ),
-            pytest.param(
-                {"user_type": "sa_user"},
-                id="test_filter_options_service_account",
-            ),
-        ],
-        indirect=["user_token_for_api_calls"],
-    )
-    @pytest.mark.sanity
-    def test_filter_options_endpoint_validation(
-        self: Self,
-        model_catalog_rest_url: list[str],
-        user_token_for_api_calls: str,
-        test_idp_user: UserTestSession,
-    ):
-        """
-        Comprehensive test for filter_options endpoint.
-        Validates all acceptance criteria:
-        - A GET request returns a 200 OK response
-        - Response includes filter options for string-based properties with values array containing distinct values
-        - Response includes filter options for numeric properties with range object containing min/max values
-        - Core properties are present (license, provider, tasks, validated_on)
-        """
-        url = f"{model_catalog_rest_url[0]}models/filter_options"
-        LOGGER.info(f"Testing filter_options endpoint: {url}")
-
-        # This will raise an exception if the status code is not 200/201 (validates acceptance criteria #1)
-        response = execute_get_command(
-            url=url,
-            headers=get_rest_headers(token=user_token_for_api_calls),
-        )
-
-        assert response is not None, "Filter options response should not be None"
-        LOGGER.info("Filter options endpoint successfully returned 200 OK")
-
-        # Expected core properties based on current API response
-        expected_properties = {"license", "provider", "tasks", "validated_on.array_value"}
-
-        # Comprehensive validation using single function (validates acceptance criteria #2, #3, #4)
-        is_valid, errors = validate_filter_options_structure(response=response, expected_properties=expected_properties)
-        assert is_valid, f"Filter options validation failed: {'; '.join(errors)}"
-
-        filters = response["filters"]
-        LOGGER.info(f"Found {len(filters)} filter properties: {list(filters.keys())}")
-        LOGGER.info("All filter options validation passed successfully")
-
     # Cannot use non-admin user for this test as it cannot list the pods in the namespace
     @pytest.mark.parametrize(
         "user_token_for_api_calls,",

tests/model_registry/model_catalog/metadata/test_sources_endpoint.py

@@ -4,7 +4,6 @@
 
 from tests.model_registry.model_catalog.constants import REDHAT_AI_CATALOG_ID
 from tests.model_registry.utils import execute_get_command
-from tests.model_registry.model_catalog.metadata.utils import validate_source_status
 
 pytestmark = [pytest.mark.usefixtures("updated_dsc_component_state_scope_session", "model_registry_namespace")]
 
@@ -14,75 +13,12 @@
 class TestSourcesEndpoint:
     """Test class for the model catalog sources endpoint."""
 
-    @pytest.mark.smoke
-    def test_available_source_status(
-        self,
-        model_catalog_rest_url: list[str],
-        model_registry_rest_headers: dict[str, str],
-    ):
-        """
-        RHOAIENG-41849: Test that the sources endpoint returns no error for available sources.
-        """
-        response = execute_get_command(url=f"{model_catalog_rest_url[0]}sources", headers=model_registry_rest_headers)
-        items = response.get("items", [])
-        assert items, "Sources not found"
-        for item in items:
-            validate_source_status(catalog=item, expected_status="available")
-            error_value = item["error"]
-            assert error_value is None or error_value == "", (
-                f"Source '{item.get('id')}' should not have error, got: {error_value}"
-            )
-
-            LOGGER.info(
-                f"Available catalog verified - ID: {item.get('id')}, Status: {item.get('status')}, Error: {error_value}"
-            )
-
-    @pytest.mark.parametrize(
-        "sparse_override_catalog_source",
-        [{"id": REDHAT_AI_CATALOG_ID, "field_name": "enabled", "field_value": False}],
-        indirect=True,
-    )
-    def test_disabled_source_status(
-        self,
-        sparse_override_catalog_source: dict,
-        model_catalog_rest_url: list[str],
-        model_registry_rest_headers: dict[str, str],
-    ):
-        """
-        RHOAIENG-41849:
-        This test disables an existing catalog and verifies:
-        - status field is "disabled"
-        - error field is null or empty
-        """
-        catalog_id = sparse_override_catalog_source["catalog_id"]
-
-        response = execute_get_command(url=f"{model_catalog_rest_url[0]}sources", headers=model_registry_rest_headers)
-        items = response.get("items", [])
-
-        # Find the disabled catalog
-        disabled_catalog = next((item for item in items if item.get("id") == catalog_id), None)
-        assert disabled_catalog is not None, f"Disabled catalog '{catalog_id}' not found in sources"
-
-        # Validate status and error fields
-        validate_source_status(catalog=disabled_catalog, expected_status="disabled")
-        error_value = disabled_catalog["error"]
-        assert error_value is None or error_value == "", (
-            f"Source '{disabled_catalog.get('id')}' should not have error, got: {error_value}"
-        )
-
-        LOGGER.info(
-            "Disabled catalog verified - "
-            f"ID: {disabled_catalog.get('id')}, "
-            f"Status: {disabled_catalog.get('status')}, "
-            f"Error: {error_value}"
-        )
-
     @pytest.mark.parametrize(
         "sparse_override_catalog_source",
         [{"id": REDHAT_AI_CATALOG_ID, "field_name": "enabled", "field_value": False}],
         indirect=True,
     )
-    @pytest.mark.sanity
+    @pytest.mark.smoke
     def test_sources_endpoint_returns_all_sources_regardless_of_enabled_field(
         self,
         sparse_override_catalog_source: dict,

tests/model_registry/model_catalog/metadata/utils.py

@@ -244,52 +244,6 @@ def extract_custom_property_values(custom_properties: dict[str, Any]) -> dict[st
     return extracted
 
 
-def validate_custom_properties_structure(custom_properties: dict[str, Any]) -> bool:
-    """
-    Validate that custom properties follow the expected MetadataStringValue structure.
-
-    Args:
-        custom_properties: Dictionary of custom properties from API response
-
-    Returns:
-        True if all custom properties have valid structure, False otherwise
-    """
-    if not custom_properties:
-        LOGGER.info("No custom properties found - structure validation skipped")
-        return True
-
-    expected_keys = ["size", "tensor_type", "variant_group_id"]
-
-    for key in expected_keys:
-        if key in custom_properties:
-            prop_data = custom_properties[key]
-
-            if not isinstance(prop_data, dict):
-                LOGGER.error(f"Custom property '{key}' is not a dictionary: {prop_data}")
-                return False
-
-            if "metadataType" not in prop_data:
-                LOGGER.error(f"Custom property '{key}' missing 'metadataType' field")
-                return False
-
-            if prop_data.get("metadataType") != "MetadataStringValue":
-                LOGGER.error(f"Custom property '{key}' has unexpected metadataType: {prop_data.get('metadataType')}")
-                return False
-
-            if "string_value" not in prop_data:
-                LOGGER.error(f"Custom property '{key}' missing 'string_value' field")
-                return False
-
-            if not isinstance(prop_data.get("string_value"), str):
-                LOGGER.error(f"Custom property '{key}' string_value is not a string: {prop_data.get('string_value')}")
-                return False
-
-            LOGGER.info(f"Custom property '{key}' has valid structure: '{prop_data.get('string_value')}'")
-
-    LOGGER.info("All custom properties have valid structure")
-    return True
-
-
 def validate_custom_properties_match_metadata(api_custom_properties: dict[str, str], metadata: dict[str, Any]) -> bool:
     """
     Compare API custom properties with metadata.json values.
@@ -345,130 +299,6 @@ def get_metadata_from_catalog_pod(model_catalog_pod: Pod, model_name: str) -> di
         raise
 
 
-def validate_filter_options_structure(
-    response: dict[Any, Any], expected_properties: set[str] | None = None
-) -> Tuple[bool, List[str]]:
-    """
-    Comprehensive validation of filter_options response structure.
-
-    Validates:
-    - Top-level structure (filters object)
-    - All property types and their required fields
-    - Core properties presence (if specified)
-    - String properties: type, values array, distinct values
-    - Numeric properties: type, range object, min/max validity
-
-    Args:
-        response: The API response to validate
-        expected_properties: Optional set of core properties that must be present
-
-    Returns:
-        Tuple of (is_valid, list_of_errors)
-    """
-    errors = []
-
-    # Validate top-level structure
-    if not isinstance(response, dict):
-        errors.append("Response should be a dictionary")
-        return False, errors
-
-    if "filters" not in response:
-        errors.append("Response should contain 'filters' object")
-        return False, errors
-
-    filters = response["filters"]
-    if not isinstance(filters, dict):
-        errors.append("Filters should be a dictionary")
-        return False, errors
-
-    if not filters:
-        errors.append("Filters object should not be empty")
-        return False, errors
-
-    # Validate expected core properties if specified
-    if expected_properties:
-        for prop in expected_properties:
-            if prop not in filters:
-                errors.append(f"Core property '{prop}' should be present in filter options")
-
-    # Validate each property structure
-    for prop_name, prop_data in filters.items():
-        if not isinstance(prop_data, dict):
-            errors.append(f"Property '{prop_name}' should be a dictionary")
-            continue
-
-        if "type" not in prop_data:
-            errors.append(f"Property '{prop_name}' should have 'type' field")
-            continue
-
-        prop_type = prop_data["type"]
-        if not isinstance(prop_type, str) or not prop_type.strip():
-            errors.append(f"Type for '{prop_name}' should be a non-empty string")
-            continue
-
-        # Validate string properties
-        if prop_type == "string":
-            if "values" not in prop_data:
-                errors.append(f"String property '{prop_name}' should have 'values' array")
-                continue
-
-            values = prop_data["values"]
-            if not isinstance(values, list):
-                errors.append(f"Values for '{prop_name}' should be a list")
-                continue
-
-            if not values:
-                errors.append(f"Values array for '{prop_name}' should not be empty")
-                continue
-
-            # Validate individual values
-            for i, value in enumerate(values):
-                if not isinstance(value, str):
-                    errors.append(f"Value at index {i} for '{prop_name}' should be string, got: {type(value)}")
-                elif not value.strip():
-                    errors.append(f"Value at index {i} for '{prop_name}' should not be empty or whitespace")
-
-            # Check for distinct values (no duplicates)
-            try:
-                if len(values) != len(set(values)):
-                    errors.append(f"Values for '{prop_name}' should be distinct (found duplicates)")
-            except TypeError:
-                errors.append(f"Values for '{prop_name}' should be a list of strings, found unhashable type")
-
-        # Validate numeric properties - checking multiple type names since we don't know what the API will return
-        elif prop_type in ["number", "numeric", "float", "integer", "int"]:
-            if "range" not in prop_data:
-                errors.append(f"Numeric property '{prop_name}' should have 'range' object")
-                continue
-
-            range_obj = prop_data["range"]
-            if not isinstance(range_obj, dict):
-                errors.append(f"Range for '{prop_name}' should be a dictionary")
-                continue
-
-            # Check min/max presence
-            if "min" not in range_obj:
-                errors.append(f"Range for '{prop_name}' should have 'min' value")
-            if "max" not in range_obj:
-                errors.append(f"Range for '{prop_name}' should have 'max' value")
-
-            if "min" in range_obj and "max" in range_obj:
-                min_val = range_obj["min"]
-                max_val = range_obj["max"]
-
-                # Validate min/max are numeric
-                if not isinstance(min_val, (int, float)):
-                    errors.append(f"Min value for '{prop_name}' should be numeric, got: {type(min_val)}")
-                if not isinstance(max_val, (int, float)):
-                    errors.append(f"Max value for '{prop_name}' should be numeric, got: {type(max_val)}")
-
-                # Validate logical relationship (min <= max)
-                if isinstance(min_val, (int, float)) and isinstance(max_val, (int, float)) and min_val > max_val:
-                    errors.append(f"Min value ({min_val}) should be <= max value ({max_val}) for '{prop_name}'")
-
-    return len(errors) == 0, errors
-
-
 def compare_filter_options_with_database(
     api_filters: dict[str, Any], db_properties: dict[str, list[str]], excluded_fields: set[str]
 ) -> Tuple[bool, List[str]]:
@@ -653,19 +483,3 @@ def verify_labels_match(expected_labels: List[Dict[str, Any]], api_labels: List[
                 break
 
         assert found, f"Expected label not found in API response: {expected_label}"
-
-
-def validate_source_status(catalog: dict[str, Any], expected_status: str) -> None:
-    """
-    Validate the status field of a catalog source.
-
-    Args:
-        catalog: The catalog source dictionary from API response
-        expected_status: The expected status value (e.g., "available", "disabled", "error")
-
-    Raises:
-        AssertionError: If status field does not match expected value
-    """
-    assert catalog.get("status") == expected_status, (
-        f"Source '{catalog.get('id')}' status should be '{expected_status}', got: {catalog.get('status')}"
-    )