mirror of
https://github.com/BerriAI/litellm.git
synced 2026-10-11 03:38:38 +00:00
feat: add guardrail violation span attributes and fix missing spans on pre-call blocks (#28364)
- Fix missing guardrail child spans when a pre-call guardrail blocks the request before reaching the LLM provider; `async_post_call_failure_hook` now calls `_emit_guardrail_spans_from_request_data` to emit spans from `request_data["metadata"]` regardless of whether `_handle_failure` already fired - Add `guardrail_status`, `guardrail_action`, and `guardrail_violation_categories` as queryable top-level OTEL span attributes so trace backends can filter/group by violation type without parsing the redacted `guardrail_response` blob - Introduce `_emit_guardrail_spans_from_request_data` helper that constructs minimal kwargs from `request_data["metadata"]` and routes through `_create_guardrail_span`, sharing the same dedupe state to prevent double-emitting when both failure hooks fire - Extend `BedrockGuardrail` with `_build_tracing_detail` and `_extract_violation_category_names` which flatten BLOCKED assessments into human-readable category labels (topic names, content-filter types, PII entity types, named regex names) before redaction, and surface Bedrock's raw `action` field via `tracing_detail` - Security: violation category extraction deliberately omits `customWords.match` and unnamed regex `match` values because those fields carry the user-submitted content that triggered the rule; only operator-defined `name`/`type` labels are emitted - Add `violation_categories` and `guardrail_action` fields to `StandardLoggingGuardrailInformation` and `GuardrailTracingDetail` TypedDicts to carry the pre-redaction metadata through the logging pipeline - Add comprehensive test suite covering: guardrail span creation on failure, dedupe between `_handle_failure` and `async_post_call_failure_hook`, per-span status attributes for multi-guardrail sequences, Bedrock category extraction for all policy types, security leak prevention, and end-to-end `CustomGuardrail` violation path Co-authored-by: Yassin Kortam <yassinkortam@g.ucla.edu>
This commit is contained in:
Yassin Kortam
GitHub
parent
b55749248d
commit
10bd7406e0
5 changed files with 1009 additions and 0 deletions
|
|
@ -726,9 +726,57 @@ class OpenTelemetry(OTELGenAISemconvMixin, CustomLogger):
|
|||
exception_logging_span.set_status(Status(StatusCode.ERROR))
|
||||
exception_logging_span.end(end_time=self._to_ns(datetime.now()))
|
||||
|
||||
# Emit guardrail spans for any guardrail invocations that
|
||||
# ran during this request. _handle_failure typically does this,
|
||||
# but for pre-call guardrail blocks the standard_logging_object
|
||||
# may not carry guardrail_information by the time _handle_failure
|
||||
# fires (the data lives only in request_data["metadata"]). Pull
|
||||
# directly from request_data so the span is recorded either way;
|
||||
# _emit_once dedupes if _handle_failure already emitted it.
|
||||
self._emit_guardrail_spans_from_request_data(
|
||||
request_data=request_data,
|
||||
parent_span=parent_otel_span,
|
||||
)
|
||||
|
||||
# End Parent OTEL Sspan
|
||||
parent_otel_span.end(end_time=self._to_ns(datetime.now()))
|
||||
|
||||
def _emit_guardrail_spans_from_request_data(
|
||||
self,
|
||||
request_data: dict,
|
||||
parent_span: Optional[Any],
|
||||
) -> None:
|
||||
"""Emit ``guardrail`` spans from ``request_data["metadata"]
|
||||
["standard_logging_guardrail_information"]``.
|
||||
|
||||
Routed through ``_create_guardrail_span`` so the dedupe state in
|
||||
``_otel_internal`` is honoured — if ``_handle_failure`` already
|
||||
emitted these spans for the same kwargs, this is a no-op.
|
||||
"""
|
||||
from opentelemetry import trace as _trace
|
||||
|
||||
metadata = (request_data or {}).get("metadata") or {}
|
||||
guardrail_information = metadata.get("standard_logging_guardrail_information")
|
||||
if not guardrail_information:
|
||||
return
|
||||
|
||||
# _create_guardrail_span reads guardrail_information from
|
||||
# kwargs["standard_logging_object"] and shares its dedupe state via
|
||||
# kwargs["litellm_params"]["metadata"]["_otel_internal"]. Pass the
|
||||
# SAME metadata dict the proxy populated so _handle_failure and
|
||||
# this hook see the same dedupe markers.
|
||||
kwargs: Dict[str, Any] = {
|
||||
"litellm_params": {"metadata": metadata},
|
||||
"standard_logging_object": {
|
||||
"guardrail_information": guardrail_information,
|
||||
"metadata": metadata,
|
||||
},
|
||||
}
|
||||
context = (
|
||||
_trace.set_span_in_context(parent_span) if parent_span is not None else None
|
||||
)
|
||||
self._create_guardrail_span(kwargs=kwargs, context=context)
|
||||
|
||||
async def async_post_call_success_hook(
|
||||
self,
|
||||
data: dict,
|
||||
|
|
@ -1617,6 +1665,37 @@ class OpenTelemetry(OTELGenAISemconvMixin, CustomLogger):
|
|||
"guardrail_response", safe_dumps(guardrail_response)
|
||||
)
|
||||
|
||||
# Surface guardrail_status (success / guardrail_intervened /
|
||||
# guardrail_failed_to_respond / not_run) as a top-level span
|
||||
# attribute so trace backends can filter on it without parsing
|
||||
# guardrail_response.
|
||||
self.safe_set_attribute(
|
||||
span=guardrail_span,
|
||||
key="guardrail_status",
|
||||
value=guardrail_information.get("guardrail_status"),
|
||||
)
|
||||
|
||||
# Provider's raw top-level action (e.g. Bedrock's
|
||||
# ``GUARDRAIL_INTERVENED`` / ``NONE``). Populated by the provider
|
||||
# hook onto StandardLoggingGuardrailInformation so this integration
|
||||
# stays provider-agnostic — we only read a normalised string.
|
||||
guardrail_action = guardrail_information.get("guardrail_action")
|
||||
if guardrail_action:
|
||||
guardrail_span.set_attribute("guardrail_action", guardrail_action)
|
||||
|
||||
# The provider hook (e.g. Bedrock) extracts violation_categories
|
||||
# from the raw response BEFORE redaction and stamps them onto
|
||||
# StandardLoggingGuardrailInformation. Surfacing them here as a
|
||||
# queryable attribute lets dashboards group by violation category
|
||||
# without parsing the redacted guardrail_response blob.
|
||||
violation_categories = guardrail_information.get("violation_categories")
|
||||
if violation_categories:
|
||||
# OTel sequence attributes must be homogeneous primitives;
|
||||
# serialise to JSON once so set_attribute never coerces.
|
||||
guardrail_span.set_attribute(
|
||||
"guardrail_violation_categories", safe_dumps(violation_categories)
|
||||
)
|
||||
|
||||
self._set_team_attributes_from_kwargs(guardrail_span, kwargs)
|
||||
|
||||
guardrail_span.end(end_time=self._to_ns(end_time_datetime))
|
||||
|
|
|
|||
|
|
@ -63,6 +63,7 @@ from litellm.types.utils import (
|
|||
CallTypesLiteral,
|
||||
Choices,
|
||||
GuardrailStatus,
|
||||
GuardrailTracingDetail,
|
||||
Message,
|
||||
ModelResponse,
|
||||
ModelResponseStream,
|
||||
|
|
@ -509,6 +510,8 @@ class BedrockGuardrail(CustomGuardrail, BaseAWSLLM):
|
|||
# Add guardrail information to request trace
|
||||
#########################################################
|
||||
_json_response = httpx_response.json()
|
||||
tracing_detail = self._build_tracing_detail(_json_response)
|
||||
|
||||
# Raw Bedrock JSON is passed here; match/regex redaction runs once inside
|
||||
# CustomGuardrail.add_standard_logging_guardrail_information_to_request_data.
|
||||
self.add_standard_logging_guardrail_information_to_request_data(
|
||||
|
|
@ -522,6 +525,7 @@ class BedrockGuardrail(CustomGuardrail, BaseAWSLLM):
|
|||
end_time=datetime.now().timestamp(),
|
||||
duration=(datetime.now() - start_time).total_seconds(),
|
||||
event_type=event_type,
|
||||
tracing_detail=tracing_detail or None,
|
||||
)
|
||||
#########################################################
|
||||
if httpx_response.status_code == 200:
|
||||
|
|
@ -640,6 +644,55 @@ class BedrockGuardrail(CustomGuardrail, BaseAWSLLM):
|
|||
return (status_code, err)
|
||||
return (status_code, message)
|
||||
|
||||
def _build_tracing_detail(
|
||||
self, response: BedrockGuardrailResponse
|
||||
) -> GuardrailTracingDetail:
|
||||
"""
|
||||
Build the tracing detail from the raw Bedrock response, before
|
||||
redaction, so downstream loggers (OTEL, Langfuse, ...) get the
|
||||
actual category names rather than the "[REDACTED]" sentinel that
|
||||
replaces customWords.match later. Bedrock's top-level ``action``
|
||||
field ("GUARDRAIL_INTERVENED" or "NONE") is also surfaced so the
|
||||
OTEL integration can expose it as a queryable span attribute
|
||||
without re-parsing the redacted guardrail_response blob.
|
||||
"""
|
||||
tracing_detail: GuardrailTracingDetail = {}
|
||||
violation_categories = self._extract_violation_category_names(response)
|
||||
if violation_categories:
|
||||
tracing_detail["violation_categories"] = violation_categories
|
||||
bedrock_action = response.get("action")
|
||||
if isinstance(bedrock_action, str):
|
||||
tracing_detail["guardrail_action"] = bedrock_action
|
||||
return tracing_detail
|
||||
|
||||
def _extract_violation_category_names(
|
||||
self, response: BedrockGuardrailResponse
|
||||
) -> List[str]:
|
||||
"""
|
||||
Flatten the BLOCKED assessments into a list of human-readable category
|
||||
names suitable for queryable OTEL / standard-logging attributes.
|
||||
|
||||
SECURITY: only emits the non-sensitive policy *label* (topic name,
|
||||
content-filter type, PII entity type, named-regex name). The raw
|
||||
``match`` field is intentionally NOT used — it carries the user's
|
||||
original input that triggered the rule (e.g. a credit-card number
|
||||
that hit a regex, or the literal custom word). Surfacing it to
|
||||
telemetry would re-introduce the sensitive content the guardrail
|
||||
was supposed to keep out. Entries that only have a ``match`` (bare
|
||||
customWords, unnamed regexes) are therefore skipped — operators
|
||||
can still see the count in ``_extract_blocked_assessments`` which
|
||||
feeds the HTTP error detail.
|
||||
"""
|
||||
names: List[str] = []
|
||||
for block in self._extract_blocked_assessments(response):
|
||||
for match in block.get("matches", []) or []:
|
||||
# Allow-list non-sensitive labels only. Never fall back to
|
||||
# `match.get("match")` — that's user-submitted content.
|
||||
label = match.get("name") or match.get("type")
|
||||
if isinstance(label, str) and label:
|
||||
names.append(label)
|
||||
return names
|
||||
|
||||
def _extract_blocked_assessments(
|
||||
self, response: BedrockGuardrailResponse
|
||||
) -> List[dict]:
|
||||
|
|
|
|||
|
|
@ -2768,6 +2768,20 @@ class StandardLoggingGuardrailInformation(TypedDict, total=False):
|
|||
risk_score: Optional[float]
|
||||
"""Risk score 0-10 indicating how risky the request was (higher = riskier). Computed by the guardrail provider."""
|
||||
|
||||
violation_categories: Optional[List[str]]
|
||||
"""Names of the policy items that intervened on this request (e.g. Bedrock
|
||||
topic-policy topic names, content-policy filter types, PII entity types).
|
||||
Populated by the provider hook before redaction so downstream loggers
|
||||
(OTEL, Langfuse, ...) can filter by violation category without parsing
|
||||
the raw guardrail_response blob. Empty/absent when the guardrail allowed
|
||||
the request through."""
|
||||
|
||||
guardrail_action: Optional[str]
|
||||
"""Provider's raw top-level action string (e.g. Bedrock's ``GUARDRAIL_INTERVENED``
|
||||
or ``NONE``). Populated by the provider hook so the OTEL integration can
|
||||
surface it as a queryable span attribute without parsing the raw
|
||||
guardrail_response blob."""
|
||||
|
||||
|
||||
class EvalVerdict(TypedDict, total=False):
|
||||
criterion_name: str
|
||||
|
|
@ -2809,6 +2823,8 @@ class GuardrailTracingDetail(TypedDict, total=False):
|
|||
patterns_checked: Optional[int]
|
||||
alert_recipients: Optional[List[str]]
|
||||
risk_score: Optional[float]
|
||||
violation_categories: Optional[List[str]]
|
||||
guardrail_action: Optional[str]
|
||||
|
||||
|
||||
StandardLoggingPayloadStatus = Literal["success", "failure"]
|
||||
|
|
|
|||
|
|
@ -0,0 +1,641 @@
|
|||
"""
|
||||
Tests for guardrail OTEL spans on violation.
|
||||
|
||||
Two distinct gaps surface together when a pre-call guardrail blocks the
|
||||
request before it reaches the LLM provider:
|
||||
|
||||
1. ``async_post_call_failure_hook`` (the OTEL hook that actually runs on
|
||||
the proxy failure path) only stamps attributes on the proxy parent
|
||||
span. It never creates the child ``guardrail`` span, even though
|
||||
``request_data["metadata"]["standard_logging_guardrail_information"]``
|
||||
is populated by the time the hook runs.
|
||||
|
||||
2. ``_create_guardrail_span`` records ``guardrail_name`` / ``guardrail_mode``
|
||||
/ ``guardrail_response`` but does not surface ``guardrail_status``
|
||||
(success / guardrail_intervened / guardrail_failed_to_respond /
|
||||
not_run) or the violation categories (Bedrock topic policy names,
|
||||
content filter types, etc.) as queryable span attributes — the data
|
||||
is buried inside the serialised ``guardrail_response`` blob and cannot
|
||||
be filtered on in the trace backend.
|
||||
|
||||
The tests below use real OTEL SDK objects (TracerProvider +
|
||||
InMemorySpanExporter + a real BatchSpanProcessor-equivalent) and the
|
||||
real ``OpenTelemetry`` integration. No monkey patching of the integration
|
||||
under test — only the OTEL exporter is in-memory.
|
||||
"""
|
||||
|
||||
import os
|
||||
import sys
|
||||
import time
|
||||
import unittest
|
||||
from datetime import datetime, timedelta, timezone
|
||||
|
||||
from opentelemetry.sdk.trace import TracerProvider
|
||||
from opentelemetry.sdk.trace.export import SimpleSpanProcessor
|
||||
from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter
|
||||
from opentelemetry.trace import StatusCode
|
||||
|
||||
sys.path.insert(0, os.path.abspath("../.."))
|
||||
|
||||
from litellm.integrations.opentelemetry import (
|
||||
LITELLM_REQUEST_SPAN_NAME,
|
||||
OpenTelemetry,
|
||||
)
|
||||
from litellm.proxy._types import UserAPIKeyAuth
|
||||
|
||||
|
||||
GUARDRAIL_SPAN_NAME = "guardrail"
|
||||
PROXY_SPAN_NAME = "Received Proxy Server Request"
|
||||
|
||||
|
||||
def _bedrock_block_response():
|
||||
"""Realistic Bedrock ApplyGuardrail response when a topic policy fires.
|
||||
|
||||
Mirrors the shape in ``litellm/types/proxy/guardrails/guardrail_hooks/
|
||||
bedrock_guardrails.py`` so the violation-category extraction can be
|
||||
tested against the exact payload Bedrock returns.
|
||||
"""
|
||||
return {
|
||||
"action": "GUARDRAIL_INTERVENED",
|
||||
"assessments": [
|
||||
{
|
||||
"topicPolicy": {
|
||||
"topics": [
|
||||
{
|
||||
"name": "Fiduciary Advice",
|
||||
"type": "DENY",
|
||||
"action": "BLOCKED",
|
||||
}
|
||||
]
|
||||
},
|
||||
"contentPolicy": {
|
||||
"filters": [
|
||||
{
|
||||
"type": "VIOLENCE",
|
||||
"confidence": "HIGH",
|
||||
"action": "BLOCKED",
|
||||
}
|
||||
]
|
||||
},
|
||||
"wordPolicy": {
|
||||
"customWords": [{"match": "secret-codeword", "action": "BLOCKED"}],
|
||||
"managedWordLists": [
|
||||
{"match": "fuck", "type": "PROFANITY", "action": "BLOCKED"}
|
||||
],
|
||||
},
|
||||
}
|
||||
],
|
||||
"outputs": [{"text": "Sorry, the model cannot respond to this request."}],
|
||||
}
|
||||
|
||||
|
||||
def _slg_entry(
|
||||
guardrail_status,
|
||||
guardrail_response,
|
||||
*,
|
||||
name="bedrock-test",
|
||||
mode="pre_call",
|
||||
provider="bedrock",
|
||||
start=1.0,
|
||||
end=2.0,
|
||||
violation_categories=None,
|
||||
guardrail_action=None,
|
||||
):
|
||||
"""Build a StandardLoggingGuardrailInformation entry the way
|
||||
``add_standard_logging_guardrail_information_to_request_data`` does."""
|
||||
entry = {
|
||||
"guardrail_name": name,
|
||||
"guardrail_provider": provider,
|
||||
"guardrail_mode": mode,
|
||||
"guardrail_response": guardrail_response,
|
||||
"guardrail_status": guardrail_status,
|
||||
"start_time": start,
|
||||
"end_time": end,
|
||||
"duration": end - start,
|
||||
}
|
||||
if violation_categories is not None:
|
||||
entry["violation_categories"] = violation_categories
|
||||
if guardrail_action is not None:
|
||||
entry["guardrail_action"] = guardrail_action
|
||||
return entry
|
||||
|
||||
|
||||
def _kwargs_with_guardrail(
|
||||
*,
|
||||
entries,
|
||||
parent_span=None,
|
||||
include_exception=False,
|
||||
):
|
||||
"""Build the kwargs / model_call_details shape that the OTEL integration
|
||||
consumes. ``litellm_params.metadata`` is the SAME dict that the proxy's
|
||||
``request_data["metadata"]`` becomes after ``update_environment_variables``,
|
||||
so ``_otel_internal`` dedupe state lives there too."""
|
||||
metadata = {"standard_logging_guardrail_information": list(entries)}
|
||||
if parent_span is not None:
|
||||
metadata["litellm_parent_otel_span"] = parent_span
|
||||
kwargs = {
|
||||
"model": "gpt-4",
|
||||
"messages": [{"role": "user", "content": "Hello"}],
|
||||
"optional_params": {},
|
||||
"litellm_params": {
|
||||
"custom_llm_provider": "openai",
|
||||
"metadata": metadata,
|
||||
},
|
||||
"standard_logging_object": {
|
||||
"id": "test-call-id",
|
||||
"call_type": "completion",
|
||||
"metadata": metadata,
|
||||
"hidden_params": {},
|
||||
"guardrail_information": list(entries),
|
||||
},
|
||||
}
|
||||
if include_exception:
|
||||
kwargs["exception"] = Exception("guardrail blocked the request")
|
||||
return kwargs
|
||||
|
||||
|
||||
def _make_otel():
|
||||
"""Spin up a real OTEL pipeline backed by an in-memory exporter."""
|
||||
exporter = InMemorySpanExporter()
|
||||
provider = TracerProvider()
|
||||
provider.add_span_processor(SimpleSpanProcessor(exporter))
|
||||
otel = OpenTelemetry(tracer_provider=provider)
|
||||
otel.tracer = provider.get_tracer(__name__)
|
||||
return otel, provider, exporter
|
||||
|
||||
|
||||
def _run(coro):
|
||||
"""Run a coroutine on a fresh event loop and close it — prevents the
|
||||
"unclosed event loop" / ResourceWarning that you get from
|
||||
asyncio.new_event_loop().run_until_complete() with no cleanup."""
|
||||
import asyncio
|
||||
|
||||
loop = asyncio.new_event_loop()
|
||||
try:
|
||||
return loop.run_until_complete(coro)
|
||||
finally:
|
||||
loop.close()
|
||||
|
||||
|
||||
def _attr(span, key):
|
||||
return (span.attributes or {}).get(key)
|
||||
|
||||
|
||||
class TestGuardrailSpanOnViolation(unittest.TestCase):
|
||||
"""Bug 1: when a pre-call guardrail blocks, the guardrail span and the
|
||||
litellm_request span must both appear with the correct status."""
|
||||
|
||||
def test_handle_failure_creates_litellm_request_and_guardrail_spans(self):
|
||||
"""Driving ``_handle_failure`` with a populated
|
||||
``standard_logging_object['guardrail_information']`` entry must
|
||||
emit both spans, parented correctly, with ERROR on the parent."""
|
||||
otel, _, exporter = _make_otel()
|
||||
|
||||
kwargs = _kwargs_with_guardrail(
|
||||
entries=[
|
||||
_slg_entry("guardrail_intervened", _bedrock_block_response()),
|
||||
],
|
||||
include_exception=True,
|
||||
)
|
||||
|
||||
start = datetime.now(timezone.utc)
|
||||
end = start + timedelta(milliseconds=20)
|
||||
otel._handle_failure(kwargs, response_obj=None, start_time=start, end_time=end)
|
||||
|
||||
spans = exporter.get_finished_spans()
|
||||
litellm_spans = [s for s in spans if s.name == LITELLM_REQUEST_SPAN_NAME]
|
||||
guardrail_spans = [s for s in spans if s.name == GUARDRAIL_SPAN_NAME]
|
||||
|
||||
self.assertEqual(
|
||||
len(litellm_spans),
|
||||
1,
|
||||
"Expected exactly one litellm_request span on guardrail block",
|
||||
)
|
||||
self.assertEqual(litellm_spans[0].status.status_code, StatusCode.ERROR)
|
||||
|
||||
self.assertEqual(
|
||||
len(guardrail_spans),
|
||||
1,
|
||||
"Expected exactly one guardrail span on guardrail block",
|
||||
)
|
||||
|
||||
# Guardrail span must be a child of the litellm_request span
|
||||
self.assertIsNotNone(
|
||||
guardrail_spans[0].parent,
|
||||
"Guardrail span must be parented (not a root span)",
|
||||
)
|
||||
self.assertEqual(
|
||||
guardrail_spans[0].parent.span_id,
|
||||
litellm_spans[0].context.span_id,
|
||||
)
|
||||
|
||||
def test_async_post_call_failure_hook_emits_guardrail_span(self):
|
||||
"""The production failure path on the proxy calls
|
||||
``async_post_call_failure_hook`` with the (still-populated)
|
||||
``request_data``. The hook currently only stamps attrs on the proxy
|
||||
span; it must also emit the guardrail span so the violation is
|
||||
visible in the trace."""
|
||||
otel, provider, exporter = _make_otel()
|
||||
parent_span = provider.get_tracer(__name__).start_span(PROXY_SPAN_NAME)
|
||||
|
||||
user_api_key_dict = UserAPIKeyAuth(
|
||||
api_key="sk-test",
|
||||
parent_otel_span=parent_span,
|
||||
request_route="/chat/completions",
|
||||
)
|
||||
|
||||
request_data = {
|
||||
"model": "gpt-4",
|
||||
"messages": [{"role": "user", "content": "Hello"}],
|
||||
"metadata": {
|
||||
"standard_logging_guardrail_information": [
|
||||
_slg_entry("guardrail_intervened", _bedrock_block_response())
|
||||
],
|
||||
},
|
||||
}
|
||||
|
||||
_run(
|
||||
otel.async_post_call_failure_hook(
|
||||
request_data=request_data,
|
||||
original_exception=Exception("guardrail blocked"),
|
||||
user_api_key_dict=user_api_key_dict,
|
||||
)
|
||||
)
|
||||
|
||||
spans = exporter.get_finished_spans()
|
||||
guardrail_spans = [s for s in spans if s.name == GUARDRAIL_SPAN_NAME]
|
||||
self.assertEqual(
|
||||
len(guardrail_spans),
|
||||
1,
|
||||
"async_post_call_failure_hook must emit the guardrail span when "
|
||||
"request_data['metadata'] carries standard_logging_guardrail_information",
|
||||
)
|
||||
|
||||
# The guardrail span must be parented to the proxy request span so
|
||||
# backends correlate it with the rest of the trace.
|
||||
self.assertIsNotNone(guardrail_spans[0].parent)
|
||||
self.assertEqual(
|
||||
guardrail_spans[0].parent.span_id,
|
||||
parent_span.context.span_id,
|
||||
)
|
||||
|
||||
def test_handle_failure_and_post_call_failure_hook_dedupe(self):
|
||||
"""When _handle_failure and async_post_call_failure_hook BOTH fire
|
||||
for the same request (the production flow on a guardrail block),
|
||||
exactly one guardrail span must be emitted. The dedupe relies on
|
||||
request_data['metadata'] and kwargs['litellm_params']['metadata']
|
||||
referencing the SAME dict so _emit_once sees its earlier marker."""
|
||||
otel, provider, exporter = _make_otel()
|
||||
parent_span = provider.get_tracer(__name__).start_span(PROXY_SPAN_NAME)
|
||||
|
||||
# Shared metadata dict — same identity, mirroring how
|
||||
# update_environment_variables wires them in the proxy.
|
||||
shared_metadata = {
|
||||
"standard_logging_guardrail_information": [
|
||||
_slg_entry(
|
||||
"guardrail_intervened",
|
||||
_bedrock_block_response(),
|
||||
violation_categories=["Fiduciary Advice"],
|
||||
)
|
||||
],
|
||||
}
|
||||
|
||||
kwargs = {
|
||||
"model": "gpt-4",
|
||||
"messages": [{"role": "user", "content": "Hello"}],
|
||||
"optional_params": {},
|
||||
"litellm_params": {
|
||||
"custom_llm_provider": "openai",
|
||||
"metadata": shared_metadata,
|
||||
},
|
||||
"standard_logging_object": {
|
||||
"id": "test-call-id",
|
||||
"call_type": "completion",
|
||||
"metadata": shared_metadata,
|
||||
"hidden_params": {},
|
||||
"guardrail_information": shared_metadata[
|
||||
"standard_logging_guardrail_information"
|
||||
],
|
||||
},
|
||||
"exception": Exception("guardrail blocked"),
|
||||
}
|
||||
request_data = {
|
||||
"model": "gpt-4",
|
||||
"messages": [{"role": "user", "content": "Hello"}],
|
||||
"metadata": shared_metadata,
|
||||
}
|
||||
user_api_key_dict = UserAPIKeyAuth(
|
||||
api_key="sk-test",
|
||||
parent_otel_span=parent_span,
|
||||
request_route="/chat/completions",
|
||||
)
|
||||
|
||||
start = datetime.now(timezone.utc)
|
||||
end = start + timedelta(milliseconds=20)
|
||||
otel._handle_failure(kwargs, response_obj=None, start_time=start, end_time=end)
|
||||
_run(
|
||||
otel.async_post_call_failure_hook(
|
||||
request_data=request_data,
|
||||
original_exception=Exception("guardrail blocked"),
|
||||
user_api_key_dict=user_api_key_dict,
|
||||
)
|
||||
)
|
||||
|
||||
guardrail_spans = [
|
||||
s for s in exporter.get_finished_spans() if s.name == GUARDRAIL_SPAN_NAME
|
||||
]
|
||||
self.assertEqual(
|
||||
len(guardrail_spans),
|
||||
1,
|
||||
"Dedupe must collapse the two emit calls into one span when the "
|
||||
"metadata dict identity is shared between kwargs and request_data",
|
||||
)
|
||||
|
||||
|
||||
class TestGuardrailSpanAttributesOnViolation(unittest.TestCase):
|
||||
"""Bug 2: the guardrail span must surface the violation status and
|
||||
violation categories as queryable span attributes, not bury them inside
|
||||
``guardrail_response`` (which is logged as a single serialised blob)."""
|
||||
|
||||
def _emit_and_get_guardrail_span(self, entry):
|
||||
otel, _, exporter = _make_otel()
|
||||
kwargs = _kwargs_with_guardrail(entries=[entry])
|
||||
otel._create_guardrail_span(kwargs=kwargs, context=None)
|
||||
|
||||
guardrail_spans = [
|
||||
s for s in exporter.get_finished_spans() if s.name == GUARDRAIL_SPAN_NAME
|
||||
]
|
||||
self.assertEqual(len(guardrail_spans), 1)
|
||||
return guardrail_spans[0]
|
||||
|
||||
def test_status_attribute_present_for_intervened(self):
|
||||
entry = _slg_entry("guardrail_intervened", _bedrock_block_response())
|
||||
span = self._emit_and_get_guardrail_span(entry)
|
||||
self.assertEqual(
|
||||
_attr(span, "guardrail_status"),
|
||||
"guardrail_intervened",
|
||||
"guardrail_status must be exposed as a top-level span attribute",
|
||||
)
|
||||
|
||||
def test_status_attribute_present_for_success(self):
|
||||
entry = _slg_entry(
|
||||
"success",
|
||||
{"action": "NONE", "assessments": []},
|
||||
)
|
||||
span = self._emit_and_get_guardrail_span(entry)
|
||||
self.assertEqual(_attr(span, "guardrail_status"), "success")
|
||||
|
||||
def test_status_attribute_present_for_failed_to_respond(self):
|
||||
entry = _slg_entry(
|
||||
"guardrail_failed_to_respond",
|
||||
{"error": "endpoint unreachable"},
|
||||
)
|
||||
span = self._emit_and_get_guardrail_span(entry)
|
||||
self.assertEqual(_attr(span, "guardrail_status"), "guardrail_failed_to_respond")
|
||||
|
||||
def test_violation_categories_surfaced_when_provider_populates_them(self):
|
||||
"""The provider hook (e.g. Bedrock) extracts violation categories
|
||||
from the raw response BEFORE redaction and stamps them onto the
|
||||
StandardLoggingGuardrailInformation entry. OTEL must surface that
|
||||
list as a queryable span attribute so dashboards can group by
|
||||
violation type without parsing the redacted guardrail_response."""
|
||||
entry = _slg_entry(
|
||||
"guardrail_intervened",
|
||||
_bedrock_block_response(),
|
||||
violation_categories=["Fiduciary Advice", "VIOLENCE", "PROFANITY"],
|
||||
)
|
||||
span = self._emit_and_get_guardrail_span(entry)
|
||||
|
||||
categories = _attr(span, "guardrail_violation_categories")
|
||||
self.assertIsNotNone(
|
||||
categories,
|
||||
"guardrail_violation_categories must be set when the entry "
|
||||
"carries violation_categories",
|
||||
)
|
||||
# Serialised as JSON to keep set_attribute typing simple.
|
||||
as_str = categories if isinstance(categories, str) else repr(list(categories))
|
||||
self.assertIn("Fiduciary Advice", as_str)
|
||||
self.assertIn("VIOLENCE", as_str)
|
||||
self.assertIn("PROFANITY", as_str)
|
||||
|
||||
def test_no_violation_categories_when_field_absent(self):
|
||||
"""When the provider didn't populate violation_categories (success
|
||||
path, or provider didn't extract them), don't pollute the trace
|
||||
with an empty attribute."""
|
||||
entry = _slg_entry("success", {"action": "NONE", "assessments": []})
|
||||
span = self._emit_and_get_guardrail_span(entry)
|
||||
self.assertIsNone(_attr(span, "guardrail_violation_categories"))
|
||||
|
||||
def test_no_violation_categories_when_field_is_empty(self):
|
||||
"""Empty list must not produce a span attribute either."""
|
||||
entry = _slg_entry(
|
||||
"guardrail_intervened",
|
||||
_bedrock_block_response(),
|
||||
violation_categories=[],
|
||||
)
|
||||
span = self._emit_and_get_guardrail_span(entry)
|
||||
self.assertIsNone(_attr(span, "guardrail_violation_categories"))
|
||||
|
||||
def test_guardrail_action_surfaced_when_provider_populates_it(self):
|
||||
"""The provider hook (e.g. Bedrock) writes its raw top-level
|
||||
``action`` string onto StandardLoggingGuardrailInformation as
|
||||
``guardrail_action``. OTEL must expose it as a queryable span
|
||||
attribute so dashboards can pivot on the raw provider verdict
|
||||
(Bedrock ``GUARDRAIL_INTERVENED`` / ``NONE``) without parsing
|
||||
the redacted guardrail_response blob."""
|
||||
entry = _slg_entry(
|
||||
"guardrail_intervened",
|
||||
_bedrock_block_response(),
|
||||
guardrail_action="GUARDRAIL_INTERVENED",
|
||||
)
|
||||
span = self._emit_and_get_guardrail_span(entry)
|
||||
self.assertEqual(
|
||||
_attr(span, "guardrail_action"),
|
||||
"GUARDRAIL_INTERVENED",
|
||||
"guardrail_action must be exposed as a top-level span attribute",
|
||||
)
|
||||
|
||||
def test_guardrail_action_surfaced_for_allowed_request(self):
|
||||
"""Even on the success path, the provider's raw action (e.g.
|
||||
Bedrock ``NONE``) should be queryable so dashboards can group
|
||||
allowed-vs-blocked counts off the same attribute."""
|
||||
entry = _slg_entry(
|
||||
"success",
|
||||
{"action": "NONE", "assessments": []},
|
||||
guardrail_action="NONE",
|
||||
)
|
||||
span = self._emit_and_get_guardrail_span(entry)
|
||||
self.assertEqual(_attr(span, "guardrail_action"), "NONE")
|
||||
|
||||
def test_no_guardrail_action_when_field_absent(self):
|
||||
"""If the provider didn't populate the field (older payloads,
|
||||
non-Bedrock providers without a top-level action), don't emit
|
||||
an empty attribute."""
|
||||
entry = _slg_entry("success", {"action": "NONE", "assessments": []})
|
||||
span = self._emit_and_get_guardrail_span(entry)
|
||||
self.assertIsNone(_attr(span, "guardrail_action"))
|
||||
|
||||
|
||||
class TestMultipleGuardrailsOneBlocks(unittest.TestCase):
|
||||
"""When several guardrails run sequentially and only the last one
|
||||
intervenes, every guardrail span must appear with its own status —
|
||||
losing the early "allowed" spans would mask which checks ran."""
|
||||
|
||||
def test_all_guardrail_spans_emitted_with_per_entry_status(self):
|
||||
otel, _, exporter = _make_otel()
|
||||
|
||||
entries = [
|
||||
_slg_entry(
|
||||
"success",
|
||||
{"action": "NONE", "assessments": []},
|
||||
name="pii-mask",
|
||||
start=1.0,
|
||||
end=1.5,
|
||||
),
|
||||
_slg_entry(
|
||||
"success",
|
||||
{"action": "NONE", "assessments": []},
|
||||
name="prompt-injection",
|
||||
start=2.0,
|
||||
end=2.2,
|
||||
),
|
||||
_slg_entry(
|
||||
"guardrail_intervened",
|
||||
_bedrock_block_response(),
|
||||
name="bedrock-policy",
|
||||
start=3.0,
|
||||
end=3.4,
|
||||
),
|
||||
]
|
||||
kwargs = _kwargs_with_guardrail(
|
||||
entries=entries,
|
||||
include_exception=True,
|
||||
)
|
||||
|
||||
start = datetime.now(timezone.utc)
|
||||
end = start + timedelta(milliseconds=50)
|
||||
otel._handle_failure(kwargs, response_obj=None, start_time=start, end_time=end)
|
||||
|
||||
spans = exporter.get_finished_spans()
|
||||
guardrail_spans = sorted(
|
||||
(s for s in spans if s.name == GUARDRAIL_SPAN_NAME),
|
||||
key=lambda s: (s.attributes or {}).get("guardrail_name", ""),
|
||||
)
|
||||
self.assertEqual(
|
||||
len(guardrail_spans),
|
||||
3,
|
||||
"Every guardrail invocation must emit a span — even the ones "
|
||||
"that allowed the request through before the blocker fired",
|
||||
)
|
||||
|
||||
statuses = {
|
||||
_attr(s, "guardrail_name"): _attr(s, "guardrail_status")
|
||||
for s in guardrail_spans
|
||||
}
|
||||
self.assertEqual(statuses["pii-mask"], "success")
|
||||
self.assertEqual(statuses["prompt-injection"], "success")
|
||||
self.assertEqual(statuses["bedrock-policy"], "guardrail_intervened")
|
||||
|
||||
|
||||
class TestCustomGuardrailEndToEnd(unittest.TestCase):
|
||||
"""End-to-end: a real ``CustomGuardrail`` subclass calls
|
||||
``add_standard_logging_guardrail_information_to_request_data`` and then
|
||||
raises. We then drive ``_handle_failure`` with the resulting kwargs
|
||||
(matching the shape ``async_failure_handler`` would build) and verify
|
||||
the guardrail span carries the recorded information."""
|
||||
|
||||
def test_real_custom_guardrail_violation_path(self):
|
||||
# Deliberately not importing fastapi here — the real Bedrock guardrail
|
||||
# raises HTTPException, but the OTEL span flow is exception-type
|
||||
# agnostic. Using a plain Exception keeps this test runnable in
|
||||
# SDK-only installs that don't ship fastapi.
|
||||
from litellm.integrations.custom_guardrail import CustomGuardrail
|
||||
from litellm.types.guardrails import GuardrailEventHooks
|
||||
|
||||
class BlockingViolation(Exception):
|
||||
pass
|
||||
|
||||
class BlockingGuardrail(CustomGuardrail):
|
||||
async def async_pre_call_hook(
|
||||
self,
|
||||
user_api_key_dict,
|
||||
cache,
|
||||
data,
|
||||
call_type,
|
||||
):
|
||||
start_ts = time.time()
|
||||
self.add_standard_logging_guardrail_information_to_request_data(
|
||||
guardrail_provider="bedrock",
|
||||
guardrail_json_response=_bedrock_block_response(),
|
||||
request_data=data,
|
||||
guardrail_status="guardrail_intervened",
|
||||
start_time=start_ts,
|
||||
end_time=start_ts + 0.01,
|
||||
duration=0.01,
|
||||
event_type=GuardrailEventHooks.pre_call,
|
||||
tracing_detail={
|
||||
"violation_categories": ["Fiduciary Advice", "VIOLENCE"]
|
||||
},
|
||||
)
|
||||
raise BlockingViolation("violation")
|
||||
|
||||
request_data = {
|
||||
"model": "gpt-4",
|
||||
"messages": [{"role": "user", "content": "Hi"}],
|
||||
"metadata": {},
|
||||
}
|
||||
guardrail = BlockingGuardrail(
|
||||
guardrail_name="blocking-test",
|
||||
event_hook=GuardrailEventHooks.pre_call,
|
||||
)
|
||||
|
||||
with self.assertRaises(BlockingViolation):
|
||||
_run(
|
||||
guardrail.async_pre_call_hook(
|
||||
user_api_key_dict=UserAPIKeyAuth(api_key="sk-test"),
|
||||
cache=None,
|
||||
data=request_data,
|
||||
call_type="completion",
|
||||
)
|
||||
)
|
||||
|
||||
slg_info = request_data["metadata"].get(
|
||||
"standard_logging_guardrail_information"
|
||||
)
|
||||
self.assertTrue(
|
||||
slg_info,
|
||||
"Guardrail must have recorded its information to request_data "
|
||||
"BEFORE raising — otherwise the OTEL hook sees nothing",
|
||||
)
|
||||
|
||||
# Now simulate the OTEL failure handler picking up this metadata
|
||||
otel, _, exporter = _make_otel()
|
||||
kwargs = _kwargs_with_guardrail(
|
||||
entries=slg_info,
|
||||
include_exception=True,
|
||||
)
|
||||
start = datetime.now(timezone.utc)
|
||||
end = start + timedelta(milliseconds=15)
|
||||
otel._handle_failure(kwargs, response_obj=None, start_time=start, end_time=end)
|
||||
|
||||
spans = exporter.get_finished_spans()
|
||||
guardrail_spans = [s for s in spans if s.name == GUARDRAIL_SPAN_NAME]
|
||||
self.assertEqual(len(guardrail_spans), 1)
|
||||
self.assertEqual(
|
||||
_attr(guardrail_spans[0], "guardrail_status"),
|
||||
"guardrail_intervened",
|
||||
)
|
||||
self.assertEqual(
|
||||
_attr(guardrail_spans[0], "guardrail_name"),
|
||||
"blocking-test",
|
||||
)
|
||||
# End-to-end: the violation_categories the guardrail passed through
|
||||
# tracing_detail must arrive as a queryable span attribute.
|
||||
categories = _attr(guardrail_spans[0], "guardrail_violation_categories")
|
||||
self.assertIsNotNone(categories)
|
||||
self.assertIn("Fiduciary Advice", str(categories))
|
||||
self.assertIn("VIOLENCE", str(categories))
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
|
|
@ -2073,6 +2073,226 @@ def test_get_http_exception_includes_assessments_and_identifier():
|
|||
assert exc.detail["assessments"][0]["matches"][0]["match"] == "[REDACTED]"
|
||||
|
||||
|
||||
def test_extract_violation_category_names_mixed_policies():
|
||||
"""Topic names, content-filter types, PII types, and managed-word types
|
||||
flatten into a single category-name list — using only the operator-
|
||||
defined `name`/`type` labels."""
|
||||
g = _make_guardrail()
|
||||
response = {
|
||||
"action": "GUARDRAIL_INTERVENED",
|
||||
"assessments": [
|
||||
{
|
||||
"topicPolicy": {
|
||||
"topics": [
|
||||
{"name": "Fiduciary Advice", "action": "BLOCKED"},
|
||||
{"name": "Tax Advice", "action": "BLOCKED"},
|
||||
]
|
||||
},
|
||||
"contentPolicy": {
|
||||
"filters": [{"type": "VIOLENCE", "action": "BLOCKED"}]
|
||||
},
|
||||
"wordPolicy": {
|
||||
"managedWordLists": [{"type": "PROFANITY", "action": "BLOCKED"}],
|
||||
},
|
||||
"sensitiveInformationPolicy": {
|
||||
"piiEntities": [{"type": "EMAIL", "action": "BLOCKED"}]
|
||||
},
|
||||
}
|
||||
],
|
||||
}
|
||||
names = g._extract_violation_category_names(response)
|
||||
assert "Fiduciary Advice" in names
|
||||
assert "Tax Advice" in names
|
||||
assert "VIOLENCE" in names
|
||||
assert "PROFANITY" in names
|
||||
assert "EMAIL" in names
|
||||
|
||||
|
||||
def test_extract_violation_category_names_does_not_leak_user_input():
|
||||
"""SECURITY: customWords.match is the raw user-submitted word that
|
||||
triggered the rule, and an unnamed regex match is the actual sensitive
|
||||
value (e.g. a credit-card number). Neither must appear in
|
||||
violation_categories — otherwise the content the guardrail blocked
|
||||
leaks straight into telemetry backends."""
|
||||
g = _make_guardrail()
|
||||
response = {
|
||||
"action": "GUARDRAIL_INTERVENED",
|
||||
"assessments": [
|
||||
{
|
||||
"wordPolicy": {
|
||||
"customWords": [
|
||||
{"match": "secret-codeword-abc-123", "action": "BLOCKED"}
|
||||
],
|
||||
},
|
||||
"sensitiveInformationPolicy": {
|
||||
"regexes": [{"match": "4111-1111-1111-1111", "action": "BLOCKED"}]
|
||||
},
|
||||
}
|
||||
],
|
||||
}
|
||||
names = g._extract_violation_category_names(response)
|
||||
assert "secret-codeword-abc-123" not in names
|
||||
assert "4111-1111-1111-1111" not in names
|
||||
assert names == []
|
||||
|
||||
|
||||
def test_extract_violation_category_names_named_regex_uses_name():
|
||||
"""A regex with a `name` field surfaces that operator-defined label
|
||||
(safe to log), not the matched value."""
|
||||
g = _make_guardrail()
|
||||
response = {
|
||||
"action": "GUARDRAIL_INTERVENED",
|
||||
"assessments": [
|
||||
{
|
||||
"sensitiveInformationPolicy": {
|
||||
"regexes": [
|
||||
{
|
||||
"name": "credit-card-pattern",
|
||||
"match": "4111-1111-1111-1111",
|
||||
"action": "BLOCKED",
|
||||
}
|
||||
]
|
||||
}
|
||||
}
|
||||
],
|
||||
}
|
||||
names = g._extract_violation_category_names(response)
|
||||
assert names == ["credit-card-pattern"]
|
||||
|
||||
|
||||
def test_extract_violation_category_names_skips_anonymized():
|
||||
"""ANONYMIZED entries are not blocks — they must not contribute to the
|
||||
violation_categories list."""
|
||||
g = _make_guardrail()
|
||||
response = {
|
||||
"action": "GUARDRAIL_INTERVENED",
|
||||
"assessments": [
|
||||
{
|
||||
"sensitiveInformationPolicy": {
|
||||
"piiEntities": [{"type": "NAME", "action": "ANONYMIZED"}]
|
||||
}
|
||||
}
|
||||
],
|
||||
}
|
||||
assert g._extract_violation_category_names(response) == []
|
||||
|
||||
|
||||
def test_extract_violation_category_names_no_assessments():
|
||||
"""Empty / missing assessments → empty list, not an error."""
|
||||
g = _make_guardrail()
|
||||
assert g._extract_violation_category_names({"action": "NONE"}) == []
|
||||
assert g._extract_violation_category_names({"assessments": None}) == []
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_make_bedrock_api_request_forwards_guardrail_action():
|
||||
"""Bedrock's top-level ``action`` string must be propagated through
|
||||
``tracing_detail`` so downstream loggers (OTEL, ...) can surface the
|
||||
raw provider verdict as a queryable attribute without re-parsing the
|
||||
redacted guardrail_response blob."""
|
||||
guardrail = BedrockGuardrail(
|
||||
guardrailIdentifier="test-guardrail", guardrailVersion="DRAFT"
|
||||
)
|
||||
mock_credentials = MagicMock()
|
||||
mock_credentials.access_key = "k"
|
||||
mock_credentials.secret_key = "s"
|
||||
mock_credentials.token = None
|
||||
|
||||
mock_bedrock_response = MagicMock()
|
||||
mock_bedrock_response.status_code = 200
|
||||
mock_bedrock_response.json.return_value = {
|
||||
"action": "GUARDRAIL_INTERVENED",
|
||||
"assessments": [
|
||||
{
|
||||
"topicPolicy": {
|
||||
"topics": [{"name": "Fiduciary Advice", "action": "BLOCKED"}]
|
||||
}
|
||||
}
|
||||
],
|
||||
}
|
||||
|
||||
request_data = {
|
||||
"model": "gpt-4o",
|
||||
"messages": [{"role": "user", "content": "hi"}],
|
||||
}
|
||||
|
||||
with (
|
||||
patch.object(
|
||||
guardrail.async_handler, "post", new_callable=AsyncMock
|
||||
) as mock_post,
|
||||
patch.object(
|
||||
guardrail, "_load_credentials", return_value=(mock_credentials, "us-east-1")
|
||||
),
|
||||
patch.object(guardrail, "_prepare_request", return_value=MagicMock()),
|
||||
patch.object(
|
||||
guardrail,
|
||||
"add_standard_logging_guardrail_information_to_request_data",
|
||||
) as mock_log,
|
||||
patch.object(
|
||||
guardrail,
|
||||
"_get_http_exception_for_blocked_guardrail",
|
||||
return_value=Exception("blocked"),
|
||||
),
|
||||
):
|
||||
mock_post.return_value = mock_bedrock_response
|
||||
|
||||
with pytest.raises(Exception):
|
||||
await guardrail.make_bedrock_api_request(
|
||||
source="INPUT",
|
||||
messages=request_data["messages"],
|
||||
request_data=request_data,
|
||||
)
|
||||
|
||||
tracing_detail = mock_log.call_args.kwargs["tracing_detail"]
|
||||
assert tracing_detail is not None
|
||||
assert tracing_detail["guardrail_action"] == "GUARDRAIL_INTERVENED"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_make_bedrock_api_request_omits_guardrail_action_when_missing():
|
||||
"""If the Bedrock response omits ``action`` (older / partial payloads),
|
||||
the field must be left off ``tracing_detail`` rather than written as
|
||||
``None`` — downstream code expects strings or absence, not nulls."""
|
||||
guardrail = BedrockGuardrail(
|
||||
guardrailIdentifier="test-guardrail", guardrailVersion="DRAFT"
|
||||
)
|
||||
mock_credentials = MagicMock()
|
||||
mock_credentials.access_key = "k"
|
||||
mock_credentials.secret_key = "s"
|
||||
mock_credentials.token = None
|
||||
|
||||
mock_bedrock_response = MagicMock()
|
||||
mock_bedrock_response.status_code = 200
|
||||
mock_bedrock_response.json.return_value = {"assessments": []}
|
||||
|
||||
with (
|
||||
patch.object(
|
||||
guardrail.async_handler, "post", new_callable=AsyncMock
|
||||
) as mock_post,
|
||||
patch.object(
|
||||
guardrail, "_load_credentials", return_value=(mock_credentials, "us-east-1")
|
||||
),
|
||||
patch.object(guardrail, "_prepare_request", return_value=MagicMock()),
|
||||
patch.object(
|
||||
guardrail,
|
||||
"add_standard_logging_guardrail_information_to_request_data",
|
||||
) as mock_log,
|
||||
):
|
||||
mock_post.return_value = mock_bedrock_response
|
||||
|
||||
await guardrail.make_bedrock_api_request(
|
||||
source="INPUT",
|
||||
messages=[{"role": "user", "content": "hi"}],
|
||||
request_data={"model": "gpt-4o", "messages": []},
|
||||
)
|
||||
|
||||
tracing_detail = mock_log.call_args.kwargs["tracing_detail"]
|
||||
# No violation categories and no action ⇒ tracing_detail stays None
|
||||
# (the hook collapses an empty dict before forwarding).
|
||||
if tracing_detail is not None:
|
||||
assert "guardrail_action" not in tracing_detail
|
||||
|
||||
|
||||
def test_get_http_exception_no_blocked_assessments_omits_field():
|
||||
"""L3: when no assessments are blocked, the `assessments` key is omitted entirely."""
|
||||
g = _make_guardrail()
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue