diff --git a/tests/integration/streaming/test_responses_top_level_error_events.py b/tests/integration/streaming/test_responses_top_level_error_events.py index 63bdadceebc..2cbc4570dec 100644 --- a/tests/integration/streaming/test_responses_top_level_error_events.py +++ b/tests/integration/streaming/test_responses_top_level_error_events.py @@ -22,12 +22,14 @@ from integration._support.responses_stream import ( data_frames, delta, error_body, + error_event, frame, function_tools, healthy_stream, messages_body, openai_error_event, openai_error_stream, + response_object, serve, spend_row, sse_events, @@ -277,6 +279,38 @@ def test_native_responses_stream_fails_with_the_provider_code_and_message(gatewa _assert_failure_logged(response.headers["x-litellm-call-id"], "429", RATE_LIMIT_MESSAGE) +_FLEX_MESSAGE: Final = "Flex processing is temporarily unavailable. Please try again later or use standard processing." +_FLEX_ERROR: Final = {"type": "resource_unavailable", "code": "flex_unavailable", "message": _FLEX_MESSAGE} + + +def _flex_outage(identity: str, kind: str) -> tuple[bytes, ...]: + failure: Final = ( + error_event(_FLEX_ERROR) + if kind == "error" + else { + "type": "response.failed", + "sequence_number": 1, + "response": response_object(identity, "failed", error=_FLEX_ERROR), + } + ) + return (frame(created(identity)), frame(failure)) + + +@pytest.mark.parametrize("kind", ["error", "response.failed"]) +def test_native_responses_stream_flex_outage_fails_as_a_rate_limit(gateway: Gateway, kind: str) -> None: + response, error = _responses_frames(gateway, _flex_outage(_identity(), kind)) + assert error["code"] == "rate_limit_exceeded" and _FLEX_MESSAGE in string_value(error["message"]), error + _assert_failure_logged(response.headers["x-litellm-call-id"], "429", _FLEX_MESSAGE) + + +@pytest.mark.parametrize("kind", ["error", "response.failed"]) +def test_chat_stream_flex_outage_is_a_throttling_error_like_the_non_streamed_429(gateway: Gateway, kind: str) -> None: + response, _ = _chat(gateway, _flex_outage(_identity(), kind)) + assert response.status_code == 429, response.text + assert_provider_typed_rate_limit(error_body(response), _FLEX_MESSAGE) + _assert_failure_logged(response.headers["x-litellm-call-id"], "429", _FLEX_MESSAGE) + + def test_native_responses_stream_error_after_text_fails_with_the_provider_code(gateway: Gateway) -> None: identity: Final = _identity() stream: Final = (