Add negative test for modelServer

tests/model_serving/model_server/kserve/negative/conftest.py

@@ -3,7 +3,6 @@
 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
@@ -15,20 +14,58 @@
     RuntimeTemplates,
 )
 from utilities.inference_utils import create_isvc
-from utilities.infra import get_pods_by_isvc_label
+from utilities.infra import create_ns, get_pods_by_isvc_label, s3_endpoint_secret
 from utilities.serving_runtime import ServingRuntimeFromTemplate
 
 
-@pytest.fixture(scope="class")
+@pytest.fixture(scope="package")
+def negative_test_namespace(
+    admin_client: DynamicClient,
+    unprivileged_client: DynamicClient,
+) -> Generator[Namespace, Any, Any]:
+    """Create a shared namespace for all negative tests."""
+    with create_ns(
+        admin_client=admin_client,
+        unprivileged_client=unprivileged_client,
+        name="negative-test-kserve",
+    ) as ns:
+        yield ns
+
+
+@pytest.fixture(scope="package")
+def negative_test_s3_secret(
+    unprivileged_client: DynamicClient,
+    negative_test_namespace: Namespace,
+    aws_access_key_id: str,
+    aws_secret_access_key: str,
+    ci_s3_bucket_name: str,
+    ci_s3_bucket_region: str,
+    ci_s3_bucket_endpoint: str,
+) -> Generator[Secret, Any, Any]:
+    """Create S3 secret shared across all negative tests."""
+    with s3_endpoint_secret(
+        client=unprivileged_client,
+        name="ci-bucket-secret",
+        namespace=negative_test_namespace.name,
+        aws_access_key=aws_access_key_id,
+        aws_secret_access_key=aws_secret_access_key,
+        aws_s3_region=ci_s3_bucket_region,
+        aws_s3_bucket=ci_s3_bucket_name,
+        aws_s3_endpoint=ci_s3_bucket_endpoint,
+    ) as secret:
+        yield secret
+
+
+@pytest.fixture(scope="package")
 def ovms_serving_runtime(
     admin_client: DynamicClient,
-    unprivileged_model_namespace: Namespace,
+    negative_test_namespace: Namespace,
 ) -> Generator[ServingRuntime, Any, Any]:
