mirror of
https://github.com/BerriAI/litellm.git
synced 2026-09-17 23:51:30 +00:00
fix(responses): route mid-stream error events through exception_type so content_policy_fallbacks fire
Mid-stream error events on the streaming Responses API were all raised as APIError, so a content_policy_violation event never matched the router's content-policy fallback dispatch and the client got the raw error instead of the fallback model's answer. Map each error event's code and status through the existing exception_type mapping, matching the non-streaming path, and unwrap the typed ContentPolicyViolationError and ContextWindowExceededError so the router routes them to the configured content_policy_fallbacks and context_window_fallbacks.
This commit is contained in:
mateo-berri
parent
30f33a949b
commit
073d4fe2b0
4 changed files with 288 additions and 32 deletions
|
|
@ -31,6 +31,7 @@ from litellm.litellm_core_utils.llm_response_utils.response_metadata import (
|
|||
update_response_metadata,
|
||||
)
|
||||
from litellm.litellm_core_utils.thread_pool_executor import executor
|
||||
from litellm.llms.base_llm.chat.transformation import BaseLLMException
|
||||
from litellm.llms.base_llm.responses.transformation import BaseResponsesAPIConfig
|
||||
from litellm.responses.utils import ResponseAPILoggingUtils, ResponsesAPIRequestUtils
|
||||
from litellm.types.llms.openai import (
|
||||
|
|
@ -221,6 +222,13 @@ def _status_code_for_error_fields(error_type: str | None, error_code: str | None
|
|||
)
|
||||
|
||||
|
||||
def _mid_stream_fallback_eligible(mapped_exception: Exception) -> bool:
|
||||
if isinstance(mapped_exception, (litellm.ContentPolicyViolationError, litellm.ContextWindowExceededError)):
|
||||
return True
|
||||
status_code: Final = getattr(mapped_exception, "status_code", None)
|
||||
return not isinstance(status_code, int) or status_code >= 500 or status_code == 429
|
||||
|
||||
|
||||
class BaseResponsesAPIStreamingIterator:
|
||||
"""
|
||||
Base class for streaming iterators that process responses from the Responses API.
|
||||
|
|
@ -521,15 +529,8 @@ class BaseResponsesAPIStreamingIterator:
|
|||
getattr(self.completed_response, "response", None) if self.completed_response else None
|
||||
)
|
||||
error_info: Final = getattr(response_obj, "error", None) if response_obj else None
|
||||
error_message, error_type, error_code = _error_event_fields(error_info)
|
||||
self._record_failed_response_usage(response_obj)
|
||||
exception: Final = litellm.APIError(
|
||||
status_code=_status_code_for_error_fields(error_type, error_code),
|
||||
message=error_message,
|
||||
llm_provider=self.custom_llm_provider or "",
|
||||
model=self.model or "",
|
||||
)
|
||||
self._handle_failure(exception)
|
||||
self._handle_failure(self._map_error_event_exception(error_info))
|
||||
|
||||
def _record_failed_response_usage(self, response_obj: ResponsesAPIResponse | None) -> None:
|
||||
if response_obj is None or self.logging_obj is None:
|
||||
|
|
@ -551,6 +552,26 @@ class BaseResponsesAPIStreamingIterator:
|
|||
self.logging_obj._response_cost_calculator(result=response_obj) or 0.0
|
||||
)
|
||||
|
||||
def _map_error_event_exception(self, error_obj: object) -> Exception:
|
||||
error_message, error_type, error_code = _error_event_fields(error_obj)
|
||||
status_code: Final = _status_code_for_error_fields(error_type, error_code)
|
||||
error_body: Final = {"message": error_message, "type": error_type, "code": error_code}
|
||||
provider_exception: Final = BaseLLMException(
|
||||
status_code=status_code,
|
||||
message=f"Error code: {status_code} - {{'error': {error_body}}}",
|
||||
body=error_body,
|
||||
)
|
||||
try:
|
||||
return litellm.exception_type(
|
||||
model=self.model or "",
|
||||
custom_llm_provider=self.custom_llm_provider or "",
|
||||
original_exception=provider_exception,
|
||||
completion_kwargs={},
|
||||
extra_kwargs={},
|
||||
)
|
||||
except Exception as mapped_exception:
|
||||
return mapped_exception
|
||||
|
||||
def _maybe_raise_for_error_event(self, result: object) -> None:
|
||||
chunk_type: Final = getattr(result, "type", None)
|
||||
if chunk_type not in ("error", "response.failed"):
|
||||
|
|
@ -562,15 +583,8 @@ class BaseResponsesAPIStreamingIterator:
|
|||
else getattr(result, "error", None)
|
||||
)
|
||||
|
||||
error_message, error_type, error_code = _error_event_fields(error_obj)
|
||||
status_code: Final = _status_code_for_error_fields(error_type, error_code)
|
||||
mapped_exception: Final = litellm.APIError(
|
||||
status_code=status_code,
|
||||
message=error_message,
|
||||
llm_provider=self.custom_llm_provider or "",
|
||||
model=self.model or "",
|
||||
)
|
||||
if 400 <= status_code < 500 and status_code != 429:
|
||||
mapped_exception: Final = self._map_error_event_exception(error_obj)
|
||||
if not _mid_stream_fallback_eligible(mapped_exception):
|
||||
raise mapped_exception
|
||||
raise MidStreamFallbackError(
|
||||
message=str(mapped_exception),
|
||||
|
|
|
|||
|
|
@ -3271,8 +3271,16 @@ class Router:
|
|||
kwargs=initial_kwargs,
|
||||
metadata_variable_name="litellm_metadata",
|
||||
)
|
||||
fallback_trigger: Final[Exception] = (
|
||||
e.original_exception
|
||||
if isinstance(
|
||||
e.original_exception,
|
||||
(litellm.ContentPolicyViolationError, litellm.ContextWindowExceededError),
|
||||
)
|
||||
else e
|
||||
)
|
||||
fallback_response = await self.async_function_with_fallbacks_common_utils(
|
||||
e=e,
|
||||
e=fallback_trigger,
|
||||
disable_fallbacks=False,
|
||||
fallbacks=fallbacks,
|
||||
context_window_fallbacks=context_window_fallbacks,
|
||||
|
|
|
|||
|
|
@ -1,9 +1,14 @@
|
|||
"""
|
||||
Regression: in-stream error events (type="error", type="response.failed") must
|
||||
raise instead of being returned as benign chunks, mirroring chat streaming
|
||||
semantics (_handle_stream_fallback_error): non-retriable 4xx (except 429)
|
||||
raise litellm.APIError directly; 429 and 5xx are wrapped in
|
||||
MidStreamFallbackError so the Router's mid-stream fallback machinery fires.
|
||||
semantics (_handle_stream_fallback_error). The event's code, type and status go
|
||||
through litellm.exception_type, so each event raises the same typed exception
|
||||
the non-streaming path raises for that provider error: non-retriable 4xx
|
||||
(except 429) raise that typed exception directly, while 429, 5xx,
|
||||
ContentPolicyViolationError and ContextWindowExceededError are wrapped in
|
||||
MidStreamFallbackError so the Router's mid-stream fallback machinery fires and
|
||||
its content_policy_fallbacks / context_window_fallbacks dispatch sees the
|
||||
trigger it matches on.
|
||||
|
||||
Status mapping must consider both the OpenAI error `type` (e.g.
|
||||
"invalid_request_error") and `code` (e.g. "invalid_prompt",
|
||||
|
|
@ -66,12 +71,12 @@ def test_maybe_raise_for_error_event_wraps_unknown_error_in_mid_stream_fallback(
|
|||
with pytest.raises(MidStreamFallbackError) as exc_info:
|
||||
iterator._maybe_raise_for_error_event(chunk)
|
||||
assert exc_info.value.status_code == 500
|
||||
assert isinstance(exc_info.value.original_exception, litellm.APIError)
|
||||
assert isinstance(exc_info.value.original_exception, litellm.InternalServerError)
|
||||
assert exc_info.value.original_exception.status_code == 500
|
||||
|
||||
|
||||
def test_maybe_raise_for_error_event_maps_rate_limit_code_to_429_mid_stream_fallback():
|
||||
"""429 is retriable: it must be wrapped so the Router can fall back, carrying the mapped APIError."""
|
||||
"""429 is retriable: it must be wrapped so the Router can fall back, carrying the mapped RateLimitError."""
|
||||
iterator = _make_iterator()
|
||||
chunk = _make_error_chunk("tokens", "rate_limit_exceeded", "Too many requests")
|
||||
with pytest.raises(MidStreamFallbackError) as exc_info:
|
||||
|
|
@ -79,15 +84,15 @@ def test_maybe_raise_for_error_event_maps_rate_limit_code_to_429_mid_stream_fall
|
|||
assert exc_info.value.status_code == 429
|
||||
assert exc_info.value.generated_content == ""
|
||||
assert exc_info.value.is_pre_first_chunk is True
|
||||
assert isinstance(exc_info.value.original_exception, litellm.APIError)
|
||||
assert isinstance(exc_info.value.original_exception, litellm.RateLimitError)
|
||||
assert exc_info.value.original_exception.status_code == 429
|
||||
|
||||
|
||||
def test_maybe_raise_for_error_event_maps_invalid_request_type_to_400():
|
||||
"""Client errors classified via the `type` field must raise APIError directly (no fallback)."""
|
||||
"""Client errors classified via the `type` field must raise BadRequestError directly (no fallback)."""
|
||||
iterator = _make_iterator()
|
||||
chunk = _make_error_chunk("invalid_request_error", "invalid_prompt", "bad request")
|
||||
with pytest.raises(litellm.APIError) as exc_info:
|
||||
with pytest.raises(litellm.BadRequestError) as exc_info:
|
||||
iterator._maybe_raise_for_error_event(chunk)
|
||||
assert exc_info.value.status_code == 400
|
||||
assert not isinstance(exc_info.value, MidStreamFallbackError)
|
||||
|
|
@ -99,12 +104,84 @@ def test_maybe_raise_for_error_event_maps_context_length_code_to_400():
|
|||
chunk = Mock()
|
||||
chunk.type = "error"
|
||||
chunk.error = {"code": "context_length_exceeded", "message": "too long"}
|
||||
with pytest.raises(litellm.APIError) as exc_info:
|
||||
with pytest.raises(litellm.BadRequestError) as exc_info:
|
||||
iterator._maybe_raise_for_error_event(chunk)
|
||||
assert exc_info.value.status_code == 400
|
||||
assert not isinstance(exc_info.value, MidStreamFallbackError)
|
||||
|
||||
|
||||
def test_maybe_raise_for_error_event_wraps_context_window_exceeded_for_context_window_fallbacks():
|
||||
"""A context-length error event maps to ContextWindowExceededError exactly like the non-streaming
|
||||
path and is wrapped so the Router's context_window_fallbacks dispatch fires mid-stream."""
|
||||
iterator = _make_iterator()
|
||||
chunk = _make_error_chunk(
|
||||
"invalid_request_error",
|
||||
"context_length_exceeded",
|
||||
"This model's maximum context length is 128000 tokens. However, your messages resulted in 130000 tokens.",
|
||||
)
|
||||
with pytest.raises(MidStreamFallbackError) as exc_info:
|
||||
iterator._maybe_raise_for_error_event(chunk)
|
||||
assert isinstance(exc_info.value.original_exception, litellm.ContextWindowExceededError)
|
||||
assert exc_info.value.status_code == 400
|
||||
|
||||
|
||||
CONTENT_POLICY_MESSAGE = "This content was flagged for possible cybersecurity risk. The response was halted mid-stream."
|
||||
|
||||
|
||||
@pytest.mark.parametrize("custom_llm_provider", ["openai", "azure"])
|
||||
def test_maybe_raise_for_error_event_wraps_content_policy_violation_for_content_policy_fallbacks(
|
||||
custom_llm_provider: str,
|
||||
):
|
||||
"""Regression: a content_policy_violation error event used to raise a bare APIError, so the Router's
|
||||
content_policy_fallbacks never fired. It must map to ContentPolicyViolationError (the same exception the
|
||||
non-streaming path raises) and be wrapped so the Router's mid-stream fallback catches it."""
|
||||
iterator = _make_iterator()
|
||||
iterator.custom_llm_provider = custom_llm_provider
|
||||
chunk = _make_error_chunk("invalid_request_error", "content_policy_violation", CONTENT_POLICY_MESSAGE)
|
||||
with pytest.raises(MidStreamFallbackError) as exc_info:
|
||||
iterator._maybe_raise_for_error_event(chunk)
|
||||
assert isinstance(exc_info.value.original_exception, litellm.ContentPolicyViolationError)
|
||||
assert exc_info.value.original_exception.status_code == 400
|
||||
assert exc_info.value.status_code == 400
|
||||
assert exc_info.value.is_pre_first_chunk is True
|
||||
assert CONTENT_POLICY_MESSAGE in str(exc_info.value.original_exception)
|
||||
|
||||
|
||||
def test_maybe_raise_for_response_failed_event_wraps_content_policy_violation():
|
||||
iterator = _make_iterator()
|
||||
chunk = _make_failed_chunk(
|
||||
{"type": "invalid_request_error", "code": "content_policy_violation", "message": CONTENT_POLICY_MESSAGE}
|
||||
)
|
||||
with pytest.raises(MidStreamFallbackError) as exc_info:
|
||||
iterator._maybe_raise_for_error_event(chunk)
|
||||
assert isinstance(exc_info.value.original_exception, litellm.ContentPolicyViolationError)
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"error_type,error_code,expected_exception",
|
||||
[
|
||||
("invalid_request_error", "content_policy_violation", litellm.ContentPolicyViolationError),
|
||||
("tokens", "rate_limit_exceeded", litellm.RateLimitError),
|
||||
("invalid_request_error", "insufficient_quota", litellm.RateLimitError),
|
||||
("server_error", "internal_error", litellm.InternalServerError),
|
||||
("invalid_request_error", "invalid_prompt", litellm.BadRequestError),
|
||||
("invalid_request_error", "model_not_found", litellm.NotFoundError),
|
||||
("server_error", "vector_store_timeout", litellm.Timeout),
|
||||
],
|
||||
)
|
||||
def test_error_event_raises_the_same_typed_exception_as_the_non_streaming_path(
|
||||
error_type: str, error_code: str, expected_exception: type[Exception]
|
||||
):
|
||||
iterator = _make_iterator()
|
||||
chunk = _make_error_chunk(error_type, error_code, "provider message")
|
||||
with pytest.raises((MidStreamFallbackError, expected_exception)) as exc_info:
|
||||
iterator._maybe_raise_for_error_event(chunk)
|
||||
raised = exc_info.value
|
||||
typed_exception = raised.original_exception if isinstance(raised, MidStreamFallbackError) else raised
|
||||
assert type(typed_exception) is expected_exception
|
||||
assert "provider message" in str(typed_exception)
|
||||
|
||||
|
||||
def test_maybe_raise_for_error_event_maps_insufficient_quota_to_429():
|
||||
"""OpenAI returns HTTP 429 for insufficient_quota; it must not map to 400 even though its type
|
||||
is invalid_request_error-adjacent, and it must be wrapped for fallback."""
|
||||
|
|
@ -113,6 +190,7 @@ def test_maybe_raise_for_error_event_maps_insufficient_quota_to_429():
|
|||
with pytest.raises(MidStreamFallbackError) as exc_info:
|
||||
iterator._maybe_raise_for_error_event(chunk)
|
||||
assert exc_info.value.status_code == 429
|
||||
assert isinstance(exc_info.value.original_exception, litellm.RateLimitError)
|
||||
|
||||
|
||||
def test_maybe_raise_for_error_event_passes_through_normal_chunk():
|
||||
|
|
@ -186,10 +264,43 @@ async def test_async_iterator_raises_mid_stream_fallback_on_rate_limit_error_eve
|
|||
assert exc_info.value.status_code == 429
|
||||
assert exc_info.value.is_pre_first_chunk is True
|
||||
assert exc_info.value.generated_content == ""
|
||||
assert isinstance(exc_info.value.original_exception, litellm.APIError)
|
||||
assert isinstance(exc_info.value.original_exception, litellm.RateLimitError)
|
||||
assert exc_info.value.original_exception.status_code == 429
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_async_iterator_content_policy_violation_after_first_chunk_carries_generated_content():
|
||||
"""The customer's case: text streams, then the provider halts the stream with a
|
||||
content_policy_violation error event. The iterator must surface ContentPolicyViolationError
|
||||
inside MidStreamFallbackError, together with the text already streamed."""
|
||||
iterator = _make_async_iterator_with_events(
|
||||
[
|
||||
{"type": "response.output_text.delta", "delta": "partial "},
|
||||
{
|
||||
"type": "error",
|
||||
"error": {
|
||||
"type": "invalid_request_error",
|
||||
"code": "content_policy_violation",
|
||||
"message": CONTENT_POLICY_MESSAGE,
|
||||
},
|
||||
},
|
||||
]
|
||||
)
|
||||
|
||||
chunks = []
|
||||
|
||||
async def _drain():
|
||||
async for chunk in iterator:
|
||||
chunks.append(chunk)
|
||||
|
||||
with pytest.raises(MidStreamFallbackError) as exc_info:
|
||||
await _drain()
|
||||
assert len(chunks) == 1
|
||||
assert isinstance(exc_info.value.original_exception, litellm.ContentPolicyViolationError)
|
||||
assert exc_info.value.is_pre_first_chunk is False
|
||||
assert exc_info.value.generated_content == "partial "
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_async_iterator_error_after_first_chunk_carries_generated_content():
|
||||
"""An error after streamed output must expose the accumulated text so the router's
|
||||
|
|
@ -265,7 +376,7 @@ def test_handle_logging_failed_response_maps_rate_limit_to_429():
|
|||
):
|
||||
iterator._handle_logging_failed_response()
|
||||
logged_exception = mock_run_async.call_args.kwargs["exception"]
|
||||
assert isinstance(logged_exception, litellm.APIError)
|
||||
assert isinstance(logged_exception, litellm.RateLimitError)
|
||||
assert logged_exception.status_code == 429
|
||||
assert "throttled" in str(logged_exception)
|
||||
|
||||
|
|
@ -282,10 +393,28 @@ def test_handle_logging_failed_response_maps_type_field_to_400():
|
|||
):
|
||||
iterator._handle_logging_failed_response()
|
||||
logged_exception = mock_run_async.call_args.kwargs["exception"]
|
||||
assert isinstance(logged_exception, litellm.APIError)
|
||||
assert isinstance(logged_exception, litellm.BadRequestError)
|
||||
assert logged_exception.status_code == 400
|
||||
|
||||
|
||||
def test_handle_logging_failed_response_logs_content_policy_violation():
|
||||
"""Failure logging must record the same typed exception the stream raises, so logging
|
||||
integrations see a content policy violation instead of a generic APIError."""
|
||||
iterator = _make_iterator()
|
||||
iterator.completed_response = _make_failed_chunk(
|
||||
{"type": "invalid_request_error", "code": "content_policy_violation", "message": CONTENT_POLICY_MESSAGE}
|
||||
)
|
||||
with (
|
||||
patch.object(import_module("litellm.responses.streaming_iterator"), "run_async_function") as mock_run_async,
|
||||
patch.object(import_module("litellm.responses.streaming_iterator"), "executor"),
|
||||
):
|
||||
iterator._handle_logging_failed_response()
|
||||
logged_exception = mock_run_async.call_args.kwargs["exception"]
|
||||
assert isinstance(logged_exception, litellm.ContentPolicyViolationError)
|
||||
assert logged_exception.status_code == 400
|
||||
assert CONTENT_POLICY_MESSAGE in str(logged_exception)
|
||||
|
||||
|
||||
def test_handle_logging_failed_response_records_usage_and_cost():
|
||||
"""Usage on a response.failed event must reach failure spend accounting via combined_usage_object."""
|
||||
iterator = _make_iterator()
|
||||
|
|
@ -357,7 +486,7 @@ def test_sync_iterator_raises_mid_stream_fallback_on_rate_limit_error_event():
|
|||
for _ in iterator:
|
||||
pass
|
||||
assert exc_info.value.status_code == 429
|
||||
assert isinstance(exc_info.value.original_exception, litellm.APIError)
|
||||
assert isinstance(exc_info.value.original_exception, litellm.RateLimitError)
|
||||
|
||||
|
||||
def test_every_openai_sdk_response_error_code_has_explicit_status_mapping():
|
||||
|
|
@ -413,7 +542,7 @@ def test_maybe_raise_for_response_failed_event_maps_image_code_to_400():
|
|||
chunk = Mock()
|
||||
chunk.type = "response.failed"
|
||||
chunk.response = mock_response_obj
|
||||
with pytest.raises(litellm.APIError) as exc_info:
|
||||
with pytest.raises(litellm.BadRequestError) as exc_info:
|
||||
iterator._maybe_raise_for_error_event(chunk)
|
||||
assert exc_info.value.status_code == 400
|
||||
assert not isinstance(exc_info.value, MidStreamFallbackError)
|
||||
|
|
|
|||
|
|
@ -3654,6 +3654,111 @@ async def test_aresponses_streaming_iterator_fallback():
|
|||
assert call_kwargs["disable_fallbacks"] is False
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_aresponses_streaming_content_policy_error_event_routes_to_content_policy_fallback():
|
||||
"""Regression: a mid-stream content_policy_violation error event never reached
|
||||
content_policy_fallbacks. The iterator raised a bare APIError the wrapper does not
|
||||
catch, and even once wrapped, the MidStreamFallbackError envelope was handed to the
|
||||
fallback dispatch, whose isinstance branch on ContentPolicyViolationError never matched.
|
||||
The stream below is the customer's shape: a raw OpenAI error event with code
|
||||
content_policy_violation, transformed by the real OpenAI config, and the router must
|
||||
call the content_policy_fallbacks target, not the general fallbacks one."""
|
||||
from litellm.llms.openai.responses.transformation import OpenAIResponsesAPIConfig
|
||||
from litellm.responses.streaming_iterator import ResponsesAPIStreamingIterator
|
||||
|
||||
router = litellm.Router(
|
||||
model_list=[
|
||||
{"model_name": "primary", "litellm_params": {"model": "openai/gpt-5.4", "api_key": "k1"}},
|
||||
{
|
||||
"model_name": "content-fallback",
|
||||
"litellm_params": {"model": "gemini/gemini-2.5-flash", "api_key": "k2"},
|
||||
},
|
||||
{"model_name": "general-fallback", "litellm_params": {"model": "openai/gpt-5-mini", "api_key": "k3"}},
|
||||
],
|
||||
fallbacks=[{"primary": ["general-fallback"]}],
|
||||
content_policy_fallbacks=[{"primary": ["content-fallback"]}],
|
||||
)
|
||||
error_event = {
|
||||
"type": "error",
|
||||
"sequence_number": 2,
|
||||
"error": {
|
||||
"type": "invalid_request_error",
|
||||
"code": "content_policy_violation",
|
||||
"message": "This content was flagged for possible cybersecurity risk. The response was halted mid-stream.",
|
||||
"param": None,
|
||||
},
|
||||
}
|
||||
|
||||
async def aiter_bytes():
|
||||
yield f"data: {json.dumps(error_event)}\n\n".encode()
|
||||
|
||||
raw_response = MagicMock()
|
||||
raw_response.headers = {}
|
||||
raw_response.aiter_bytes = aiter_bytes
|
||||
logging_obj = MagicMock(spec=LiteLLMLogging)
|
||||
logging_obj.model_call_details = {"litellm_params": {}}
|
||||
logging_obj.completion_start_time = None
|
||||
source = ResponsesAPIStreamingIterator(
|
||||
response=raw_response,
|
||||
model="gpt-5.4",
|
||||
responses_api_provider_config=OpenAIResponsesAPIConfig(),
|
||||
logging_obj=logging_obj,
|
||||
custom_llm_provider="openai",
|
||||
)
|
||||
fallback_chunks = [MagicMock(type="response.output_text.delta"), MagicMock(type="response.completed")]
|
||||
fallback_call = AsyncMock(return_value=_AsyncList(fallback_chunks))
|
||||
|
||||
wrapped = await router._aresponses_streaming_iterator(
|
||||
response=source,
|
||||
initial_kwargs={
|
||||
"model": "primary",
|
||||
"stream": True,
|
||||
"input": "Hi",
|
||||
"original_generic_function": fallback_call,
|
||||
},
|
||||
)
|
||||
collected = [chunk async for chunk in wrapped]
|
||||
|
||||
assert collected == fallback_chunks
|
||||
fallback_call.assert_awaited_once()
|
||||
assert fallback_call.await_args.kwargs["model"] == "gemini/gemini-2.5-flash"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_aresponses_streaming_iterator_unwraps_content_policy_trigger_for_fallback_dispatch():
|
||||
"""The fallback dispatch matches on the trigger's own type, so the wrapper must hand it the
|
||||
ContentPolicyViolationError carried inside MidStreamFallbackError, not the envelope."""
|
||||
router = _make_router_with_fallback("openai/gpt-5.4", "openai/gpt-5-mini")
|
||||
content_policy_error = litellm.ContentPolicyViolationError(
|
||||
message="flagged mid-stream", llm_provider="openai", model="openai/gpt-5.4"
|
||||
)
|
||||
src = _make_responses_iterator(
|
||||
chunks=[MagicMock(type="response.created")],
|
||||
error=MidStreamFallbackError(
|
||||
message=str(content_policy_error),
|
||||
model="openai/gpt-5.4",
|
||||
llm_provider="openai",
|
||||
original_exception=content_policy_error,
|
||||
is_pre_first_chunk=True,
|
||||
),
|
||||
model="openai/gpt-5.4",
|
||||
)
|
||||
|
||||
with patch.object(
|
||||
router,
|
||||
"async_function_with_fallbacks_common_utils",
|
||||
new=AsyncMock(return_value=_AsyncList([MagicMock(type="response.completed")])),
|
||||
) as mock_fallback_utils:
|
||||
wrapped = await router._aresponses_streaming_iterator(
|
||||
response=src,
|
||||
initial_kwargs={"model": "openai/gpt-5.4", "stream": True, "input": "Hi"},
|
||||
)
|
||||
[chunk async for chunk in wrapped]
|
||||
|
||||
mock_fallback_utils.assert_awaited_once()
|
||||
assert mock_fallback_utils.await_args.kwargs["e"] is content_policy_error
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize(
|
||||
"fallback_headers",
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue