feat: add tests for guardrails builtin detectors (#357)

* feat: add tests for guardrails builtin detectors

* fix: comments

* fix: missing fstring

* feat: add helper function to check detection field of response

* feat: add helper function to check detection field of response

* fix: unify negative scenarios, add temperature (0.0) to request, improve logging and input msgs

* fix: remove unnecessary admin_client from signature and fix docstring spacing

Author: adolfo-ab

tests/model_explainability/guardrails/conftest.py

@@ -14,6 +14,7 @@
 from ocp_resources.service import Service
 from ocp_resources.serving_runtime import ServingRuntime
 
+from utilities.certificates_utils import create_ca_bundle_file
 from utilities.constants import (
     KServeDeploymentType,
     Labels,
@@ -29,21 +30,6 @@
 GUARDRAILS_ORCHESTRATOR_PORT: int = 8032
 
 
-@pytest.fixture(scope="class")
-def guardrails_orchestrator_health_route(
-    admin_client: DynamicClient,
-    model_namespace: Namespace,
-    guardrails_orchestrator: GuardrailsOrchestrator,
-) -> Generator[Route, Any, Any]:
-    route = Route(
-        name=f"{guardrails_orchestrator.name}-health",
-        namespace=guardrails_orchestrator.namespace,
-        wait_for_resource=True,
-        ensure_exists=True,
-    )
-    yield route
-
-
 @pytest.fixture(scope="class")
 def guardrails_orchestrator(
     admin_client: DynamicClient,
@@ -67,6 +53,32 @@ def guardrails_orchestrator(
         yield gorch
 
 
+@pytest.fixture(scope="class")
+def guardrails_orchestrator_health_route(
+    admin_client: DynamicClient,
+    model_namespace: Namespace,
+    guardrails_orchestrator: GuardrailsOrchestrator,
+) -> Generator[Route, Any, Any]:
+    yield Route(
+        name=f"{guardrails_orchestrator.name}-health",
+        namespace=guardrails_orchestrator.namespace,
+        wait_for_resource=True,
+    )
+
+
+@pytest.fixture(scope="class")
+def guardrails_orchestrator_route(
+    admin_client: DynamicClient,
+    model_namespace: Namespace,
+    guardrails_orchestrator: GuardrailsOrchestrator,
+) -> Generator[Route, Any, Any]:
+    yield Route(
+        name=f"{guardrails_orchestrator.name}",
+        namespace=guardrails_orchestrator.namespace,
+        wait_for_resource=True,
+    )
+
+
 @pytest.fixture(scope="class")
 def guardrails_orchestrator_pod(
     admin_client: DynamicClient, model_namespace: Namespace, guardrails_orchestrator: GuardrailsOrchestrator
@@ -204,3 +216,10 @@ def guardrails_gateway_config(
         },
     ) as cm:
         yield cm
+
+
+@pytest.fixture(scope="class")
+def openshift_ca_bundle_file(
+    admin_client: DynamicClient,
+) -> str:
+    return create_ca_bundle_file(client=admin_client, ca_type="openshift")

tests/model_explainability/guardrails/test_guardrails.py

@@ -2,10 +2,41 @@
 
 import pytest
 import requests
+from simple_logger.logger import get_logger
 from timeout_sampler import retry
 
+from tests.model_explainability.guardrails.utils import (
+    verify_builtin_detector_unsuitable_input_response,
+    verify_negative_detection_response,
+    verify_builtin_detector_unsuitable_output_response,
+    get_auth_headers,
+    get_chat_payload,
+)
 from tests.model_explainability.utils import validate_tai_component_images
 from utilities.constants import Timeout, MinIo
+from utilities.plugins.constant import OpenAIEnpoints
+
+LOGGER = get_logger(name=__name__)
+EXAMPLE_EMAIL_ADDRESS: str = "myemail@domain.com"
+INPUT_WITH_EMAIL_ADDRESS: str = f"This is my email address: {EXAMPLE_EMAIL_ADDRESS}, just answer ACK."
+PII_ENDPOINT: str = "/pii"
+
+
+@pytest.mark.parametrize(
+    "model_namespace",
+    [
+        pytest.param(
+            {"name": "test-guardrails-image"},
+        )
+    ],
+    indirect=True,
+)
+@pytest.mark.smoke
+def test_validate_guardrails_orchestrator_images(guardrails_orchestrator_pod, trustyai_operator_configmap):
+    """Test to verify Guardrails pod images.
+    Checks if the image tag from the ConfigMap is used within the Pod and if it's pinned using a sha256 digest.
+    """
+    validate_tai_component_images(pod=guardrails_orchestrator_pod, tai_operator_configmap=trustyai_operator_configmap)
 
 
 @pytest.mark.parametrize(
@@ -21,8 +52,21 @@
 )
 @pytest.mark.rawdeployment
 @pytest.mark.smoke
-class TestGuardrailsOrchestrator:
-    def test_guardrails_health_endpoint(self, admin_client, qwen_isvc, guardrails_orchestrator_health_route):
+class TestGuardrailsOrchestratorWithBuiltInDetectors:
+    """
+    Tests that the basic functionality of the GuardrailsOrchestrator work properly with the built-in (regex) detectors.
+        1. Deploy an LLM using vLLM as a SR.
+        2. Deploy the Guardrails Orchestrator.
+        3. Check that the Orchestrator is healthy by querying the health and info endpoints of its /health route.
+        4. Check that the built-in regex detectors work as expected:
+         4.1. Unsuitable input detection.
+         4.2. Unsuitable output detection.
+         4.3. No detection.
+        5. Check that the /passthrough endpoint forwards the
+         query directly to the model without performing any detection.
+    """
+
+    def test_guardrails_health_endpoint(self, qwen_isvc, guardrails_orchestrator_health_route):
         # It takes a bit for the endpoint to come online, so we retry for a brief period of time
         @retry(wait_timeout=Timeout.TIMEOUT_1MIN, sleep=1)
         def check_health_endpoint():
@@ -34,7 +78,7 @@ def check_health_endpoint():
         response = check_health_endpoint()
         assert "fms-guardrails-orchestr8" in response.text
 
-    def test_guardrails_info_endpoint(self, admin_client, qwen_isvc, guardrails_orchestrator_health_route):
+    def test_guardrails_info_endpoint(self, qwen_isvc, guardrails_orchestrator_health_route):
         response = requests.get(url=f"https://{guardrails_orchestrator_health_route.host}/info", verify=False)
         assert response.status_code == http.HTTPStatus.OK
 
@@ -43,10 +87,66 @@ def test_guardrails_info_endpoint(self, admin_client, qwen_isvc, guardrails_orch
         assert response_data["services"]["chat_generation"]["status"] == healthy_status
         assert response_data["services"]["regex"]["status"] == healthy_status
 
-    def test_validate_guardrails_orchestrator_images(self, guardrails_orchestrator_pod, trustyai_operator_configmap):
-        """Test to verify Guardrails pod images.
-        Checks if the image tag from the ConfigMap is used within the Pod and if it's pinned using a sha256 digest.
-        """
-        validate_tai_component_images(
-            pod=guardrails_orchestrator_pod, tai_operator_configmap=trustyai_operator_configmap
+    def test_guardrails_builtin_detectors_unsuitable_input(
+        self, current_client_token, openshift_ca_bundle_file, guardrails_orchestrator_route
+    ):
+        response = requests.post(
+            url=f"https://{guardrails_orchestrator_route.host}{PII_ENDPOINT}{OpenAIEnpoints.CHAT_COMPLETIONS}",
+            headers=get_auth_headers(token=current_client_token),
+            json=get_chat_payload(content=INPUT_WITH_EMAIL_ADDRESS),
+            verify=openshift_ca_bundle_file,
+        )
+
+        verify_builtin_detector_unsuitable_input_response(
+            response=response,
+            detector_id="regex",
+            detection_name="EmailAddress",
+            detection_type="pii",
+            detection_text=EXAMPLE_EMAIL_ADDRESS,
+        )
+
+    def test_guardrails_builtin_detectors_unsuitable_output(
+        self, current_client_token, openshift_ca_bundle_file, guardrails_orchestrator_route
+    ):
+        response = requests.post(
+            url=f"https://{guardrails_orchestrator_route.host}{PII_ENDPOINT}{OpenAIEnpoints.CHAT_COMPLETIONS}",
+            headers=get_auth_headers(token=current_client_token),
+            json=get_chat_payload(
+                content="Hi, write three and only three examples of email adresses "
+                "that I can use to create an account for an online service."
+                "Don't provide any additional explanation."
+            ),
+            verify=openshift_ca_bundle_file,
         )
+
+        verify_builtin_detector_unsuitable_output_response(
+            response=response, detector_id="regex", detection_name="EmailAddress", detection_type="pii"
+        )
+
+    @pytest.mark.parametrize(
+        "message, url_path",
+        [
+            pytest.param(
+                "What is the opposite of up?",
+                PII_ENDPOINT,
+                id="harmless_input",
+            ),
+            pytest.param(INPUT_WITH_EMAIL_ADDRESS, "/passthrough", id="pastthrough_endpoint"),
+        ],
+    )
+    def test_guardrails_builtin_detectors_negative_detection(
+        self,
+        current_client_token,
+        openshift_ca_bundle_file,
+        guardrails_orchestrator_route,
+        message,
+        url_path,
+    ):
+        response = requests.post(
+            url=f"https://{guardrails_orchestrator_route.host}{url_path}{OpenAIEnpoints.CHAT_COMPLETIONS}",
+            headers=get_auth_headers(token=current_client_token),
+            json=get_chat_payload(content=str(message)),
+            verify=openshift_ca_bundle_file,
+        )
+
+        verify_negative_detection_response(response=response)