-    """Create OVMS serving runtime for negative tests."""
+    """Create OVMS serving runtime shared across all negative tests."""
     with ServingRuntimeFromTemplate(
         client=admin_client,
         name="negative-test-ovms-runtime",
-        namespace=unprivileged_model_namespace.name,
+        namespace=negative_test_namespace.name,
         template_name=RuntimeTemplates.OVMS_KSERVE,
         multi_model=False,
         enable_http=True,
@@ -37,27 +74,26 @@ def ovms_serving_runtime(
         yield runtime
 
 
-@pytest.fixture(scope="class")
+@pytest.fixture(scope="package")
 def negative_test_ovms_isvc(
-    request: FixtureRequest,
     admin_client: DynamicClient,
-    unprivileged_model_namespace: Namespace,
+    negative_test_namespace: Namespace,
     ovms_serving_runtime: ServingRuntime,
     ci_s3_bucket_name: str,
-    ci_endpoint_s3_secret: Secret,
+    negative_test_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']}/"
+    """Create InferenceService with OVMS runtime shared across all negative tests."""
+    storage_uri = f"s3://{ci_s3_bucket_name}/test-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,
+        namespace=negative_test_namespace.name,
         runtime=ovms_serving_runtime.name,
-        storage_key=ci_endpoint_s3_secret.name,
+        storage_key=negative_test_s3_secret.name,
         storage_path=urlparse(storage_uri).path,
         model_format=supported_formats[0].name,
         deployment_mode=KServeDeploymentType.RAW_DEPLOYMENT,

tests/model_serving/model_server/kserve/negative/test_invalid_model_name.py

@@ -0,0 +1,86 @@
+"""Tests for invalid model name in inference endpoint.
+
+Jira: RHOAIENG-48282
+"""
+
+import json
+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 (
+    VALID_OVMS_INFERENCE_BODY,
+    assert_pods_healthy,
+    send_inference_request,
+)
+
+pytestmark = pytest.mark.usefixtures("valid_aws_config")
+
+VALID_BODY_RAW = json.dumps(VALID_OVMS_INFERENCE_BODY)
+
+
+@pytest.mark.tier1
+class TestInvalidModelName:
+    """Test class for verifying error handling when targeting a non-existent model.
+
+    Preconditions:
+        - InferenceService "negative-test-ovms-isvc" deployed and ready
+        - No InferenceService with name "nonexistent-model"
+
+    Test Steps:
+        1. Create InferenceService with OVMS runtime
+        2. Wait for InferenceService status = Ready
+        3. Send inference request to /v2/models/nonexistent-model/infer
+        4. Verify error response and existing service health
+
+    Expected Results:
+        - HTTP Status Code: 404 Not Found
+        - Error message indicates model not found
+        - No impact on existing model service
+    """
+
+    def test_nonexistent_model_returns_404(
+        self,
+        negative_test_ovms_isvc: InferenceService,
+    ) -> None:
+        """Verify that inference to a non-existent model returns 404 status code.
+
+        Given an InferenceService is deployed and ready
+        When sending a POST request targeting a non-existent model name
+        Then the response should have HTTP status code 404 (Not Found)
+        """
+        status_code, response_body = send_inference_request(
+            inference_service=negative_test_ovms_isvc,
+            body=VALID_BODY_RAW,
+            model_name="nonexistent-model",
+        )
+
+        assert status_code == HTTPStatus.NOT_FOUND, (
+            f"Expected 404 Not Found for nonexistent model, got {status_code}. Response: {response_body}"
+        )
+
+    def test_existing_service_unaffected_after_invalid_model_request(
+        self,
+        admin_client: DynamicClient,
+        negative_test_ovms_isvc: InferenceService,
+        initial_pod_state: dict[str, dict[str, Any]],
+    ) -> None:
+        """Verify that the existing service remains healthy after invalid model requests.
+
+        Given an InferenceService is deployed and ready
+        When sending a request targeting a non-existent model name
+        Then the existing service pods should remain running without restarts
+        """
+        send_inference_request(
+            inference_service=negative_test_ovms_isvc,
+            body=VALID_BODY_RAW,
+            model_name="nonexistent-model",
+        )
+        assert_pods_healthy(
+            admin_client=admin_client,
+            isvc=negative_test_ovms_isvc,
+            initial_pod_state=initial_pod_state,
+        )

tests/model_serving/model_server/kserve/negative/test_malformed_json_payload.py

@@ -0,0 +1,98 @@
+"""Tests for malformed JSON payload handling in inference requests.
+
+Jira: RHOAIENG-48279
+"""
+
+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 (
+    assert_pods_healthy,
+    send_inference_request,
+)
+
+pytestmark = pytest.mark.usefixtures("valid_aws_config")
+
+MALFORMED_JSON_EXPECTED_CODES: set[int] = {
+    HTTPStatus.BAD_REQUEST,
+    HTTPStatus.PRECONDITION_FAILED,
+}
+MISSING_BRACE_BODY = '{"inputs": [{"name": "Input3"'
+TRAILING_COMMA_BODY = '{"inputs": [{"name": "Input3",}]}'
+
+
+@pytest.mark.tier1
+@pytest.mark.rawdeployment
+class TestMalformedJsonPayload:
+    """Test class for verifying error handling when receiving malformed JSON payloads.
+
+    Preconditions:
+        - InferenceService deployed with OVMS runtime (RawDeployment)
+        - Model is ready and serving
+
+    Test Steps:
+        1. Create InferenceService with OVMS runtime
+        2. Wait for InferenceService status = Ready
+        3. Send POST with malformed JSON bodies (missing brace, trailing comma, plain text)
+        4. Verify error responses and pod health
+
+    Expected Results:
+        - HTTP Status Code: 400 Bad Request or 412 Precondition Failed
+          (OVMS returns 412 for JSON parse errors)
+        - Response indicates JSON parse failure
+        - No pod crash or restart
+    """
+
+    @pytest.mark.parametrize(
+        "malformed_body",
+        [
+            pytest.param(MISSING_BRACE_BODY, id="missing_closing_brace"),
+            pytest.param(TRAILING_COMMA_BODY, id="trailing_comma"),
+            pytest.param("not json at all", id="plain_text"),
+        ],
+    )
+    def test_malformed_json_returns_error(
+        self,
+        negative_test_ovms_isvc: InferenceService,
+        malformed_body: str,
+    ) -> None:
+        """Verify that malformed JSON payloads return an error status code.
+
+        Given an InferenceService is deployed and ready
+        When sending a POST request with a malformed JSON body
+        Then the response should have HTTP status code 400 or 412
+        """
+        status_code, response_body = send_inference_request(
+            inference_service=negative_test_ovms_isvc,
+            body=malformed_body,
+        )
+
+        assert status_code in MALFORMED_JSON_EXPECTED_CODES, (
+            f"Expected 400 or 412 for malformed JSON, got {status_code}. Response: {response_body}"
+        )
+
+    def test_model_pod_remains_healthy_after_malformed_json(
+        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 malformed JSON.
+
+        Given an InferenceService is deployed and ready
+        When sending requests with malformed JSON payloads
+        Then the same pods should still be running without additional restarts
+        """
+        send_inference_request(
+            inference_service=negative_test_ovms_isvc,
+            body=MISSING_BRACE_BODY,
+        )
+        assert_pods_healthy(
+            admin_client=admin_client,
+            isvc=negative_test_ovms_isvc,
+            initial_pod_state=initial_pod_state,
+        )

tests/model_serving/model_server/kserve/negative/test_missing_required_fields.py

@@ -0,0 +1,91 @@
+"""Tests for missing required fields in inference requests.
+
+Jira: RHOAIENG-48281
+"""
+
+import json
+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 (
+    assert_pods_healthy,
+    send_inference_request,
+)
+
+pytestmark = pytest.mark.usefixtures("valid_aws_config")
+
+
+@pytest.mark.tier1
+@pytest.mark.rawdeployment
+class TestMissingRequiredFields:
+    """Test class for verifying error handling when required fields are missing.
+
+    Preconditions:
+        - InferenceService deployed with OVMS runtime
+        - Model is ready and serving
+
+    Test Steps:
+        1. Create InferenceService with OVMS runtime
+        2. Wait for InferenceService status = Ready
+        3. Send POST with empty body {}
+        4. Send POST with body missing "inputs" field
+        5. Verify error responses and pod health
+
+    Expected Results:
+        - HTTP Status Code: 400 Bad Request
+        - Error message indicates missing required field
+        - No server crash
+    """
+
+    @pytest.mark.parametrize(
+        "incomplete_body",
+        [
+            pytest.param("{}", id="empty_body"),
+            pytest.param(json.dumps({"id": "test-123"}), id="missing_inputs_field"),
+        ],
+    )
+    def test_missing_required_fields_returns_400(
+        self,
+        negative_test_ovms_isvc: InferenceService,
+        incomplete_body: str,
+    ) -> None:
+        """Verify that requests missing required fields return 400 status code.
+
+        Given an InferenceService is deployed and ready
+        When sending a POST request with missing required fields
+        Then the response should have HTTP status code 400 (Bad Request)
+        """
+        status_code, response_body = send_inference_request(
+            inference_service=negative_test_ovms_isvc,
+            body=incomplete_body,
+        )
+
+        assert status_code == HTTPStatus.BAD_REQUEST, (
+            f"Expected 400 Bad Request for incomplete payload, got {status_code}. Response: {response_body}"
+        )
+
+    def test_model_pod_remains_healthy_after_missing_fields(
+        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 incomplete requests.
+
+        Given an InferenceService is deployed and ready
+        When sending requests with missing required fields
+        Then the same pods should still be running without additional restarts
+        """
+        send_inference_request(
+            inference_service=negative_test_ovms_isvc,
+            body="{}",
+        )
+        assert_pods_healthy(
+            admin_client=admin_client,
+            isvc=negative_test_ovms_isvc,
+            initial_pod_state=initial_pod_state,
+        )