Add new test for Raw routs reconciliation

tests/model_serving/model_server/conftest.py

@@ -362,6 +362,29 @@ def ovms_kserve_inference_service(
         yield isvc
 
 
+@pytest.fixture(scope="class")
+def ovms_raw_inference_service(
+    request: FixtureRequest,
+    admin_client: DynamicClient,
+    model_namespace: Namespace,
+    openvino_kserve_serving_runtime: ServingRuntime,
+    ci_endpoint_s3_secret: Secret,
+) -> Generator[InferenceService, Any, Any]:
+    with create_isvc(
+        client=admin_client,
+        name=f"{request.param['name']}-raw",
+        namespace=model_namespace.name,
+        external_route=True,
+        runtime=openvino_kserve_serving_runtime.name,
+        storage_path=request.param["model-dir"],
+        storage_key=ci_endpoint_s3_secret.name,
+        model_format=ModelAndFormat.OPENVINO_IR,
+        deployment_mode=KServeDeploymentType.RAW_DEPLOYMENT,
+        model_version=request.param["model-version"],
+    ) as isvc:
+        yield isvc
+
+
 @pytest.fixture(scope="class")
 def http_s3_tensorflow_model_mesh_inference_service(
     request: FixtureRequest,

tests/model_serving/model_server/raw_deployment/test_kserve_raw_routes_reconciliation.py

@@ -0,0 +1,54 @@
+import pytest
+
+from tests.model_serving.model_server.utils import verify_inference_response
+from tests.model_serving.model_server.raw_deployment.utils import assert_ingress_status_changed
+from utilities.constants import ModelFormat, ModelVersion, Protocols, RunTimeConfigs
+from utilities.inference_utils import Inference
+from utilities.manifests.onnx import ONNX_INFERENCE_CONFIG
+
+
+pytestmark = [pytest.mark.rawdeployment, pytest.mark.usefixtures("valid_aws_config")]
+
+
+@pytest.mark.parametrize(
+    "model_namespace, openvino_kserve_serving_runtime, ovms_raw_inference_service",
+    [
+        pytest.param(
+            {"name": "kserve-raw-route-reconciliation"},
+            RunTimeConfigs.ONNX_OPSET13_RUNTIME_CONFIG,
+            {"name": ModelFormat.ONNX, "model-version": ModelVersion.OPSET13, "model-dir": "test-dir"},
+        )
+    ],
+    indirect=True,
+)
+class TestONNXRawRouteReconciliation:
+    """Test suite for  Validating reconciliation"""
+
+    @pytest.mark.smoke
+    def test_raw_onnx_rout_reconciliation(self, admin_client, ovms_raw_inference_service):
+        """
+        Verify that the KServe Raw ONNX model can be queried using REST
+        and ensure that the model rout reconciliation works correctly .
+        """
+        # Initial inference validation
+        verify_inference_response(
+            inference_service=ovms_raw_inference_service,
+            inference_config=ONNX_INFERENCE_CONFIG,
+            inference_type=Inference.INFER,
+            protocol=Protocols.HTTPS,
+            use_default_query=True,
+        )
+
+    def test_route_value_before_and_after_deletion(self, admin_client, ovms_raw_inference_service):
+        # Validate ingress status before and after route deletion
+        assert_ingress_status_changed(admin_client=admin_client, inference_service=ovms_raw_inference_service)
+
+    def test_model_works_after_route_is_recreated(self, ovms_raw_inference_service):
+        # Final inference validation after route update
+        verify_inference_response(
+            inference_service=ovms_raw_inference_service,
+            inference_config=ONNX_INFERENCE_CONFIG,
+            inference_type=Inference.INFER,
+            protocol=Protocols.HTTPS,
+            use_default_query=True,
+        )

tests/model_serving/model_server/raw_deployment/utils.py

@@ -0,0 +1,60 @@
+from kubernetes.dynamic import DynamicClient
+from kubernetes.dynamic.exceptions import ResourceNotFoundError
+from ocp_resources.inference_service import InferenceService
+from utilities.constants import Timeout
+from utilities.infra import get_model_route
+
+
+def assert_ingress_status_changed(admin_client: DynamicClient, inference_service: InferenceService) -> None:
+    """
+    Validates that the ingress status changes correctly after route deletion.
+
+    Args:
+        admin_client (DynamicClient): The administrative client used to manage the model route.
+        inference_service (InferenceService): The inference service whose route status is being checked.
+
+    Raises:
+        ResourceNotFoundError: If the route does not exist before or after deletion.
+        AssertionError: If any of the validation checks fail.
+
+    Returns:
+        None
+    """
+    route = get_model_route(admin_client, inference_service)
+    if not route.exists:
+        raise ResourceNotFoundError("Route before deletion not found: No active route is currently available.")
+
+    initial_status = route.instance.status["ingress"][0]["conditions"][0]
+    initial_host = route.host
+    initial_transition_time = initial_status["lastTransitionTime"]
+    initial_status_value = initial_status["status"]
+
+    route.delete(wait=True, timeout=Timeout.TIMEOUT_1MIN)
+
+    if not route.exists:
+        raise ResourceNotFoundError("Route after deletion not found: No active route is currently available.")
+
+    updated_status = route.instance.status["ingress"][0]["conditions"][0]
+    updated_host = route.host
+    updated_transition_time = updated_status["lastTransitionTime"]
+    updated_status_value = updated_status["status"]
+
+    # Collect failures instead of stopping at the first failed assertion
+    failures = []
+
+    if updated_host != initial_host:
+        failures.append(f"Host mismatch: before={initial_host}, after={updated_host}")
+
+    if updated_transition_time == initial_transition_time:
+        failures.append(
+            f"Transition time did not change: before={initial_transition_time}, after={updated_transition_time}"
+        )
+
+    if updated_status_value != "True":
+        failures.append(f"Updated ingress status incorrect: expected=True, actual={updated_status_value}")
+
+    if initial_status_value != "True":
+        failures.append(f"Initial ingress status incorrect: expected=True, actual={initial_status_value}")
+
+    # Assert all failures at once
+    assert not failures, "Ingress status validation failed:\n" + "\n".join(failures)

utilities/inference_utils.py

@@ -20,7 +20,7 @@
 from utilities.exceptions import InvalidStorageArgumentError
 from utilities.infra import (
     get_inference_serving_runtime,
-    get_model_mesh_route,
+    get_model_route,
     get_pods_by_isvc_label,
     get_services_by_isvc_label,
     wait_for_inference_deployment_replicas,
@@ -93,7 +93,7 @@ def get_inference_url(self) -> str:
                 return urlparse(url=url).netloc
 
             elif self.deployment_mode == KServeDeploymentType.MODEL_MESH:
-                route = get_model_mesh_route(client=self.inference_service.client, isvc=self.inference_service)
+                route = get_model_route(client=self.inference_service.client, isvc=self.inference_service)
                 return route.instance.spec.host
 
             else:

utilities/infra.py

@@ -491,8 +491,9 @@ def get_inference_serving_runtime(isvc: InferenceService) -> ServingRuntime:
     raise ResourceNotFoundError(f"{isvc.name} runtime {runtime.name} does not exist")
 
 
-def get_model_mesh_route(client: DynamicClient, isvc: InferenceService) -> Route:
+def get_model_route(client: DynamicClient, isvc: InferenceService) -> Route:
     """
+    Get model route using  InferenceService
     Args:
         client (DynamicClient): OCP Client to use.
         isvc (InferenceService):InferenceService object.