From a459ac932fc6ba4c22ac0844582c0127b9b1f840 Mon Sep 17 00:00:00 2001 From: Yujong Lee Date: Tue, 8 Sep 2026 08:00:57 -0700 Subject: [PATCH] fix(guardrails): validate logging-only mode with production integration coverage --- litellm/integrations/custom_guardrail.py | 10 +- .../integrations/test_custom_guardrail.py | 26 ++- tests/test_litellm_rust/test_integrations.py | 166 +++++++++++++++++- 3 files changed, 198 insertions(+), 4 deletions(-) diff --git a/litellm/integrations/custom_guardrail.py b/litellm/integrations/custom_guardrail.py index 2d66a280663..6befbab72e9 100644 --- a/litellm/integrations/custom_guardrail.py +++ b/litellm/integrations/custom_guardrail.py @@ -601,6 +601,12 @@ class CustomGuardrail(CustomLogger): event_hook: GuardrailEventHooks | list[GuardrailEventHooks] | Mode | None, supported_event_hooks: list[GuardrailEventHooks], ) -> None: + allowed_hooks: Final = frozenset(supported_event_hooks) | ( + frozenset((GuardrailEventHooks.logging_only,)) + if self.uses_apply_guardrail_interface() and not self.use_native_lifecycle_hooks + else frozenset() + ) + def _validate_event_hook_list_is_in_supported_event_hooks( event_hook: list[GuardrailEventHooks] | list[str], supported_event_hooks: list[GuardrailEventHooks], @@ -608,7 +614,7 @@ class CustomGuardrail(CustomLogger): for hook in event_hook: if isinstance(hook, str): hook = GuardrailEventHooks(hook) - if hook not in supported_event_hooks: + if hook not in allowed_hooks: raise ValueError(f"Event hook {hook} is not in the supported event hooks {supported_event_hooks}") if event_hook is None: @@ -629,7 +635,7 @@ class CustomGuardrail(CustomLogger): default_list = event_hook.default if isinstance(event_hook.default, list) else [event_hook.default] _validate_event_hook_list_is_in_supported_event_hooks(default_list, supported_event_hooks) elif isinstance(event_hook, GuardrailEventHooks): - if event_hook not in supported_event_hooks: + if event_hook not in allowed_hooks: raise ValueError(f"Event hook {event_hook} is not in the supported event hooks {supported_event_hooks}") @staticmethod diff --git a/tests/test_litellm/integrations/test_custom_guardrail.py b/tests/test_litellm/integrations/test_custom_guardrail.py index 885bd1d4d72..baf3c6ae41f 100644 --- a/tests/test_litellm/integrations/test_custom_guardrail.py +++ b/tests/test_litellm/integrations/test_custom_guardrail.py @@ -1,5 +1,5 @@ import asyncio -from typing import TYPE_CHECKING, Literal, Optional +from typing import TYPE_CHECKING, Final, Literal, Optional from unittest.mock import AsyncMock import pytest @@ -2383,6 +2383,30 @@ class TestLoggingOnlyApplyGuardrail: apply_guardrail must still run against the logged request and response and record guardrail_information, instead of inheriting the CustomLogger no-op.""" + @pytest.mark.asyncio + async def test_content_filter_accepts_logging_only_and_records_detection(self): + from litellm.proxy.guardrails.guardrail_hooks.litellm_content_filter.content_filter import ( + ContentFilterGuardrail, + ) + from litellm.types.guardrails import BlockedWord, ContentFilterAction, GuardrailEventHooks + + guardrail: Final = ContentFilterGuardrail( + guardrail_name="content-review", + event_hook=GuardrailEventHooks.logging_only, + default_on=True, + blocked_words=[BlockedWord(keyword="hello", action=ContentFilterAction.BLOCK)], + ) + kwargs, response = _logged_call([{"role": "user", "content": "hello there"}]) + + out_kwargs, out_response = await guardrail.async_logging_hook(kwargs, response, CallTypes.acompletion.value) + + assert out_response is response + assert out_kwargs["messages"] == kwargs["messages"] + assert ( + out_kwargs["standard_logging_object"]["guardrail_information"][0]["guardrail_status"] + == "guardrail_intervened" + ) + @pytest.mark.asyncio async def test_runs_apply_guardrail_observe_only_and_records_verdict(self): guardrail = _ApplyOnlyObserver() diff --git a/tests/test_litellm_rust/test_integrations.py b/tests/test_litellm_rust/test_integrations.py index e2afcb57d36..4f9447bcb6f 100644 --- a/tests/test_litellm_rust/test_integrations.py +++ b/tests/test_litellm_rust/test_integrations.py @@ -2,12 +2,17 @@ import json from typing import Final import pytest +from fastapi import HTTPException from opentelemetry.trace import StatusCode from prometheus_client import CollectorRegistry, Counter import litellm +from litellm.integrations.generic_api.generic_api_callback import GenericAPILogger from litellm.integrations.prometheus import PrometheusLogger from litellm.litellm_core_utils import litellm_logging +from litellm.proxy.guardrails.guardrail_hooks.generic_guardrail_api.generic_guardrail_api import GenericGuardrailAPI +from litellm.proxy.guardrails.guardrail_hooks.litellm_content_filter.content_filter import ContentFilterGuardrail +from litellm.types.guardrails import BlockedWord, ContentFilterAction, GuardrailEventHooks from litellm.types.utils import CallTypes from tests._prometheus_helpers import isolated_prometheus_registry from tests.test_litellm_rust.callback_recorder import RecordingLogger, drain_logging @@ -32,6 +37,165 @@ pytestmark = pytest.mark.requires_rust_extension FAILURE_RESPONSE: Final = ResponseSpec(body={"message": "provider unavailable"}, status=500) +@pytest.mark.asyncio +@pytest.mark.parametrize("route", ASYNC_ROUTES, ids=route_id) +async def test_generic_api_logger_exports_success_over_http(route: Route, provider: RecordingServer) -> None: + provider.expected_requests = 2 + logger: Final = GenericAPILogger(endpoint=f"{provider.base_url}/logs", batch_size=1, log_format="single") + recorder: Final = RecordingLogger() + + await route.invoke(provider, callbacks=[logger, recorder]) + await recorder.wait_for_async("async_log_success_event") + + exports: Final = [request for request in provider.requests if request.path == "/logs"] + assert len(exports) == 1 + payload: Final = exports[0].body + assert payload["status"] == "success" + assert payload["call_type"] == route.call_type + assert payload["model"] == route.provider_model + assert payload["response_cost"] == pytest.approx(route.expected_cost) + assert route.response_text in json.dumps(payload["response"]) + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "route", + ( + OCR_ASYNC, + pytest.param( + MESSAGES_ROUTE, + marks=pytest.mark.xfail( + strict=True, + raises=pytest.fail.Exception, + reason="Native Messages provider failure raises TypeError and falls back to a second Python request", + ), + ), + ), + ids=route_id, +) +async def test_generic_api_logger_exports_provider_failure_over_http(route: Route, provider: RecordingServer) -> None: + provider.expected_requests = None + provider.enqueue(FAILURE_RESPONSE) + logger: Final = GenericAPILogger(endpoint=f"{provider.base_url}/logs", batch_size=1, log_format="single") + recorder: Final = RecordingLogger() + + try: + with pytest.raises(litellm.InternalServerError): + await route.invoke(provider, callbacks=[logger, recorder], num_retries=0) + finally: + await drain_logging() + await recorder.wait_for_async("async_log_failure_event") + + exports: Final = [request for request in provider.requests if request.path == "/logs"] + assert len(provider.requests) == 2 + assert len(exports) == 1 + payload: Final = exports[0].body + assert payload["status"] == "failure" + assert payload["call_type"] == route.call_type + assert payload["error_information"]["error_class"] == "InternalServerError" + assert payload["error_information"]["error_code"] == "500" + + +@pytest.mark.asyncio +@pytest.mark.parametrize("route", NON_STREAM_ASYNC_ROUTES, ids=route_id) +@pytest.mark.parametrize( + ("verdict", "expected_status"), + ( + (ResponseSpec(body={"action": "NONE"}), "success"), + (ResponseSpec(body={"action": "BLOCKED", "blocked_reason": "policy violation"}), "guardrail_intervened"), + (ResponseSpec(body={"action": "GUARDRAIL_INTERVENED", "texts": ["redacted"]}), "success"), + (FAILURE_RESPONSE, "guardrail_failed_to_respond"), + ), + ids=("allow", "block", "rewrite", "unavailable"), +) +async def test_generic_guardrail_logging_only_verdict_is_exported_over_http( + route: Route, provider: RecordingServer, verdict: ResponseSpec, expected_status: str +) -> None: + provider.expected_requests = 3 + provider.enqueue(ResponseSpec(body=route.provider_response)) + provider.enqueue(verdict) + provider.enqueue(ResponseSpec(body={})) + guardrail: Final = GenericGuardrailAPI( + api_base=provider.base_url, + guardrail_name="http-review", + event_hook=GuardrailEventHooks.logging_only, + default_on=True, + ) + logger: Final = GenericAPILogger(endpoint=f"{provider.base_url}/logs", batch_size=1, log_format="single") + recorder: Final = RecordingLogger() + + response: Final = await route.invoke(provider, callbacks=[guardrail, logger, recorder]) + await recorder.wait_for_async("async_log_success_event") + + scan: Final = provider.requests[1] + assert scan.path == "/beta/litellm_basic_guardrail_api" + assert scan.body["input_type"] == route.logging_only_scan[0] + assert scan.body["texts"] == list(route.logging_only_scan[1]) + assert provider.requests[2].path == "/logs" + payload: Final = provider.requests[2].body + assert payload["status"] == "success" + assert [(entry["guardrail_name"], entry["guardrail_status"]) for entry in payload["guardrail_information"]] == [ + ("http-review", expected_status) + ] + assert route.response_text in json.dumps(payload["response"]) + if route == OCR_ASYNC: + assert response.pages[0].markdown == route.response_text + else: + assert response["content"][0]["text"] == route.response_text + assert "guardrails" not in provider.requests[0].body + + +@pytest.mark.asyncio +@pytest.mark.parametrize("route", NON_STREAM_ASYNC_ROUTES, ids=route_id) +@pytest.mark.parametrize("action", (ContentFilterAction.BLOCK, ContentFilterAction.MASK)) +async def test_content_filter_logging_only_detects_real_content_without_changing_response( + route: Route, provider: RecordingServer, action: ContentFilterAction +) -> None: + guardrail: Final = ContentFilterGuardrail( + guardrail_name="content-review", + event_hook=GuardrailEventHooks.logging_only, + default_on=True, + blocked_words=[BlockedWord(keyword=route.logging_only_scan[1][0], action=action)], + ) + recorder: Final = RecordingLogger() + + response: Final = await route.invoke(provider, callbacks=[guardrail, recorder]) + payload: Final = (await recorder.wait_for_async("async_log_success_event"))[0].kwargs["standard_logging_object"] + + assert len(payload["guardrail_information"]) == 1 + verdict: Final = payload["guardrail_information"][0] + assert verdict["guardrail_name"] == "content-review" + assert verdict["guardrail_status"] == ("guardrail_intervened" if action == ContentFilterAction.BLOCK else "success") + assert verdict["guardrail_response"] == [ + {"action": action.value, "keyword": route.logging_only_scan[1][0].lower(), "type": "blocked_word"} + ] + assert route.response_text in json.dumps(payload["response"]) + if route == OCR_ASYNC: + assert response.pages[0].markdown == route.response_text + else: + assert response["content"][0]["text"] == route.response_text + + +@pytest.mark.asyncio +@pytest.mark.parametrize("route", NON_STREAM_ASYNC_ROUTES, ids=route_id) +@pytest.mark.xfail( + strict=True, + raises=pytest.fail.Exception, + reason="CustomGuardrail deployment post-call hook does not dispatch to production apply_guardrail implementations", +) +async def test_content_filter_post_call_blocks_provider_response(route: Route, provider: RecordingServer) -> None: + guardrail: Final = ContentFilterGuardrail( + guardrail_name="enforced-content-review", + event_hook=GuardrailEventHooks.post_call, + blocked_words=[BlockedWord(keyword=route.response_text, action=ContentFilterAction.BLOCK)], + ) + litellm.callbacks.append(guardrail) + + with pytest.raises(HTTPException, match="Content blocked") as blocked: + await route.invoke(provider, guardrails=["enforced-content-review"]) + assert blocked.value.status_code == 400 + + def test_prometheus_registry_restores_collectors_after_failure() -> None: registry: Final = CollectorRegistry() original: Final = Counter("original", "Original collector", registry=registry) @@ -174,7 +338,7 @@ async def test_prometheus_counts_one_failed_request( @pytest.mark.parametrize("route", NON_STREAM_ASYNC_ROUTES, ids=route_id) async def test_prometheus_by_string_name_is_initialized_once(route: Route, provider: RecordingServer) -> None: provider.expected_requests = 2 - litellm.success_callback = ["prometheus"] # test-quality-ok: tests public string-name registration; isolate_rust_state restores this registry + litellm.success_callback = ["prometheus"] # test-quality-ok: public registration; fixture restores globals recorder: Final = RecordingLogger() await route.invoke(provider, callbacks=[recorder])