automate isvc model status

tests/model_serving/model_server/components/conftest.py

@@ -0,0 +1,50 @@
+import pytest
+from _pytest.fixtures import FixtureRequest
+from kubernetes.dynamic import DynamicClient
+from ocp_resources.inference_service import InferenceService
+from ocp_resources.namespace import Namespace
+from ocp_resources.resource import ResourceEditor
+from ocp_resources.service_account import ServiceAccount
+from ocp_resources.serving_runtime import ServingRuntime
+
+from tests.model_serving.model_server.utils import create_isvc
+
+
+@pytest.fixture(scope="class")
+def invalid_s3_models_inference_service(
+    request: FixtureRequest,
+    admin_client: DynamicClient,
+    model_namespace: Namespace,
+    serving_runtime_from_template: ServingRuntime,
+    models_s3_bucket_name: str,
+    model_service_account: ServiceAccount,
+) -> InferenceService:
+    with create_isvc(
+        client=admin_client,
+        name=request.param["name"],
+        namespace=model_namespace.name,
+        runtime=serving_runtime_from_template.name,
+        storage_uri=f"s3://{models_s3_bucket_name}/non-existing-path/",
+        model_format=serving_runtime_from_template.instance.spec.supportedModelFormats[0].name,
+        model_service_account=model_service_account.name,
+        deployment_mode=request.param["deployment-mode"],
+        wait=False,
+        wait_for_predictor_pods=False,
+    ) as isvc:
+        yield isvc
+
+
+@pytest.fixture
+def updated_s3_models_inference_service(
+    invalid_s3_models_inference_service: InferenceService, s3_models_storage_uri: str
+) -> InferenceService:
+    with ResourceEditor(
+        patches={
+            invalid_s3_models_inference_service: {
+                "spec": {
+                    "predictor": {"model": {"storageUri": s3_models_storage_uri}},
+                }
+            }
+        }
+    ):
+        yield invalid_s3_models_inference_service

tests/model_serving/model_server/components/test_custom_resources.py

@@ -0,0 +1,78 @@
+import pytest
+from ocp_resources.inference_service import InferenceService
+from simple_logger.logger import get_logger
+from timeout_sampler import TimeoutExpiredError, TimeoutSampler
+
+from utilities.constants import (
+    KServeDeploymentType,
+    ModelStoragePath,
+    Protocols,
+    ModelInferenceRuntime,
+    RuntimeTemplates,
+)
+
+LOGGER = get_logger(name=__name__)
+
+pytestmark = pytest.mark.usefixtures("valid_aws_config")
+
+
+def wait_for_isvc_model_status(isvc: InferenceService, target_model_state: str, transition_status: str) -> None:
+    LOGGER.info(
+        f"Wait for {isvc.name} target model state {target_model_state} and transition status {transition_status}."
+    )
+
+    samples = TimeoutSampler(wait_timeout=60 * 10, sleep=5, func=lambda: isvc.instance.status.modelStatus)
+
+    sample = None
+    try:
+        for sample in samples:
+            if sample.states.targetModelState == target_model_state and sample.transitionStatus == transition_status:
+                return
+
+    except TimeoutExpiredError:
+        LOGGER.error(f"Status of {isvc.name} is {sample}")
+        raise
+
+
+@pytest.mark.serverless
+@pytest.mark.jira("RHOAIENG-10765")
+@pytest.mark.parametrize(
+    "model_namespace, serving_runtime_from_template, invalid_s3_models_inference_service",
+    [
+        pytest.param(
+            {"name": "non-existing-models-storage-path"},
+            {
+                "name": f"{Protocols.HTTP}-{ModelInferenceRuntime.CAIKIT_TGIS_RUNTIME}",
+                "template-name": RuntimeTemplates.CAIKIT_TGIS_SERVING,
+                "multi-model": False,
+                "enable-http": True,
+            },
+            {
+                "name": "non-existing-models-storage-path",
+                "deployment-mode": KServeDeploymentType.SERVERLESS,
+            },
+        )
+    ],
+    indirect=True,
+)
+class TestInferenceServiceCustomResources:
+    def test_isvc_with_invalid_models_s3_path(self, invalid_s3_models_inference_service):
+        """Test ISVC status with invalid models storage path"""
+        wait_for_isvc_model_status(
+            isvc=invalid_s3_models_inference_service,
+            target_model_state="FailedToLoad",
+            transition_status="BlockedByFailedLoad",
+        )
+
+    @pytest.mark.parametrize(
+        "s3_models_storage_uri",
+        [pytest.param({"model-dir": ModelStoragePath.FLAN_T5_SMALL})],
+        indirect=True,
+    )
+    def test_isvc_with_updated_valid_models_s3_path(self, s3_models_storage_uri, updated_s3_models_inference_service):
+        """Test inference status after updating the model storage path"""
+        wait_for_isvc_model_status(
+            isvc=updated_s3_models_inference_service,
+            target_model_state="Loaded",
+            transition_status="UpToDate",
+        )