Add negative test for model server (#1120)

mwaykole · web-flow · commit b8a709fe6d3f · 2026-02-18T10:35:18.000+01:00
Signed-off-by: Milind Waykole &lt;mwaykole@redhat.com&gt;
diff --git a/tests/model_serving/model_server/kserve/negative/__init__.py b/tests/model_serving/model_server/kserve/negative/__init__.py
diff --git a/tests/model_serving/model_server/kserve/negative/conftest.py b/tests/model_serving/model_server/kserve/negative/conftest.py
@@ -0,0 +1,96 @@
+from typing import Any, Generator
+
+from urllib.parse import urlparse
+
+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.secret import Secret
+from ocp_resources.serving_runtime import ServingRuntime
+
+from utilities.constants import (
+    KServeDeploymentType,
+    RuntimeTemplates,
+)
+from utilities.inference_utils import create_isvc
+from utilities.infra import get_pods_by_isvc_label
+from utilities.serving_runtime import ServingRuntimeFromTemplate
+
+
+@pytest.fixture(scope="class")
+def ovms_serving_runtime(
+    admin_client: DynamicClient,
+    unprivileged_model_namespace: Namespace,
+) -> Generator[ServingRuntime, Any, Any]:
+    """Create OVMS serving runtime for negative tests."""
+    with ServingRuntimeFromTemplate(
+        client=admin_client,
+        name="negative-test-ovms-runtime",
+        namespace=unprivileged_model_namespace.name,
+        template_name=RuntimeTemplates.OVMS_KSERVE,
+        multi_model=False,
+        enable_http=True,
+        enable_grpc=False,
+    ) as runtime:
+        yield runtime
+
+
+@pytest.fixture(scope="class")
+def negative_test_ovms_isvc(
+    request: FixtureRequest,
+    admin_client: DynamicClient,
+    unprivileged_model_namespace: Namespace,
+    ovms_serving_runtime: ServingRuntime,
+    ci_s3_bucket_name: str,
+    ci_endpoint_s3_secret: Secret,
+) -> Generator[InferenceService, Any, Any]:
+    """Create InferenceService with OVMS runtime for negative tests."""
+    storage_uri = f"s3://{ci_s3_bucket_name}/{request.param['model-dir']}/"
+    supported_formats = ovms_serving_runtime.instance.spec.supportedModelFormats
+    if not supported_formats:
+        raise ValueError(f"ServingRuntime '{ovms_serving_runtime.name}' has no supportedModelFormats")
+
+    with create_isvc(
+        client=admin_client,
+        name="negative-test-ovms-isvc",
+        namespace=unprivileged_model_namespace.name,
+        runtime=ovms_serving_runtime.name,
+        storage_key=ci_endpoint_s3_secret.name,
+        storage_path=urlparse(storage_uri).path,
+        model_format=supported_formats[0].name,
+        deployment_mode=KServeDeploymentType.RAW_DEPLOYMENT,
+        external_route=True,
+    ) as isvc:
+        yield isvc
+
+
+@pytest.fixture(scope="class")
+def initial_pod_state(
+    admin_client: DynamicClient,
+    negative_test_ovms_isvc: InferenceService,
+) -> dict[str, dict[str, Any]]:
+    """Capture initial pod state (UIDs, restart counts) before tests run.
+
+    Returns:
+        A dictionary mapping pod UIDs to their initial state including
+        name, restart counts per container.
+    """
+    pods = get_pods_by_isvc_label(
+        client=admin_client,
+        isvc=negative_test_ovms_isvc,
+    )
+
+    pod_state: dict[str, dict[str, Any]] = {}
+    for pod in pods:
+        uid = pod.instance.metadata.uid
+        container_restart_counts = {
+            container.name: container.restartCount for container in (pod.instance.status.containerStatuses or [])
+        }
+        pod_state[uid] = {
+            "name": pod.name,
+            "restart_counts": container_restart_counts,
+        }
+
+    return pod_state
diff --git a/tests/model_serving/model_server/kserve/negative/test_invalid_inference_requests.py b/tests/model_serving/model_server/kserve/negative/test_invalid_inference_requests.py
@@ -0,0 +1,147 @@
+"""Tests for invalid inference requests handling.
+
+This module verifies that KServe properly handles inference requests with
+unsupported Content-Type headers, returning appropriate error responses.
+
+Jira: RHOAIENG-48283
+"""
+
+from http import HTTPStatus
+from typing import Any
+
+import pytest
+from kubernetes.dynamic import DynamicClient
+from ocp_resources.inference_service import InferenceService
+
+from tests.model_serving.model_server.kserve.negative.utils import (
+    send_inference_request_with_content_type,
+)
+from utilities.infra import get_pods_by_isvc_label
+
+
+pytestmark = pytest.mark.usefixtures("valid_aws_config")
+
+
+@pytest.mark.jira("RHOAIENG-48283", run=False)
+@pytest.mark.tier1
+@pytest.mark.rawdeployment
+@pytest.mark.parametrize(
+    "unprivileged_model_namespace, negative_test_ovms_isvc",
+    [
+        pytest.param(
+            {"name": "negative-test-content-type"},
+            {"model-dir": "test-dir"},
+        )
+    ],
+    indirect=True,
+)
+class TestUnsupportedContentType:
+    """Test class for verifying error handling when using unsupported Content-Type headers.
+
+    Preconditions:
+        - InferenceService deployed and ready
+        - Model accepts application/json content type
+
+    Test Steps:
+        1. Create InferenceService with OVMS runtime
+        2. Wait for InferenceService status = Ready
+        3. Send POST to inference endpoint with header Content-Type: text/xml
+        4. Send POST with header Content-Type: application/x-www-form-urlencoded
+        5. Capture responses for both requests
+        6. Verify model pod health status
+
+    Expected Results:
+        - HTTP Status Code: 415 Unsupported Media Type for invalid Content-Types
+        - Error indicates expected content type is application/json
+        - Model pod remains healthy (Running, no restarts)
+    """
+
+    VALID_INFERENCE_BODY: dict[str, Any] = {
+        "inputs": [
+            {
+                "name": "Input3",
+                "shape": [1, 1, 28, 28],
+                "datatype": "FP32",
+                "data": [0.0] * 784,
+            }
+        ]
+    }
+
+    @pytest.mark.parametrize(
+        "content_type",
+        [
+            pytest.param("text/xml", id="text_xml"),
+            pytest.param("application/x-www-form-urlencoded", id="form_urlencoded"),
+        ],
+    )
+    def test_unsupported_content_type_returns_415(
+        self,
+        negative_test_ovms_isvc: InferenceService,
+        content_type: str,
+    ) -> None:
+        """Verify that unsupported Content-Type headers return 415 status code.
+
+        Given an InferenceService is deployed and ready
+        When sending a POST request with an unsupported Content-Type header
+        Then the response should have HTTP status code 415 (Unsupported Media Type)
+        """
+        status_code, response_body = send_inference_request_with_content_type(
+            inference_service=negative_test_ovms_isvc,
+            content_type=content_type,
+            body=self.VALID_INFERENCE_BODY,
+        )
+
+        assert status_code == HTTPStatus.UNSUPPORTED_MEDIA_TYPE, (
+            f"Expected 415 Unsupported Media Type for Content-Type '{content_type}', "
+            f"got {status_code}. Response: {response_body}"
+        )
+
+    def test_model_pod_remains_healthy_after_invalid_requests(
+        self,
+        admin_client: DynamicClient,
+        negative_test_ovms_isvc: InferenceService,
+        initial_pod_state: dict[str, dict[str, Any]],
+    ) -> None:
+        """Verify that the model pod remains healthy after receiving invalid requests.
+
+        Given an InferenceService is deployed and ready
+        When sending requests with unsupported Content-Type headers
+        Then the same pods (by UID) should still be running without additional restarts
+        """
+        send_inference_request_with_content_type(
+            inference_service=negative_test_ovms_isvc,
+            content_type="text/xml",
+            body=self.VALID_INFERENCE_BODY,
+        )
+
+        current_pods = get_pods_by_isvc_label(
+            client=admin_client,
+            isvc=negative_test_ovms_isvc,
+        )
+
+        assert len(current_pods) > 0, "No pods found for the InferenceService"
+
+        current_pod_uids = {pod.instance.metadata.uid for pod in current_pods}
+        initial_pod_uids = set(initial_pod_state.keys())
+
+        assert current_pod_uids == initial_pod_uids, (
+            f"Pod UIDs changed after invalid requests. "
+            f"Initial: {initial_pod_uids}, Current: {current_pod_uids}. "
+            f"This indicates pods were recreated."
+        )
+
+        for pod in current_pods:
+            uid = pod.instance.metadata.uid
+            initial_state = initial_pod_state[uid]
+
+            assert pod.instance.status.phase == "Running", (
+                f"Pod {pod.name} is not running, status: {pod.instance.status.phase}"
+            )
+
+            container_statuses = pod.instance.status.containerStatuses or []
+            for container in container_statuses:
+                initial_restart_count = initial_state["restart_counts"].get(container.name, 0)
+                assert container.restartCount == initial_restart_count, (
+                    f"Container {container.name} in pod {pod.name} restarted after invalid requests. "
+                    f"Initial count: {initial_restart_count}, Current count: {container.restartCount}"
+                )
diff --git a/tests/model_serving/model_server/kserve/negative/utils.py b/tests/model_serving/model_server/kserve/negative/utils.py
@@ -0,0 +1,56 @@
+"""Utility functions for negative inference tests."""
+
+import json
+import shlex
+from typing import Any
+
+from ocp_resources.inference_service import InferenceService
+from pyhelper_utils.shell import run_command
+
+
+def send_inference_request_with_content_type(
+    inference_service: InferenceService,
+    content_type: str,
+    body: dict[str, Any],
+) -> tuple[int, str]:
+    """Send an inference request with a specific Content-Type header.
+
+    This function is used for negative testing to verify error handling
+    when sending requests with unsupported Content-Type headers.
+
+    Args:
+        inference_service: The InferenceService to send the request to.
+        content_type: The Content-Type header value to use.
+        body: The request body to send.
+
+    Returns:
+        A tuple of (status_code, response_body).
+
+    Raises:
+        ValueError: If the InferenceService has no URL or curl output is malformed.
+    """
+    url = inference_service.instance.status.url
+    if not url:
+        raise ValueError(f"InferenceService '{inference_service.name}' has no URL; is it Ready?")
+
+    endpoint = f"{url}/v2/models/{inference_service.name}/infer"
+
+    cmd = (
+        f"curl -s -w '\\n%{{http_code}}' "
+        f"-X POST {endpoint} "
+        f"-H 'Content-Type: {content_type}' "
+        f"-d '{json.dumps(body)}' "
+        f"--insecure"
+    )
+
+    _, out, _ = run_command(command=shlex.split(cmd), verify_stderr=False, check=False)
+
+    lines = out.strip().split("\n")
+    try:
+        status_code = int(lines[-1])
+    except ValueError as exc:
+        raise ValueError(f"Could not parse HTTP status code from curl output: {out!r}") from exc
+
+    response_body = "\n".join(lines[:-1])
+
+    return status_code, response_body
