mirror of
https://github.com/BerriAI/litellm.git
synced 2026-10-16 04:29:31 +00:00
fix(responses): keep the provider code and message on top-level stream error events (#46438)
* fix(responses): keep the provider code and message on top-level stream error events Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * fix(responses): log a mid-stream failure before the router falls back CustomStreamWrapper now waits for failure logging before it raises the error the router falls back on, so the cooldown lands on the deployment that failed instead of the fallback that replaced it. Also keeps numeric error codes and the top-level message on the WebSocket and ChatGPT paths. * fix(streaming): run sync failure callbacks before a failed stream falls back to another deployment The router cools a deployment down from a sync failure callback that reads the request's model_info. A failed stream submitted that callback to the logging executor without waiting, so it often ran after the router's fallback had overwritten model_info and cooled down the healthy fallback deployment instead, leaving the model group with no deployments * test(responses): cover top-level OpenAI stream error events end to end * test(streaming): read live worker pids so the second SIGKILL chaos variant sees two workers * fix(streaming): run waited sync failure logging off the shared logging pool * revert(streaming): keep failed-stream failure logging in the background Waiting for sync failure callbacks before a failed stream raises fixed the router cooldown race, but it slowed healthy calls under many failing streams with a slow callback, since the wait shares a thread pool with every call's start. The race is pre-existing on main and is tracked separately, so this PR goes back to main's logging timing and keeps only the top-level error event changes --------- Co-authored-by: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
This commit is contained in:
Mateo Wang
GitHub
parent
a114356542
commit
dba46c0513
13 changed files with 963 additions and 130 deletions
|
|
@ -226,7 +226,8 @@ class ChatGPTResponsesAPIConfig(OpenAIResponsesAPIConfig):
|
|||
def _extract_error_message(self, parsed_chunk: dict[str, Any]) -> str | None:
|
||||
error_obj: Final = parsed_chunk.get("error") or (parsed_chunk.get("response") or {}).get("error")
|
||||
if error_obj is None:
|
||||
return None
|
||||
top_level_message: Final = parsed_chunk.get("message")
|
||||
return top_level_message if isinstance(top_level_message, str) else None
|
||||
if isinstance(error_obj, dict):
|
||||
return error_obj.get("message") or str(error_obj)
|
||||
return str(error_obj)
|
||||
|
|
|
|||
|
|
@ -282,6 +282,21 @@ def _map_stream_error_to_exception(error_obj: object, model: str, custom_llm_pro
|
|||
return mapped_exception
|
||||
|
||||
|
||||
def _event_field(event: object, name: str) -> object:
|
||||
return event.get(name) if _is_json_object(event) else getattr(event, name, None)
|
||||
|
||||
|
||||
def _is_error_field_value(value: object) -> bool:
|
||||
return isinstance(value, (str, int)) and not isinstance(value, bool)
|
||||
|
||||
|
||||
def _error_event_error_object(event: object) -> object:
|
||||
nested_error: Final = _event_field(event, "error")
|
||||
if nested_error is not None:
|
||||
return nested_error
|
||||
return {field: value for field in ("message", "code") if _is_error_field_value(value := _event_field(event, field))}
|
||||
|
||||
|
||||
def _mid_stream_fallback_eligible(mapped_exception: Exception) -> bool:
|
||||
if isinstance(mapped_exception, litellm.ContentPolicyViolationError):
|
||||
return True
|
||||
|
|
@ -659,7 +674,7 @@ class BaseResponsesAPIStreamingIterator:
|
|||
error_obj: Final[object] = (
|
||||
getattr(getattr(result, "response", None), "error", None)
|
||||
if chunk_type == "response.failed"
|
||||
else getattr(result, "error", None)
|
||||
else _error_event_error_object(result)
|
||||
)
|
||||
|
||||
mapped_exception: Final = self._map_error_event_exception(error_obj)
|
||||
|
|
@ -1837,7 +1852,7 @@ _RESPONSES_WS_OUTPUT_ITEM_EVENT_TYPES: Final = frozenset({"response.output_item.
|
|||
|
||||
def _ws_event_error(event: Mapping[str, object]) -> object:
|
||||
if event.get("type") == "error":
|
||||
return event.get("error")
|
||||
return _error_event_error_object(event)
|
||||
response: Final = event.get("response")
|
||||
return response.get("error") if _is_json_object(response) else None
|
||||
|
||||
|
|
|
|||
|
|
@ -1932,7 +1932,10 @@ class ErrorEventError(BaseLiteLLMOpenAIResponseObject):
|
|||
class ErrorEvent(BaseLiteLLMOpenAIResponseObject):
|
||||
type: Literal[ResponsesAPIStreamEvents.ERROR]
|
||||
sequence_number: int
|
||||
error: ErrorEventError
|
||||
error: ErrorEventError | None = None
|
||||
code: str | int | None = None
|
||||
message: str | None = None
|
||||
param: str | Mapping[str, object] | None = None
|
||||
|
||||
|
||||
class GenericEvent(BaseLiteLLMOpenAIResponseObject):
|
||||
|
|
|
|||
|
|
@ -2,13 +2,17 @@ import json
|
|||
from collections.abc import Callable, Iterator, Mapping, Sequence
|
||||
from typing import Final
|
||||
|
||||
from integration._support.client import object_value, string_value
|
||||
import httpx
|
||||
from integration._support.client import JSON_OBJECT, eventually, object_value, string_value
|
||||
from integration._support.database import read_rows
|
||||
from integration._support.wire import Reply, Request
|
||||
from pydantic import JsonValue
|
||||
|
||||
AZURE_TARGET: Final = "/openai/v1/responses?api-version="
|
||||
OPENAI_TARGET: Final = "/responses"
|
||||
RATE_LIMIT_MESSAGE: Final = "Your requests to gpt-6 have exceeded token rate limit."
|
||||
RATE_LIMIT_PREFIX: Final = "litellm.RateLimitError: "
|
||||
SENTINEL_PREFIX: Final = "litellm.MidStreamFallbackError: "
|
||||
|
||||
|
||||
def frame(event: Mapping[str, JsonValue]) -> bytes:
|
||||
|
|
@ -36,6 +40,14 @@ def error_event(error: Mapping[str, JsonValue] | None) -> Mapping[str, JsonValue
|
|||
return {"type": "error", "sequence_number": 1, **({} if error is None else {"error": dict(error)})}
|
||||
|
||||
|
||||
def openai_error_event(code: str, message: str) -> Mapping[str, JsonValue]:
|
||||
return {"type": "error", "sequence_number": 1, "code": code, "message": message, "param": None}
|
||||
|
||||
|
||||
def openai_error_stream(identity: str, code: str, message: str) -> tuple[bytes, ...]:
|
||||
return (frame(created(identity)), frame(openai_error_event(code, message)))
|
||||
|
||||
|
||||
def azure_rate_limit() -> Mapping[str, JsonValue]:
|
||||
return {
|
||||
"type": "too_many_requests",
|
||||
|
|
@ -126,3 +138,78 @@ def chat_content(frames: Sequence[Mapping[str, JsonValue]]) -> str:
|
|||
yield string_value(object_value(object_value(choice)["delta"]).get("content") or "")
|
||||
|
||||
return "".join(deltas())
|
||||
|
||||
|
||||
def chat_body(model: str, marker: str, **extra: JsonValue) -> dict[str, JsonValue]:
|
||||
return {
|
||||
"model": model,
|
||||
"stream": True,
|
||||
"messages": [{"role": "user", "content": marker}],
|
||||
"tools": function_tools(),
|
||||
"num_retries": 0,
|
||||
"cache": {"no-cache": True},
|
||||
**extra,
|
||||
}
|
||||
|
||||
|
||||
def messages_body(model: str, marker: str) -> dict[str, JsonValue]:
|
||||
return {
|
||||
"model": model,
|
||||
"max_tokens": 64,
|
||||
"stream": True,
|
||||
"messages": [{"role": "user", "content": marker}],
|
||||
"tools": [
|
||||
{
|
||||
"name": "get_weather",
|
||||
"description": "Weather for a city",
|
||||
"input_schema": {"type": "object", "properties": {"city": {"type": "string"}}, "required": ["city"]},
|
||||
}
|
||||
],
|
||||
"num_retries": 0,
|
||||
"cache": {"no-cache": True},
|
||||
}
|
||||
|
||||
|
||||
def error_body(response: httpx.Response) -> Mapping[str, JsonValue]:
|
||||
return object_value(JSON_OBJECT.validate_json(response.content)["error"])
|
||||
|
||||
|
||||
def data_frames(text: str) -> tuple[Mapping[str, JsonValue], ...]:
|
||||
return tuple(
|
||||
JSON_OBJECT.validate_json(line.removeprefix("data: "))
|
||||
for line in text.splitlines()
|
||||
if line.startswith("data: ") and line != "data: [DONE]"
|
||||
)
|
||||
|
||||
|
||||
def sse_events(text: str) -> tuple[tuple[str, Mapping[str, JsonValue]], ...]:
|
||||
def parse(block: str) -> tuple[str, Mapping[str, JsonValue]]:
|
||||
lines: Final = block.splitlines()
|
||||
event: Final = next(line.removeprefix("event: ") for line in lines if line.startswith("event: "))
|
||||
data: Final = next(line.removeprefix("data: ") for line in lines if line.startswith("data: "))
|
||||
return event, JSON_OBJECT.validate_json(data)
|
||||
|
||||
return tuple(parse(block) for block in text.strip().split("\n\n") if "event: " in block)
|
||||
|
||||
|
||||
def spend_row(call_id: str) -> Mapping[str, JsonValue]:
|
||||
rows: Final = eventually(
|
||||
lambda: read_rows(
|
||||
'SELECT litellm_call_id, status, model_group, metadata FROM "LiteLLM_SpendLogs" WHERE litellm_call_id = %s',
|
||||
(call_id,),
|
||||
),
|
||||
lambda found: len(found) >= 1,
|
||||
seconds=70,
|
||||
)
|
||||
assert len(rows) == 1, rows
|
||||
return rows[0]
|
||||
|
||||
|
||||
def assert_provider_typed_rate_limit(
|
||||
error: Mapping[str, JsonValue], provider_message: str = RATE_LIMIT_MESSAGE
|
||||
) -> None:
|
||||
assert error["type"] == "throttling_error", error
|
||||
assert str(error["code"]) == "429", error
|
||||
message: Final = string_value(error["message"])
|
||||
assert message.startswith(RATE_LIMIT_PREFIX) and provider_message in message, message
|
||||
assert SENTINEL_PREFIX not in message, message
|
||||
|
|
|
|||
|
|
@ -0,0 +1,85 @@
|
|||
import uuid
|
||||
from collections.abc import Iterator, Mapping
|
||||
from dataclasses import dataclass
|
||||
from typing import Final
|
||||
|
||||
import httpx
|
||||
import pytest
|
||||
from integration._support import codex_vendor as cv
|
||||
from integration._support.client import Gateway, gateway_from_environment, object_value, string_value
|
||||
from integration._support.process import graceful_stop_seconds, owned_proxy_process
|
||||
from integration._support.responses_stream import (
|
||||
RATE_LIMIT_MESSAGE,
|
||||
RATE_LIMIT_PREFIX,
|
||||
data_frames,
|
||||
error_body,
|
||||
openai_error_stream,
|
||||
)
|
||||
from integration._support.wire import Reply, Request, Wire, wire_server
|
||||
from pydantic import JsonValue
|
||||
|
||||
pytestmark: Final = pytest.mark.timeout(2 * graceful_stop_seconds() + 120)
|
||||
|
||||
_MODEL: Final = "chatgpt/gpt-6.1-sol"
|
||||
_TOP_LEVEL_RATE_LIMIT: Final = "codex-top-level-rate-limit"
|
||||
_VENDOR: Final = cv.CodexVendor()
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class _Rig:
|
||||
gateway: Gateway
|
||||
wire: Wire
|
||||
|
||||
|
||||
def _respond(request: Request) -> Reply:
|
||||
if _TOP_LEVEL_RATE_LIMIT.encode() not in request.body:
|
||||
return _VENDOR.respond(request)
|
||||
stream: Final = openai_error_stream("resp_" + uuid.uuid4().hex, "rate_limit_exceeded", RATE_LIMIT_MESSAGE)
|
||||
return Reply(content_type="text/event-stream", chunks=stream)
|
||||
|
||||
|
||||
@pytest.fixture(scope="module")
|
||||
def rig(tmp_path_factory: pytest.TempPathFactory) -> Iterator[_Rig]:
|
||||
directory: Final = tmp_path_factory.mktemp("chatgpt-top-level-error-rig")
|
||||
with gateway_from_environment() as gateway, wire_server(_respond) as wire:
|
||||
overrides: Final = {"CHATGPT_TOKEN_DIR": str(cv.login(directory)), "CHATGPT_API_BASE": wire.url}
|
||||
config: Final = cv.proxy_config(directory, probe=False)
|
||||
with owned_proxy_process(gateway, directory, overrides, config=config, workers=2) as owned:
|
||||
yield _Rig(owned.gateway, wire)
|
||||
|
||||
|
||||
def _rate_limited(rig: _Rig, *, stream: bool) -> httpx.Response:
|
||||
rig.wire.drain()
|
||||
marker: Final = uuid.uuid4().hex
|
||||
with rig.gateway.scenario() as scenario:
|
||||
model: Final = scenario.model(model=_MODEL, api_base=rig.wire.url, api_key=None)
|
||||
body: Final[Mapping[str, JsonValue]] = {
|
||||
"model": model,
|
||||
"input": [{"role": "user", "content": f"Say marker-{marker} {_TOP_LEVEL_RATE_LIMIT}"}],
|
||||
"stream": stream,
|
||||
"cache": {"no-cache": True},
|
||||
}
|
||||
response: Final = rig.gateway.request("POST", "/v1/responses", body)
|
||||
(request,) = rig.wire.drain()
|
||||
cv.forwarded(request, marker)
|
||||
return response
|
||||
|
||||
|
||||
def test_chatgpt_stream_top_level_rate_limit_fails_with_the_provider_code_and_message(rig: _Rig) -> None:
|
||||
response: Final = _rate_limited(rig, stream=True)
|
||||
assert response.status_code == 200, response.text
|
||||
assert response.headers["content-type"].startswith("text/event-stream"), response.headers
|
||||
frames: Final = data_frames(response.text)
|
||||
assert [frame["type"] for frame in frames] == ["response.created", "response.failed"], response.text
|
||||
error: Final = object_value(object_value(frames[-1]["response"])["error"])
|
||||
assert error == {"code": "rate_limit_exceeded", "message": RATE_LIMIT_MESSAGE}, error
|
||||
|
||||
|
||||
def test_chatgpt_aggregated_top_level_rate_limit_answers_429_with_the_provider_message(rig: _Rig) -> None:
|
||||
response: Final = _rate_limited(rig, stream=False)
|
||||
assert response.status_code == 429, (response.status_code, response.text)
|
||||
assert response.headers["content-type"].startswith("application/json"), response.headers
|
||||
message: Final = string_value(error_body(response)["message"])
|
||||
assert message.startswith(RATE_LIMIT_PREFIX) and RATE_LIMIT_MESSAGE in message, message
|
||||
assert "event:" not in message, message
|
||||
assert rig.gateway.request("GET", "/health/liveliness", None).status_code == 200
|
||||
|
|
@ -1,30 +1,35 @@
|
|||
import asyncio
|
||||
import hashlib
|
||||
import itertools
|
||||
import json
|
||||
import ssl
|
||||
import threading
|
||||
import uuid
|
||||
from collections.abc import Iterator
|
||||
from collections.abc import Callable, Iterator
|
||||
from contextlib import contextmanager
|
||||
from dataclasses import dataclass
|
||||
from pathlib import Path
|
||||
from queue import SimpleQueue
|
||||
from typing import Final
|
||||
from typing import Final, TypeAlias
|
||||
|
||||
import pytest
|
||||
import websockets
|
||||
from integration._support.client import Gateway, gateway_from_environment
|
||||
from integration._support.process import owned_proxy
|
||||
from integration._support.client import Gateway, eventually, gateway_from_environment, object_value, string_value
|
||||
from integration._support.database import read_rows
|
||||
from integration._support.process import graceful_stop_seconds, owned_proxy
|
||||
from integration._support.responses_stream import RATE_LIMIT_MESSAGE, openai_error_event
|
||||
from integration._support.tls import server_context, write_self_signed_cert
|
||||
from pydantic import JsonValue
|
||||
from websockets.asyncio.server import ServerConnection, serve
|
||||
|
||||
pytestmark: Final = pytest.mark.timeout(180)
|
||||
pytestmark: Final = pytest.mark.timeout(2 * graceful_stop_seconds() + 120)
|
||||
|
||||
PROVIDER_MODEL: Final = "ws-peer-model"
|
||||
USAGE: Final = {"input_tokens": 5, "output_tokens": 2, "total_tokens": 7}
|
||||
TERMINAL: Final = frozenset({"response.completed", "response.failed", "error"})
|
||||
|
||||
_Script: TypeAlias = Callable[[str], tuple[dict[str, JsonValue], ...]]
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class Peer:
|
||||
|
|
@ -58,8 +63,20 @@ def _events(response_id: str, text: str) -> tuple[dict[str, JsonValue], ...]:
|
|||
)
|
||||
|
||||
|
||||
def _seven(response_id: str) -> tuple[dict[str, JsonValue], ...]:
|
||||
return _events(response_id, "seven")
|
||||
|
||||
|
||||
def _top_level_rate_limit(response_id: str) -> tuple[dict[str, JsonValue], ...]:
|
||||
created: Final = _events(response_id, "")[0]
|
||||
return (created, dict(openai_error_event("rate_limit_exceeded", RATE_LIMIT_MESSAGE)))
|
||||
|
||||
|
||||
async def _answer(
|
||||
connection: ServerConnection, paths: SimpleQueue[str], frames: SimpleQueue[dict[str, JsonValue]]
|
||||
connection: ServerConnection,
|
||||
paths: SimpleQueue[str],
|
||||
frames: SimpleQueue[dict[str, JsonValue]],
|
||||
script: _Script,
|
||||
) -> None:
|
||||
paths.put(connection.request.path if connection.request is not None else "")
|
||||
turns: Final = itertools.count(1)
|
||||
|
|
@ -68,7 +85,7 @@ async def _answer(
|
|||
frames.put(frame)
|
||||
if frame.get("type") != "response.create":
|
||||
continue
|
||||
for event in _events(f"resp_peer_{next(turns)}", "seven"):
|
||||
for event in script(f"resp_peer_{next(turns)}"):
|
||||
await connection.send(json.dumps(event))
|
||||
|
||||
|
||||
|
|
@ -78,21 +95,24 @@ async def _serve(
|
|||
frames: SimpleQueue[dict[str, JsonValue]],
|
||||
ports: SimpleQueue[int],
|
||||
stop: asyncio.Event,
|
||||
script: _Script,
|
||||
) -> None:
|
||||
async with serve(lambda connection: _answer(connection, paths, frames), "127.0.0.1", 0, ssl=tls) as server:
|
||||
async with serve(lambda connection: _answer(connection, paths, frames, script), "127.0.0.1", 0, ssl=tls) as server:
|
||||
ports.put(next(iter(server.sockets)).getsockname()[1])
|
||||
await stop.wait()
|
||||
|
||||
|
||||
@contextmanager
|
||||
def responses_peer(cert: tuple[Path, Path]) -> Iterator[Peer]:
|
||||
def responses_peer(cert: tuple[Path, Path], script: _Script = _seven) -> Iterator[Peer]:
|
||||
loop: Final = asyncio.new_event_loop()
|
||||
stop: Final = asyncio.Event()
|
||||
paths: Final = SimpleQueue[str]()
|
||||
frames: Final = SimpleQueue[dict[str, JsonValue]]()
|
||||
ports: Final = SimpleQueue[int]()
|
||||
thread: Final = threading.Thread(
|
||||
target=loop.run_until_complete, args=(_serve(server_context(*cert), paths, frames, ports, stop),), daemon=True
|
||||
target=loop.run_until_complete,
|
||||
args=(_serve(server_context(*cert), paths, frames, ports, stop, script),),
|
||||
daemon=True,
|
||||
)
|
||||
thread.start()
|
||||
try:
|
||||
|
|
@ -207,3 +227,29 @@ def test_previous_response_id_from_the_first_turn_reaches_the_provider_as_its_ow
|
|||
assert [frame["input"][0]["content"][0]["text"] for frame in forwarded] == list(texts)
|
||||
assert "previous_response_id" not in forwarded[0]
|
||||
assert forwarded[1]["previous_response_id"] == "resp_peer_1"
|
||||
|
||||
|
||||
def test_a_top_level_error_frame_reaches_the_client_and_logs_the_provider_code_and_message(
|
||||
candidate: Gateway, cert: tuple[Path, Path]
|
||||
) -> None:
|
||||
with responses_peer(cert, _top_level_rate_limit) as peer, candidate.scenario() as scenario:
|
||||
model: Final = scenario.model(model=f"openai/{PROVIDER_MODEL}", api_base=peer.url)
|
||||
key: Final = scenario.key(models=[model])
|
||||
text: Final = f"say seven {uuid.uuid4().hex}"
|
||||
(events,) = asyncio.run(_session(str(candidate.client.base_url), key, model, (text,)))
|
||||
assert [event["type"] for event in events] == ["response.created", "error"], events
|
||||
assert (events[-1]["code"], events[-1]["message"]) == ("rate_limit_exceeded", RATE_LIMIT_MESSAGE), events
|
||||
(forwarded,) = _drain(peer.frames)
|
||||
assert forwarded["input"][0]["content"][0]["text"] == text
|
||||
(row,) = eventually(
|
||||
lambda: read_rows(
|
||||
'SELECT status, metadata FROM "LiteLLM_SpendLogs" WHERE api_key = %s',
|
||||
(hashlib.sha256(key.encode()).hexdigest(),),
|
||||
),
|
||||
lambda rows: len(rows) >= 1,
|
||||
seconds=70,
|
||||
)
|
||||
assert row["status"] == "failure", row
|
||||
error: Final = object_value(object_value(row["metadata"])["error_information"])
|
||||
assert str(error["error_code"]) == "429", error
|
||||
assert RATE_LIMIT_MESSAGE in string_value(error["error_message"]), error
|
||||
|
|
|
|||
|
|
@ -6,6 +6,8 @@ from integration._support.responses_stream import (
|
|||
AZURE_TARGET,
|
||||
RATE_LIMIT_MESSAGE,
|
||||
function_tools,
|
||||
healthy_stream,
|
||||
openai_error_stream,
|
||||
rate_limited_stream,
|
||||
serve,
|
||||
)
|
||||
|
|
@ -13,7 +15,7 @@ from integration._support.wire import Wire, wire_server
|
|||
|
||||
import litellm
|
||||
from litellm import Router
|
||||
from litellm.exceptions import MidStreamFallbackError, RateLimitError
|
||||
from litellm.exceptions import APIError, MidStreamFallbackError, RateLimitError
|
||||
from litellm.litellm_core_utils.streaming_handler import CustomStreamWrapper
|
||||
|
||||
_MODEL: Final = "azure/gpt-6"
|
||||
|
|
@ -135,3 +137,130 @@ async def test_router_async_stream_in_stream_rate_limit_with_fallbacks_disabled_
|
|||
async for _ in response:
|
||||
pass
|
||||
_assert_surfaces_the_provider_exception(raised.value, wire, marker)
|
||||
|
||||
|
||||
def _openai_error_stream(marker: str, code: str, message: str) -> tuple[bytes, ...]:
|
||||
return openai_error_stream(f"resp_{marker}", code, message)
|
||||
|
||||
|
||||
def test_sync_completion_stream_openai_error_event_raises_the_provider_code_and_message() -> None:
|
||||
marker: Final = uuid.uuid4().hex
|
||||
message: Final = f"The server had an error processing your request {marker}"
|
||||
with wire_server(serve(_openai_error_stream(marker, "server_error", message), AZURE_TARGET)) as wire:
|
||||
response: Final = litellm.completion(
|
||||
model=_MODEL,
|
||||
messages=_messages(marker),
|
||||
tools=_TOOLS,
|
||||
stream=True,
|
||||
num_retries=0,
|
||||
api_base=wire.url,
|
||||
api_key=_API_KEY,
|
||||
)
|
||||
with pytest.raises(MidStreamFallbackError) as raised:
|
||||
for _ in response:
|
||||
pass
|
||||
assert type(raised.value.original_exception) is APIError, repr(raised.value.original_exception)
|
||||
assert raised.value.status_code == 500 and message in str(raised.value), str(raised.value)
|
||||
_assert_one_attempt(wire, marker)
|
||||
|
||||
|
||||
async def test_router_async_stream_openai_error_event_rate_limit_surfaces_the_provider_exception() -> None:
|
||||
marker: Final = uuid.uuid4().hex
|
||||
stream: Final = _openai_error_stream(marker, "rate_limit_exceeded", RATE_LIMIT_MESSAGE)
|
||||
with wire_server(serve(stream, AZURE_TARGET)) as wire:
|
||||
response: Final = await _router(wire).acompletion(
|
||||
model=_GROUP, messages=_messages(marker), tools=_TOOLS, stream=True
|
||||
)
|
||||
with pytest.raises(RateLimitError) as raised:
|
||||
async for _ in response:
|
||||
pass
|
||||
_assert_surfaces_the_provider_exception(raised.value, wire, marker)
|
||||
|
||||
|
||||
async def test_async_completion_stream_top_level_rate_limit_wraps_the_provider_exception_once() -> None:
|
||||
marker: Final = uuid.uuid4().hex
|
||||
stream: Final = _openai_error_stream(marker, "rate_limit_exceeded", RATE_LIMIT_MESSAGE)
|
||||
with wire_server(serve(stream, AZURE_TARGET)) as wire:
|
||||
response: Final = await litellm.acompletion(
|
||||
model=_MODEL,
|
||||
messages=_messages(marker),
|
||||
tools=_TOOLS,
|
||||
stream=True,
|
||||
num_retries=0,
|
||||
api_base=wire.url,
|
||||
api_key=_API_KEY,
|
||||
)
|
||||
with pytest.raises(MidStreamFallbackError) as raised:
|
||||
async for _ in response:
|
||||
pass
|
||||
_assert_wraps_the_provider_exception_once(raised.value, wire, marker)
|
||||
|
||||
|
||||
def test_router_sync_stream_top_level_rate_limit_with_fallbacks_disabled_surfaces_the_provider_exception() -> None:
|
||||
marker: Final = uuid.uuid4().hex
|
||||
stream: Final = _openai_error_stream(marker, "rate_limit_exceeded", RATE_LIMIT_MESSAGE)
|
||||
with wire_server(serve(stream, AZURE_TARGET)) as wire:
|
||||
response: Final = _router(wire).completion(
|
||||
model=_GROUP, messages=_messages(marker), tools=_TOOLS, stream=True, disable_fallbacks=True
|
||||
)
|
||||
with pytest.raises(RateLimitError) as raised:
|
||||
for _ in response:
|
||||
pass
|
||||
_assert_surfaces_the_provider_exception(raised.value, wire, marker)
|
||||
|
||||
|
||||
async def test_router_async_stream_top_level_rate_limit_falls_back_to_the_healthy_deployment() -> None:
|
||||
marker: Final = uuid.uuid4().hex
|
||||
limited: Final = _openai_error_stream(marker, "rate_limit_exceeded", RATE_LIMIT_MESSAGE)
|
||||
with (
|
||||
wire_server(serve(limited, AZURE_TARGET)) as primary,
|
||||
wire_server(serve(healthy_stream(f"resp_{marker}", "fallback answer"), AZURE_TARGET)) as spare,
|
||||
):
|
||||
router: Final = Router(
|
||||
model_list=[
|
||||
{"model_name": name, "litellm_params": {"model": _MODEL, "api_base": wire.url, "api_key": _API_KEY}}
|
||||
for name, wire in ((_GROUP, primary), ("bridged-spare", spare))
|
||||
],
|
||||
num_retries=0,
|
||||
disable_cooldowns=True,
|
||||
fallbacks=[{_GROUP: ["bridged-spare"]}],
|
||||
)
|
||||
response: Final = await router.acompletion(model=_GROUP, messages=_messages(marker), tools=_TOOLS, stream=True)
|
||||
content: Final = "".join([chunk.choices[0].delta.content or "" async for chunk in response if chunk.choices])
|
||||
assert content == "fallback answer", content
|
||||
_assert_one_attempt(primary, marker)
|
||||
_assert_one_attempt(spare, marker)
|
||||
|
||||
|
||||
def _provider_rate_limit(raised: Exception, message: str) -> None:
|
||||
inner: Final = raised.original_exception if isinstance(raised, MidStreamFallbackError) else raised
|
||||
assert type(inner) is RateLimitError, repr(inner)
|
||||
assert getattr(raised, "status_code", None) == 429 and message in str(raised), str(raised)
|
||||
|
||||
|
||||
def test_sync_responses_stream_top_level_rate_limit_raises_the_provider_rate_limit() -> None:
|
||||
marker: Final = uuid.uuid4().hex
|
||||
stream: Final = _openai_error_stream(marker, "rate_limit_exceeded", RATE_LIMIT_MESSAGE)
|
||||
with wire_server(serve(stream, AZURE_TARGET)) as wire:
|
||||
response: Final = litellm.responses(
|
||||
model=_MODEL, input=marker, stream=True, api_base=wire.url, api_key=_API_KEY
|
||||
)
|
||||
with pytest.raises((MidStreamFallbackError, RateLimitError)) as raised:
|
||||
for _ in response:
|
||||
pass
|
||||
_provider_rate_limit(raised.value, RATE_LIMIT_MESSAGE)
|
||||
_assert_one_attempt(wire, marker)
|
||||
|
||||
|
||||
async def test_async_responses_stream_top_level_insufficient_quota_raises_a_rate_limit() -> None:
|
||||
marker: Final = uuid.uuid4().hex
|
||||
quota: Final = "You exceeded your current quota, please check your plan and billing details."
|
||||
with wire_server(serve(_openai_error_stream(marker, "insufficient_quota", quota), AZURE_TARGET)) as wire:
|
||||
response: Final = await litellm.aresponses(
|
||||
model=_MODEL, input=marker, stream=True, api_base=wire.url, api_key=_API_KEY
|
||||
)
|
||||
with pytest.raises((MidStreamFallbackError, RateLimitError)) as raised:
|
||||
async for _ in response:
|
||||
pass
|
||||
_provider_rate_limit(raised.value, quota)
|
||||
_assert_one_attempt(wire, marker)
|
||||
|
|
|
|||
|
|
@ -27,9 +27,11 @@ from integration._support.database import read_rows
|
|||
from integration._support.process import OwnedProxy, graceful_stop_seconds, owned_proxy_process
|
||||
from integration._support.responses_stream import (
|
||||
AZURE_TARGET,
|
||||
RATE_LIMIT_MESSAGE,
|
||||
chat_content,
|
||||
function_tools,
|
||||
healthy_stream,
|
||||
openai_error_stream,
|
||||
rate_limited_stream,
|
||||
)
|
||||
from integration._support.wire import Reply, Request, wire_server
|
||||
|
|
@ -39,15 +41,33 @@ pytestmark: Final = pytest.mark.timeout(2 * graceful_stop_seconds() + 120)
|
|||
|
||||
_MODEL: Final = "bridged-stream-chaos"
|
||||
_MARKER: Final = re.compile(r"marker-([0-9a-f]{32})")
|
||||
_STARTED_WORKER: Final = re.compile(r"Started server process \[(\d+)\]")
|
||||
_SENTINEL_PREFIX: Final = "litellm.MidStreamFallbackError: "
|
||||
_RATE_LIMIT_PREFIX: Final = "litellm.RateLimitError: "
|
||||
_HEADERS: Final[Mapping[str, str]] = MappingProxyType({"anthropic-version": "2023-06-01"})
|
||||
|
||||
Kind: TypeAlias = Literal["chat_limited", "chat_healthy", "messages_limited", "responses_limited"]
|
||||
Kind: TypeAlias = Literal[
|
||||
"chat_limited",
|
||||
"chat_healthy",
|
||||
"messages_limited",
|
||||
"responses_limited",
|
||||
"chat_top_level",
|
||||
"messages_top_level",
|
||||
"responses_top_level",
|
||||
"chat_dropped",
|
||||
]
|
||||
_KINDS: Final[tuple[Kind, ...]] = ("chat_limited", "chat_healthy", "messages_limited", "responses_limited")
|
||||
_TOP_LEVEL_KINDS: Final[tuple[Kind, ...]] = (
|
||||
"chat_top_level",
|
||||
"chat_healthy",
|
||||
"messages_top_level",
|
||||
"responses_top_level",
|
||||
"chat_dropped",
|
||||
)
|
||||
_CHAT_KINDS: Final[tuple[Kind, ...]] = ("chat_limited", "chat_healthy")
|
||||
_LOGGED_KINDS: Final[frozenset[Kind]] = frozenset({"chat_limited", "chat_healthy", "responses_limited"})
|
||||
_TOP_LEVEL_CHAT_KINDS: Final[tuple[Kind, ...]] = ("chat_top_level", "chat_healthy")
|
||||
_LOGGED_KINDS: Final[frozenset[Kind]] = frozenset(
|
||||
{"chat_limited", "chat_healthy", "responses_limited", "chat_top_level", "responses_top_level", "chat_dropped"}
|
||||
)
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
|
|
@ -128,16 +148,25 @@ def _respond(request: Request) -> Reply:
|
|||
identity: Final = f"resp_{uuid.uuid4().hex}"
|
||||
if b"chat_healthy" in request.body:
|
||||
return Reply(content_type="text/event-stream", chunks=healthy_stream(identity, f"answer marker-{marker}"))
|
||||
if b"chat_dropped" in request.body:
|
||||
return Reply(
|
||||
content_type="text/event-stream", chunks=healthy_stream(identity, f"answer marker-{marker}"), abort_after=1
|
||||
)
|
||||
if b"_top_level" in request.body:
|
||||
return Reply(
|
||||
content_type="text/event-stream",
|
||||
chunks=openai_error_stream(identity, "rate_limit_exceeded", RATE_LIMIT_MESSAGE),
|
||||
)
|
||||
return Reply(content_type="text/event-stream", chunks=rate_limited_stream(identity))
|
||||
|
||||
|
||||
def _path(kind: Kind) -> str:
|
||||
match kind:
|
||||
case "chat_limited" | "chat_healthy":
|
||||
case "chat_limited" | "chat_healthy" | "chat_top_level" | "chat_dropped":
|
||||
return "/v1/chat/completions"
|
||||
case "messages_limited":
|
||||
case "messages_limited" | "messages_top_level":
|
||||
return "/v1/messages"
|
||||
case "responses_limited":
|
||||
case "responses_limited" | "responses_top_level":
|
||||
return "/v1/responses"
|
||||
|
||||
|
||||
|
|
@ -150,9 +179,9 @@ def _body(call: _Call) -> Mapping[str, JsonValue]:
|
|||
"cache": {"no-cache": True},
|
||||
}
|
||||
match call.kind:
|
||||
case "chat_limited" | "chat_healthy":
|
||||
case "chat_limited" | "chat_healthy" | "chat_top_level" | "chat_dropped":
|
||||
return {**common, "messages": [{"role": "user", "content": prompt}], "tools": function_tools()}
|
||||
case "messages_limited":
|
||||
case "messages_limited" | "messages_top_level":
|
||||
return {
|
||||
**common,
|
||||
"max_tokens": 64,
|
||||
|
|
@ -165,7 +194,7 @@ def _body(call: _Call) -> Mapping[str, JsonValue]:
|
|||
}
|
||||
],
|
||||
}
|
||||
case "responses_limited":
|
||||
case "responses_limited" | "responses_top_level":
|
||||
return {**common, "input": prompt}
|
||||
|
||||
|
||||
|
|
@ -216,13 +245,14 @@ def _assert_answered_in_its_own_shape(served: _Served) -> None:
|
|||
case "chat_healthy":
|
||||
assert served.status == 200, served.text
|
||||
assert chat_content(_data_frames(served.text)) == f"answer marker-{served.call.marker}", served.text
|
||||
case "chat_limited":
|
||||
case "chat_limited" | "chat_top_level":
|
||||
assert served.status == 429, served.text
|
||||
error: Final = object_value(JSON_OBJECT.validate_json(served.text)["error"])
|
||||
assert error["type"] == "throttling_error" and str(error["code"]) == "429", error
|
||||
message: Final = string_value(error["message"])
|
||||
assert message.startswith(_RATE_LIMIT_PREFIX) and _SENTINEL_PREFIX not in message, message
|
||||
case "messages_limited":
|
||||
assert RATE_LIMIT_MESSAGE in message, message
|
||||
case "messages_limited" | "messages_top_level":
|
||||
assert served.status == 200, served.text
|
||||
events: Final = _sse_events(served.text)
|
||||
assert events[0][0] == "message_start" and events[-1][0] == "error", events
|
||||
|
|
@ -230,10 +260,21 @@ def _assert_answered_in_its_own_shape(served: _Served) -> None:
|
|||
assert frame_error["type"] == "rate_limit_error", frame_error
|
||||
frame_message: Final = string_value(frame_error["message"])
|
||||
assert frame_message.startswith(_RATE_LIMIT_PREFIX) and _SENTINEL_PREFIX not in frame_message, frame_message
|
||||
assert RATE_LIMIT_MESSAGE in frame_message, frame_message
|
||||
case "responses_limited":
|
||||
assert served.status == 200, served.text
|
||||
kinds: Final = [frame["type"] for frame in _data_frames(served.text)]
|
||||
assert kinds == ["response.created", "response.failed"], served.text
|
||||
case "responses_top_level":
|
||||
assert served.status == 200, served.text
|
||||
frames: Final = _data_frames(served.text)
|
||||
assert [frame["type"] for frame in frames] == ["response.created", "response.failed"], served.text
|
||||
failure: Final = object_value(object_value(frames[-1]["response"])["error"])
|
||||
assert failure["code"] == "rate_limit_exceeded", failure
|
||||
assert RATE_LIMIT_MESSAGE in string_value(failure["message"]), failure
|
||||
case "chat_dropped":
|
||||
assert served.status == 500, served.text
|
||||
assert "error" in JSON_OBJECT.validate_json(served.text), served.text
|
||||
|
||||
|
||||
def _assert_forwarded(received: tuple[Request, ...], calls: tuple[_Call, ...]) -> None:
|
||||
|
|
@ -263,12 +304,8 @@ def _assert_each_lands_once(served: tuple[_Served, ...]) -> None:
|
|||
assert by_call[item.call_id]["status"] == expected, (item.call_id, rows)
|
||||
|
||||
|
||||
def _worker_pids(log: Path, count: int) -> tuple[int, ...]:
|
||||
return eventually(
|
||||
lambda: tuple(int(found.group(1)) for found in _STARTED_WORKER.finditer(log.read_text())),
|
||||
lambda pids: len(pids) == count,
|
||||
seconds=30,
|
||||
)
|
||||
def _worker_pids(proxy: OwnedProxy, count: int) -> frozenset[int]:
|
||||
return eventually(proxy.worker_pids, lambda pids: len(pids) == count, seconds=30)
|
||||
|
||||
|
||||
def _open_upstream_connections(pid: int, upstream: str) -> int:
|
||||
|
|
@ -299,24 +336,28 @@ async def _held_burst(gateway: Gateway, calls: tuple[_Call, ...], held: _Held) -
|
|||
return burst
|
||||
|
||||
|
||||
@pytest.mark.parametrize("kinds", [_KINDS, _TOP_LEVEL_KINDS], ids=["nested", "top_level"])
|
||||
async def test_mixed_burst_of_bridged_streams_answers_each_call_in_its_own_shape_and_logs_each_once(
|
||||
rig: _Rig,
|
||||
rig: _Rig, kinds: tuple[Kind, ...]
|
||||
) -> None:
|
||||
calls: Final = _calls(24, _KINDS)
|
||||
calls: Final = _calls(6 * len(kinds), kinds)
|
||||
with wire_server(_respond, port=rig.port) as wire:
|
||||
served: Final = await _burst(rig.gateway, calls)
|
||||
assert len(served) == 24
|
||||
assert len(served) == len(calls)
|
||||
for item in served:
|
||||
_assert_answered_in_its_own_shape(item)
|
||||
_assert_forwarded(wire.drain(), calls)
|
||||
_assert_each_lands_once(served)
|
||||
|
||||
|
||||
async def test_worker_sigkill_mid_burst_leaves_the_sibling_answering_the_bridged_streams(rig: _Rig) -> None:
|
||||
calls: Final = _calls(20, _CHAT_KINDS)
|
||||
@pytest.mark.parametrize("kinds", [_CHAT_KINDS, _TOP_LEVEL_CHAT_KINDS], ids=["nested", "top_level"])
|
||||
async def test_worker_sigkill_mid_burst_leaves_the_sibling_answering_the_bridged_streams(
|
||||
rig: _Rig, kinds: tuple[Kind, ...]
|
||||
) -> None:
|
||||
calls: Final = _calls(20, kinds)
|
||||
held: Final = _Held(threading.Event(), SimpleQueue())
|
||||
with wire_server(held.respond, port=rig.port) as wire:
|
||||
workers: Final = _worker_pids(rig.proxy.log, 2)
|
||||
workers: Final = _worker_pids(rig.proxy, 2)
|
||||
burst: Final = await _held_burst(rig.gateway, calls, held)
|
||||
held_by: Final = MappingProxyType({pid: _open_upstream_connections(pid, wire.url) for pid in workers})
|
||||
assert sum(held_by.values()) == 20, held_by
|
||||
|
|
|
|||
|
|
@ -13,104 +13,47 @@ import openai
|
|||
import pytest
|
||||
import yaml
|
||||
from integration._support.client import (
|
||||
JSON_OBJECT,
|
||||
Gateway,
|
||||
eventually,
|
||||
gateway_from_environment,
|
||||
object_value,
|
||||
string_value,
|
||||
)
|
||||
from integration._support.database import read_rows
|
||||
from integration._support.process import graceful_stop_seconds, owned_proxy
|
||||
from integration._support.responses_stream import (
|
||||
AZURE_TARGET,
|
||||
OPENAI_TARGET,
|
||||
RATE_LIMIT_MESSAGE,
|
||||
RATE_LIMIT_PREFIX,
|
||||
SENTINEL_PREFIX,
|
||||
assert_provider_typed_rate_limit,
|
||||
azure_rate_limit,
|
||||
chat_body,
|
||||
chat_content,
|
||||
created,
|
||||
data_frames,
|
||||
delta,
|
||||
error_body,
|
||||
error_event,
|
||||
failed,
|
||||
frame,
|
||||
function_tools,
|
||||
healthy_stream,
|
||||
messages_body,
|
||||
openai_error_event,
|
||||
openai_error_stream,
|
||||
rate_limited_stream,
|
||||
serve,
|
||||
spend_row,
|
||||
sse_events,
|
||||
)
|
||||
from integration._support.wire import Reply, Request, Wire, wire_server
|
||||
from pydantic import JsonValue
|
||||
|
||||
_RATE_LIMIT_PREFIX: Final = "litellm.RateLimitError: "
|
||||
_SENTINEL_PREFIX: Final = "litellm.MidStreamFallbackError: "
|
||||
_PRIMARY: Final = "bridged-primary"
|
||||
_SPARE: Final = "bridged-spare"
|
||||
pytestmark: Final = pytest.mark.timeout(2 * graceful_stop_seconds() + 120)
|
||||
|
||||
|
||||
def chat_body(model: str, marker: str, **extra: JsonValue) -> dict[str, JsonValue]:
|
||||
return {
|
||||
"model": model,
|
||||
"stream": True,
|
||||
"messages": [{"role": "user", "content": marker}],
|
||||
"tools": function_tools(),
|
||||
"num_retries": 0,
|
||||
"cache": {"no-cache": True},
|
||||
**extra,
|
||||
}
|
||||
|
||||
|
||||
def messages_body(model: str, marker: str) -> dict[str, JsonValue]:
|
||||
return {
|
||||
"model": model,
|
||||
"max_tokens": 64,
|
||||
"stream": True,
|
||||
"messages": [{"role": "user", "content": marker}],
|
||||
"tools": [
|
||||
{
|
||||
"name": "get_weather",
|
||||
"description": "Weather for a city",
|
||||
"input_schema": {"type": "object", "properties": {"city": {"type": "string"}}, "required": ["city"]},
|
||||
}
|
||||
],
|
||||
"num_retries": 0,
|
||||
"cache": {"no-cache": True},
|
||||
}
|
||||
|
||||
|
||||
def error_body(response: httpx.Response) -> Mapping[str, JsonValue]:
|
||||
return object_value(JSON_OBJECT.validate_json(response.content)["error"])
|
||||
|
||||
|
||||
def data_frames(text: str) -> tuple[Mapping[str, JsonValue], ...]:
|
||||
return tuple(
|
||||
JSON_OBJECT.validate_json(line.removeprefix("data: "))
|
||||
for line in text.splitlines()
|
||||
if line.startswith("data: ") and line != "data: [DONE]"
|
||||
)
|
||||
|
||||
|
||||
def spend_row(call_id: str) -> Mapping[str, JsonValue]:
|
||||
rows: Final = eventually(
|
||||
lambda: read_rows(
|
||||
'SELECT litellm_call_id, status, model_group FROM "LiteLLM_SpendLogs" WHERE litellm_call_id = %s',
|
||||
(call_id,),
|
||||
),
|
||||
lambda found: len(found) >= 1,
|
||||
seconds=70,
|
||||
)
|
||||
assert len(rows) == 1, rows
|
||||
return rows[0]
|
||||
|
||||
|
||||
def assert_provider_typed_rate_limit(error: Mapping[str, JsonValue]) -> None:
|
||||
assert error["type"] == "throttling_error", error
|
||||
assert str(error["code"]) == "429", error
|
||||
message: Final = string_value(error["message"])
|
||||
assert message.startswith(_RATE_LIMIT_PREFIX) and RATE_LIMIT_MESSAGE in message, message
|
||||
assert _SENTINEL_PREFIX not in message, message
|
||||
|
||||
|
||||
def assert_failed_once(wire: Wire, call_id: str, model: str, attempts: int = 1) -> tuple[Request, ...]:
|
||||
received: Final = wire.drain()
|
||||
assert len(received) == attempts, [request.target for request in received]
|
||||
|
|
@ -237,6 +180,55 @@ def test_bridged_in_stream_rate_limit_whose_fallback_is_also_rate_limited_answer
|
|||
assert spend_row(response.headers["x-litellm-call-id"])["status"] == "failure"
|
||||
|
||||
|
||||
def test_bridged_top_level_rate_limit_falls_back_to_the_healthy_deployment(fallback_proxy: FallbackProxy) -> None:
|
||||
identity: Final = "resp_" + uuid.uuid4().hex
|
||||
limited: Final = openai_error_stream(identity, "rate_limit_exceeded", RATE_LIMIT_MESSAGE)
|
||||
with (
|
||||
wire_server(serve(limited, AZURE_TARGET), port=fallback_proxy.primary_port) as primary,
|
||||
wire_server(
|
||||
serve(healthy_stream(identity, "fallback answer"), AZURE_TARGET), port=fallback_proxy.spare_port
|
||||
) as spare,
|
||||
):
|
||||
response: Final = _chat(fallback_proxy.gateway, chat_body(_PRIMARY, identity))
|
||||
assert response.status_code == 200, response.text
|
||||
assert chat_content(data_frames(response.text)) == "fallback answer", response.text
|
||||
assert len(primary.drain()) == 1 and len(spare.drain()) == 1
|
||||
row: Final = spend_row(response.headers["x-litellm-call-id"])
|
||||
assert row["status"] == "success" and row["model_group"] == _SPARE, row
|
||||
|
||||
|
||||
_INVALID_PROMPT: Final = "Invalid prompt: your prompt was flagged."
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"rejection",
|
||||
[
|
||||
pytest.param(
|
||||
error_event({"type": "invalid_request_error", "code": "invalid_prompt", "message": _INVALID_PROMPT}),
|
||||
id="nested",
|
||||
),
|
||||
pytest.param(openai_error_event("invalid_prompt", _INVALID_PROMPT), id="top_level"),
|
||||
],
|
||||
)
|
||||
def test_bridged_invalid_prompt_answers_a_bad_request_without_trying_the_fallback(
|
||||
fallback_proxy: FallbackProxy, rejection: Mapping[str, JsonValue]
|
||||
) -> None:
|
||||
identity: Final = "resp_" + uuid.uuid4().hex
|
||||
with (
|
||||
wire_server(
|
||||
serve((frame(created(identity)), frame(rejection)), AZURE_TARGET), port=fallback_proxy.primary_port
|
||||
) as primary,
|
||||
wire_server(
|
||||
serve(healthy_stream(identity, "fallback answer"), AZURE_TARGET), port=fallback_proxy.spare_port
|
||||
) as spare,
|
||||
):
|
||||
response: Final = _chat(fallback_proxy.gateway, chat_body(_PRIMARY, identity))
|
||||
assert response.status_code == 400, response.text
|
||||
message: Final = string_value(error_body(response)["message"])
|
||||
assert message.startswith("litellm.BadRequestError: ") and _INVALID_PROMPT in message, message
|
||||
assert len(primary.drain()) == 1 and len(spare.drain()) == 0
|
||||
|
||||
|
||||
def _in_stream_error_status(gateway: Gateway, stream: tuple[bytes, ...]) -> tuple[httpx.Response, str]:
|
||||
with wire_server(serve(stream, AZURE_TARGET)) as wire, gateway.scenario() as scenario:
|
||||
model: Final = scenario.model(model="azure/gpt-6", api_base=wire.url, api_key="synthetic-azure-key")
|
||||
|
|
@ -258,7 +250,7 @@ def test_bridged_in_stream_server_error_reaches_the_client_as_the_provider_error
|
|||
message: Final = string_value(error["message"])
|
||||
assert str(error["code"]) == "500", error
|
||||
assert message.startswith("litellm.APIError: ") and "The server had an error" in message, message
|
||||
assert _SENTINEL_PREFIX not in message, message
|
||||
assert SENTINEL_PREFIX not in message, message
|
||||
|
||||
|
||||
def test_bridged_in_stream_invalid_prompt_is_a_bad_request_on_both_legs(gateway: Gateway) -> None:
|
||||
|
|
@ -274,7 +266,7 @@ def test_bridged_in_stream_invalid_prompt_is_a_bad_request_on_both_legs(gateway:
|
|||
message: Final = string_value(error["message"])
|
||||
assert str(error["code"]) == "400", error
|
||||
assert message.startswith("litellm.BadRequestError: ") and "Invalid prompt" in message, message
|
||||
assert _SENTINEL_PREFIX not in message, message
|
||||
assert SENTINEL_PREFIX not in message, message
|
||||
|
||||
|
||||
def test_bridged_error_event_without_an_error_object_is_a_provider_typed_internal_error(gateway: Gateway) -> None:
|
||||
|
|
@ -286,7 +278,7 @@ def test_bridged_error_event_without_an_error_object_is_a_provider_typed_interna
|
|||
message: Final = string_value(error["message"])
|
||||
assert str(error["code"]) == "500", error
|
||||
assert message.startswith("litellm.APIError: ") and "Response API in-stream error" in message, message
|
||||
assert _SENTINEL_PREFIX not in message, message
|
||||
assert SENTINEL_PREFIX not in message, message
|
||||
|
||||
|
||||
def test_bridged_error_event_with_a_numeric_code_is_a_throttling_error(gateway: Gateway) -> None:
|
||||
|
|
@ -297,6 +289,14 @@ def test_bridged_error_event_with_a_numeric_code_is_a_throttling_error(gateway:
|
|||
assert_provider_typed_rate_limit(error_body(response))
|
||||
|
||||
|
||||
def test_bridged_openai_error_event_keeps_the_provider_code_and_message(gateway: Gateway) -> None:
|
||||
identity: Final = "resp_" + uuid.uuid4().hex
|
||||
stream: Final = (frame(created(identity)), frame(openai_error_event("rate_limit_exceeded", RATE_LIMIT_MESSAGE)))
|
||||
response, _ = _in_stream_error_status(gateway, stream)
|
||||
assert response.status_code == 429, response.text
|
||||
assert_provider_typed_rate_limit(error_body(response))
|
||||
|
||||
|
||||
def test_bridged_response_failed_without_an_error_event_is_a_throttling_error(gateway: Gateway) -> None:
|
||||
identity: Final = "resp_" + uuid.uuid4().hex
|
||||
stream: Final = (frame(created(identity)), frame(failed(identity, "rate_limit_exceeded", RATE_LIMIT_MESSAGE)))
|
||||
|
|
@ -323,8 +323,8 @@ def test_bridged_rate_limit_after_output_is_a_provider_typed_error_frame_behind_
|
|||
assert str(error["code"]) == "429", error
|
||||
assert error["type"] == "throttling_error", error
|
||||
message: Final = string_value(error["message"])
|
||||
assert message.startswith(_RATE_LIMIT_PREFIX) and RATE_LIMIT_MESSAGE in message, message
|
||||
assert _SENTINEL_PREFIX not in message, message
|
||||
assert message.startswith(RATE_LIMIT_PREFIX) and RATE_LIMIT_MESSAGE in message, message
|
||||
assert SENTINEL_PREFIX not in message, message
|
||||
|
||||
|
||||
def test_bridged_transport_drop_after_response_created_is_a_500_without_the_sentinel(gateway: Gateway) -> None:
|
||||
|
|
@ -346,7 +346,7 @@ def test_bridged_transport_drop_after_response_created_is_a_500_without_the_sent
|
|||
message: Final = string_value(error["message"])
|
||||
assert str(error["code"]) == "500", error
|
||||
assert "never sent" not in response.text
|
||||
assert _SENTINEL_PREFIX not in message, message
|
||||
assert SENTINEL_PREFIX not in message, message
|
||||
assert_failed_once(wire, response.headers["x-litellm-call-id"], model)
|
||||
|
||||
|
||||
|
|
@ -372,28 +372,18 @@ def test_plain_chat_http_rate_limit_is_a_throttling_error_on_both_legs(gateway:
|
|||
error: Final = object_value(raised.value.body)
|
||||
assert error["type"] == "throttling_error" and str(error["code"]) == "429", error
|
||||
message: Final = string_value(error["message"])
|
||||
assert message.startswith(_RATE_LIMIT_PREFIX) and "Rate limit reached" in message, message
|
||||
assert message.startswith(RATE_LIMIT_PREFIX) and "Rate limit reached" in message, message
|
||||
assert_failed_once(wire, raised.value.response.headers["x-litellm-call-id"], model)
|
||||
|
||||
|
||||
def sse_events(text: str) -> tuple[tuple[str, Mapping[str, JsonValue]], ...]:
|
||||
def parse(block: str) -> tuple[str, Mapping[str, JsonValue]]:
|
||||
lines: Final = block.splitlines()
|
||||
event: Final = next(line.removeprefix("event: ") for line in lines if line.startswith("event: "))
|
||||
data: Final = next(line.removeprefix("data: ") for line in lines if line.startswith("data: "))
|
||||
return event, JSON_OBJECT.validate_json(data)
|
||||
|
||||
return tuple(parse(block) for block in text.strip().split("\n\n") if "event: " in block)
|
||||
|
||||
|
||||
def assert_messages_errorframe(events: Sequence[tuple[str, Mapping[str, JsonValue]]]) -> None:
|
||||
assert events[0][0] == "message_start", events
|
||||
assert events[-1][0] == "error", events
|
||||
error: Final = object_value(events[-1][1]["error"])
|
||||
assert error["type"] == "rate_limit_error", error
|
||||
message: Final = string_value(error["message"])
|
||||
assert message.startswith(_RATE_LIMIT_PREFIX) and RATE_LIMIT_MESSAGE in message, message
|
||||
assert message.count(_RATE_LIMIT_PREFIX) == 1 and _SENTINEL_PREFIX not in message, message
|
||||
assert message.startswith(RATE_LIMIT_PREFIX) and RATE_LIMIT_MESSAGE in message, message
|
||||
assert message.count(RATE_LIMIT_PREFIX) == 1 and SENTINEL_PREFIX not in message, message
|
||||
|
||||
|
||||
def test_messages_over_the_bridged_stream_carry_the_provider_error_once_in_the_errorframe(gateway: Gateway) -> None:
|
||||
|
|
|
|||
|
|
@ -0,0 +1,407 @@
|
|||
import json
|
||||
import uuid
|
||||
from collections.abc import Mapping
|
||||
from dataclasses import dataclass
|
||||
from typing import Final
|
||||
|
||||
import anthropic
|
||||
import httpx
|
||||
import openai
|
||||
import pytest
|
||||
from integration._support.client import Gateway, Scenario, object_value, string_value
|
||||
from integration._support.responses_stream import (
|
||||
AZURE_TARGET,
|
||||
OPENAI_TARGET,
|
||||
RATE_LIMIT_MESSAGE,
|
||||
RATE_LIMIT_PREFIX,
|
||||
SENTINEL_PREFIX,
|
||||
assert_provider_typed_rate_limit,
|
||||
chat_body,
|
||||
chat_content,
|
||||
created,
|
||||
data_frames,
|
||||
delta,
|
||||
error_body,
|
||||
frame,
|
||||
function_tools,
|
||||
healthy_stream,
|
||||
messages_body,
|
||||
openai_error_event,
|
||||
openai_error_stream,
|
||||
serve,
|
||||
spend_row,
|
||||
sse_events,
|
||||
)
|
||||
from integration._support.wire import Wire, wire_server
|
||||
from openai.types.responses import ResponseFailedEvent, ResponseStreamEvent
|
||||
from pydantic import JsonValue
|
||||
|
||||
_AZURE_KEY: Final = "synthetic-azure-key"
|
||||
_QUOTA_MESSAGE: Final = "You exceeded your current quota, please check your plan and billing details."
|
||||
_SERVER_MESSAGE: Final = "The server had an error while processing your request. Sorry about that!"
|
||||
_ANTHROPIC_HEADERS: Final = {"anthropic-version": "2023-06-01"}
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class _Served:
|
||||
response: httpx.Response
|
||||
wire: Wire
|
||||
|
||||
|
||||
def _azure_model(scenario: Scenario, wire: Wire) -> str:
|
||||
return scenario.model(model="azure/gpt-6", api_base=wire.url, api_key=_AZURE_KEY)
|
||||
|
||||
|
||||
def _post(gateway: Gateway, path: str, body: Mapping[str, JsonValue], stream: tuple[bytes, ...]) -> _Served:
|
||||
with wire_server(serve(stream, AZURE_TARGET)) as wire, gateway.scenario() as scenario:
|
||||
model: Final = _azure_model(scenario, wire)
|
||||
response: Final = gateway.request("POST", path, {**body, "model": model}, headers=_ANTHROPIC_HEADERS)
|
||||
return _Served(response, wire)
|
||||
|
||||
|
||||
def _chat(gateway: Gateway, stream: tuple[bytes, ...]) -> tuple[httpx.Response, str]:
|
||||
marker: Final = uuid.uuid4().hex
|
||||
served: Final = _post(gateway, "/v1/chat/completions", chat_body("", marker), stream)
|
||||
_assert_one_attempt(served.wire, marker)
|
||||
return served.response, marker
|
||||
|
||||
|
||||
def _assert_one_attempt(wire: Wire, marker: str) -> None:
|
||||
received: Final = wire.drain()
|
||||
assert len(received) == 1 and marker.encode() in received[0].body, [request.target for request in received]
|
||||
|
||||
|
||||
def _assert_failure_logged(call_id: str, logged_code: str, provider_message: str) -> None:
|
||||
row: Final = spend_row(call_id)
|
||||
assert row["status"] == "failure", row
|
||||
error: Final = object_value(object_value(row["metadata"])["error_information"])
|
||||
assert str(error["error_code"]) == logged_code, error
|
||||
assert provider_message in string_value(error["error_message"]), error
|
||||
|
||||
|
||||
def _identity() -> str:
|
||||
return "resp_" + uuid.uuid4().hex
|
||||
|
||||
|
||||
def test_openai_sdk_chat_stream_top_level_rate_limit_is_a_throttling_error_with_the_provider_message(
|
||||
gateway: Gateway,
|
||||
) -> None:
|
||||
marker: Final = uuid.uuid4().hex
|
||||
stream: Final = openai_error_stream(_identity(), "rate_limit_exceeded", RATE_LIMIT_MESSAGE)
|
||||
with wire_server(serve(stream, AZURE_TARGET)) as wire, gateway.scenario() as scenario:
|
||||
model: Final = _azure_model(scenario, wire)
|
||||
client: Final = openai.OpenAI(base_url=f"{gateway.client.base_url}/v1", api_key=gateway.key, max_retries=0)
|
||||
with pytest.raises(openai.RateLimitError) as raised:
|
||||
client.chat.completions.create(
|
||||
model=model,
|
||||
messages=[{"role": "user", "content": marker}],
|
||||
tools=function_tools(),
|
||||
stream=True,
|
||||
extra_body={"num_retries": 0, "cache": {"no-cache": True}},
|
||||
)
|
||||
assert raised.value.status_code == 429
|
||||
assert_provider_typed_rate_limit(object_value(raised.value.body))
|
||||
_assert_one_attempt(wire, marker)
|
||||
_assert_failure_logged(raised.value.response.headers["x-litellm-call-id"], "429", RATE_LIMIT_MESSAGE)
|
||||
|
||||
|
||||
async def test_async_openai_sdk_codex_chat_stream_top_level_rate_limit_is_a_throttling_error(gateway: Gateway) -> None:
|
||||
marker: Final = uuid.uuid4().hex
|
||||
stream: Final = openai_error_stream(_identity(), "rate_limit_exceeded", RATE_LIMIT_MESSAGE)
|
||||
with wire_server(serve(stream, OPENAI_TARGET)) as wire, gateway.scenario() as scenario:
|
||||
model: Final = scenario.model(model="openai/gpt-5.3-codex", api_base=wire.url, api_key="synthetic-openai-key")
|
||||
client: Final = openai.AsyncOpenAI(base_url=f"{gateway.client.base_url}/v1", api_key=gateway.key, max_retries=0)
|
||||
with pytest.raises(openai.RateLimitError) as raised:
|
||||
await client.chat.completions.create(
|
||||
model=model,
|
||||
messages=[{"role": "user", "content": marker}],
|
||||
stream=True,
|
||||
extra_body={"num_retries": 0, "cache": {"no-cache": True}},
|
||||
)
|
||||
assert raised.value.status_code == 429
|
||||
assert_provider_typed_rate_limit(object_value(raised.value.body))
|
||||
_assert_one_attempt(wire, marker)
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("code", "message", "status"),
|
||||
[
|
||||
pytest.param("insufficient_quota", _QUOTA_MESSAGE, 429, id="insufficient_quota"),
|
||||
pytest.param("server_error", _SERVER_MESSAGE, 500, id="server_error"),
|
||||
pytest.param("invalid_prompt", "Invalid prompt: your prompt was flagged.", 400, id="invalid_prompt"),
|
||||
pytest.param(
|
||||
"context_length_exceeded",
|
||||
"Your input exceeds the context window of this model.",
|
||||
400,
|
||||
id="context_length_exceeded",
|
||||
),
|
||||
pytest.param("503", "The engine is currently overloaded.", 503, id="decimal_503"),
|
||||
pytest.param("brand_new_error_code", "Something new went wrong upstream.", 500, id="unknown_code"),
|
||||
pytest.param("", "An error with an empty code.", 500, id="empty_code"),
|
||||
],
|
||||
)
|
||||
def test_top_level_error_code_sets_the_chat_status_and_keeps_the_provider_message(
|
||||
gateway: Gateway, code: str, message: str, status: int
|
||||
) -> None:
|
||||
response, _ = _chat(gateway, openai_error_stream(_identity(), code, message))
|
||||
assert response.status_code == status, response.text
|
||||
error: Final = error_body(response)
|
||||
assert str(error["code"]) == str(status), error
|
||||
client_message: Final = string_value(error["message"])
|
||||
assert message in client_message and SENTINEL_PREFIX not in client_message, client_message
|
||||
logged_code: Final = code if status == 400 else str(status)
|
||||
_assert_failure_logged(response.headers["x-litellm-call-id"], logged_code, message)
|
||||
|
||||
|
||||
def test_top_level_rate_limit_after_text_is_a_throttling_error_frame_behind_the_text(gateway: Gateway) -> None:
|
||||
identity: Final = _identity()
|
||||
stream: Final = (
|
||||
frame(created(identity)),
|
||||
frame(delta(identity, "Hello")),
|
||||
frame(delta(identity, " there")),
|
||||
frame(openai_error_event("rate_limit_exceeded", RATE_LIMIT_MESSAGE)),
|
||||
)
|
||||
response, _ = _chat(gateway, stream)
|
||||
assert response.status_code == 200, response.text
|
||||
frames: Final = data_frames(response.text)
|
||||
assert chat_content(frames) == "Hello there", response.text
|
||||
assert_provider_typed_rate_limit(object_value(frames[-1]["error"]))
|
||||
|
||||
|
||||
def test_nested_error_object_wins_over_top_level_fields(gateway: Gateway) -> None:
|
||||
event: Final = {
|
||||
**openai_error_event("rate_limit_exceeded", "top-level says slow down"),
|
||||
"error": {"type": "server_error", "code": "server_error", "message": "nested says the server broke"},
|
||||
}
|
||||
response, _ = _chat(gateway, (frame(created(_identity())), frame(event)))
|
||||
assert response.status_code == 500, response.text
|
||||
message: Final = string_value(error_body(response)["message"])
|
||||
assert "nested says the server broke" in message and "top-level says" not in message, message
|
||||
|
||||
|
||||
def _raw_event(body: str) -> bytes:
|
||||
return f"event: error\ndata: {body}\n\n".encode()
|
||||
|
||||
|
||||
_FIVE_KB_MESSAGE: Final = "Rate limit reached for gpt-6. " + "x" * 5120
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("event", "provider_message"),
|
||||
[
|
||||
pytest.param(
|
||||
frame({**openai_error_event("rate_limit_exceeded", RATE_LIMIT_MESSAGE), "error": None}),
|
||||
RATE_LIMIT_MESSAGE,
|
||||
id="error_null",
|
||||
),
|
||||
pytest.param(
|
||||
frame({"type": "error", "sequence_number": 1, "code": 429, "message": RATE_LIMIT_MESSAGE, "param": None}),
|
||||
RATE_LIMIT_MESSAGE,
|
||||
id="integer_code",
|
||||
),
|
||||
pytest.param(frame(openai_error_event("rate_limit_exceeded", "")), "", id="empty_message"),
|
||||
pytest.param(
|
||||
frame({"type": "error", "sequence_number": 1, "code": "rate_limit_exceeded", "message": {"text": "x"}}),
|
||||
"",
|
||||
id="object_message",
|
||||
),
|
||||
pytest.param(frame(openai_error_event("rate_limit_exceeded", _FIVE_KB_MESSAGE)), _FIVE_KB_MESSAGE, id="5kb"),
|
||||
pytest.param(
|
||||
_raw_event(
|
||||
'{"type": "error", "sequence_number": 1, "code": "rate_limit_exceeded", '
|
||||
f'"code": "rate_limit_exceeded", "message": {json.dumps(RATE_LIMIT_MESSAGE)}, "param": null}}'
|
||||
),
|
||||
RATE_LIMIT_MESSAGE,
|
||||
id="duplicated_code",
|
||||
),
|
||||
],
|
||||
)
|
||||
def test_malformed_top_level_fields_still_answer_a_throttling_error(
|
||||
gateway: Gateway, event: bytes, provider_message: str
|
||||
) -> None:
|
||||
response, _ = _chat(gateway, (frame(created(_identity())), event))
|
||||
assert response.status_code == 429, response.text
|
||||
assert_provider_typed_rate_limit(error_body(response), provider_message)
|
||||
healthy: Final = _post(
|
||||
gateway, "/v1/chat/completions", chat_body("", "after"), healthy_stream(_identity(), "still serving")
|
||||
)
|
||||
assert chat_content(data_frames(healthy.response.text)) == "still serving", healthy.response.text
|
||||
|
||||
|
||||
def _cached_chat(gateway: Gateway, stream: tuple[bytes, ...]) -> tuple[tuple[httpx.Response, ...], int]:
|
||||
marker: Final = uuid.uuid4().hex
|
||||
body: Final = {key: value for key, value in chat_body("", marker).items() if key != "cache"}
|
||||
with wire_server(serve(stream, AZURE_TARGET)) as wire, gateway.scenario() as scenario:
|
||||
model: Final = _azure_model(scenario, wire)
|
||||
responses: Final = tuple(
|
||||
gateway.request("POST", "/v1/chat/completions", {**body, "model": model}) for _ in range(2)
|
||||
)
|
||||
return responses, len(wire.drain())
|
||||
|
||||
|
||||
def test_a_top_level_error_stream_is_never_served_from_the_response_cache(gateway: Gateway) -> None:
|
||||
responses, attempts = _cached_chat(
|
||||
gateway, openai_error_stream(_identity(), "rate_limit_exceeded", RATE_LIMIT_MESSAGE)
|
||||
)
|
||||
assert attempts == 2, attempts
|
||||
for response in responses:
|
||||
assert response.status_code == 429, response.text
|
||||
assert_provider_typed_rate_limit(error_body(response))
|
||||
|
||||
|
||||
def _responses_frames(gateway: Gateway, stream: tuple[bytes, ...]) -> tuple[httpx.Response, Mapping[str, JsonValue]]:
|
||||
marker: Final = uuid.uuid4().hex
|
||||
served: Final = _post(
|
||||
gateway, "/v1/responses", {"input": marker, "stream": True, "cache": {"no-cache": True}}, stream
|
||||
)
|
||||
_assert_one_attempt(served.wire, marker)
|
||||
assert served.response.status_code == 200, served.response.text
|
||||
assert served.response.text.rstrip().endswith("data: [DONE]"), served.response.text
|
||||
frames: Final = data_frames(served.response.text)
|
||||
assert frames[-1]["type"] == "response.failed", served.response.text
|
||||
return served.response, object_value(object_value(frames[-1]["response"])["error"])
|
||||
|
||||
|
||||
def _assert_rate_limit_failure(error: Mapping[str, JsonValue]) -> None:
|
||||
assert error["code"] == "rate_limit_exceeded", error
|
||||
assert RATE_LIMIT_MESSAGE in string_value(error["message"]), error
|
||||
|
||||
|
||||
def test_native_responses_stream_fails_with_the_provider_code_and_message(gateway: Gateway) -> None:
|
||||
identity: Final = _identity()
|
||||
response, error = _responses_frames(
|
||||
gateway, openai_error_stream(identity, "rate_limit_exceeded", RATE_LIMIT_MESSAGE)
|
||||
)
|
||||
assert [frame["type"] for frame in data_frames(response.text)] == ["response.created", "response.failed"]
|
||||
_assert_rate_limit_failure(error)
|
||||
_assert_failure_logged(response.headers["x-litellm-call-id"], "429", RATE_LIMIT_MESSAGE)
|
||||
|
||||
|
||||
def test_native_responses_stream_error_after_text_fails_with_the_provider_code(gateway: Gateway) -> None:
|
||||
identity: Final = _identity()
|
||||
stream: Final = (
|
||||
frame(created(identity)),
|
||||
frame(delta(identity, "partial")),
|
||||
frame(openai_error_event("rate_limit_exceeded", RATE_LIMIT_MESSAGE)),
|
||||
)
|
||||
response, error = _responses_frames(gateway, stream)
|
||||
kinds: Final = [frame["type"] for frame in data_frames(response.text)]
|
||||
assert kinds == ["response.created", "response.output_text.delta", "response.failed"], kinds
|
||||
_assert_rate_limit_failure(error)
|
||||
|
||||
|
||||
def _failed_error(events: list[ResponseStreamEvent]) -> tuple[str, str]:
|
||||
assert [event.type for event in events] == ["response.created", "response.failed"], events
|
||||
failed: Final = events[-1]
|
||||
assert isinstance(failed, ResponseFailedEvent) and failed.response.error is not None, failed
|
||||
return failed.response.error.code, failed.response.error.message
|
||||
|
||||
|
||||
def test_openai_sdk_responses_stream_fails_with_insufficient_quota(gateway: Gateway) -> None:
|
||||
marker: Final = uuid.uuid4().hex
|
||||
stream: Final = openai_error_stream(_identity(), "insufficient_quota", _QUOTA_MESSAGE)
|
||||
with wire_server(serve(stream, OPENAI_TARGET)) as wire, gateway.scenario() as scenario:
|
||||
model: Final = scenario.model(model="openai/gpt-5.3-codex", api_base=wire.url, api_key="synthetic-openai-key")
|
||||
client: Final = openai.OpenAI(base_url=f"{gateway.client.base_url}/v1", api_key=gateway.key, max_retries=0)
|
||||
events: Final = list(
|
||||
client.responses.create(model=model, input=marker, stream=True, extra_body={"cache": {"no-cache": True}})
|
||||
)
|
||||
code, message = _failed_error(events)
|
||||
assert code == "insufficient_quota" and _QUOTA_MESSAGE in message, events
|
||||
_assert_one_attempt(wire, marker)
|
||||
|
||||
|
||||
async def test_async_openai_sdk_responses_stream_fails_with_the_server_error_message(gateway: Gateway) -> None:
|
||||
marker: Final = uuid.uuid4().hex
|
||||
stream: Final = openai_error_stream(_identity(), "server_error", _SERVER_MESSAGE)
|
||||
with wire_server(serve(stream, AZURE_TARGET)) as wire, gateway.scenario() as scenario:
|
||||
model: Final = _azure_model(scenario, wire)
|
||||
client: Final = openai.AsyncOpenAI(base_url=f"{gateway.client.base_url}/v1", api_key=gateway.key, max_retries=0)
|
||||
response: Final = await client.responses.create(
|
||||
model=model, input=marker, stream=True, extra_body={"cache": {"no-cache": True}}
|
||||
)
|
||||
events: Final = [event async for event in response]
|
||||
code, message = _failed_error(events)
|
||||
assert code == "server_error" and _SERVER_MESSAGE in message, events
|
||||
_assert_one_attempt(wire, marker)
|
||||
|
||||
|
||||
def _assert_messages_error(error: Mapping[str, JsonValue], error_type: str, provider_message: str) -> None:
|
||||
assert error["type"] == error_type, error
|
||||
message: Final = string_value(error["message"])
|
||||
assert provider_message in message and SENTINEL_PREFIX not in message, message
|
||||
|
||||
|
||||
def test_messages_stream_top_level_rate_limit_is_a_rate_limit_errorframe(gateway: Gateway) -> None:
|
||||
marker: Final = uuid.uuid4().hex
|
||||
stream: Final = openai_error_stream(_identity(), "rate_limit_exceeded", RATE_LIMIT_MESSAGE)
|
||||
served: Final = _post(gateway, "/v1/messages", messages_body("", marker), stream)
|
||||
assert served.response.status_code == 200, served.response.text
|
||||
events: Final = sse_events(served.response.text)
|
||||
assert events[0][0] == "message_start" and events[-1][0] == "error", events
|
||||
error: Final = object_value(events[-1][1]["error"])
|
||||
_assert_messages_error(error, "rate_limit_error", RATE_LIMIT_MESSAGE)
|
||||
assert string_value(error["message"]).count(RATE_LIMIT_PREFIX) == 1, error
|
||||
_assert_one_attempt(served.wire, marker)
|
||||
|
||||
|
||||
def _anthropic_tools() -> list[anthropic.types.ToolParam]: # mutable-ok: the Anthropic SDK tools parameter is a list
|
||||
return [
|
||||
{
|
||||
"name": "get_weather",
|
||||
"description": "Weather for a city",
|
||||
"input_schema": {"type": "object", "properties": {"city": {"type": "string"}}},
|
||||
}
|
||||
]
|
||||
|
||||
|
||||
async def _drain_async_messages_stream(client: anthropic.AsyncAnthropic, model: str, marker: str) -> None:
|
||||
async with client.messages.stream(
|
||||
model=model,
|
||||
max_tokens=64,
|
||||
messages=[{"role": "user", "content": marker}],
|
||||
tools=_anthropic_tools(),
|
||||
extra_body={"num_retries": 0, "cache": {"no-cache": True}},
|
||||
) as events:
|
||||
async for _ in events:
|
||||
pass
|
||||
|
||||
|
||||
def _drain_messages_stream(client: anthropic.Anthropic, model: str, marker: str) -> None:
|
||||
with client.messages.stream(
|
||||
model=model,
|
||||
max_tokens=64,
|
||||
messages=[{"role": "user", "content": marker}],
|
||||
tools=_anthropic_tools(),
|
||||
extra_body={"num_retries": 0, "cache": {"no-cache": True}},
|
||||
) as events:
|
||||
for _ in events:
|
||||
pass
|
||||
|
||||
|
||||
async def test_async_anthropic_sdk_stream_raises_the_top_level_rate_limit(gateway: Gateway) -> None:
|
||||
marker: Final = uuid.uuid4().hex
|
||||
stream: Final = openai_error_stream(_identity(), "rate_limit_exceeded", RATE_LIMIT_MESSAGE)
|
||||
with wire_server(serve(stream, AZURE_TARGET)) as wire, gateway.scenario() as scenario:
|
||||
model: Final = _azure_model(scenario, wire)
|
||||
client: Final = anthropic.AsyncAnthropic(
|
||||
base_url=str(gateway.client.base_url), api_key=gateway.key, max_retries=0
|
||||
)
|
||||
with pytest.raises(anthropic.APIStatusError) as raised:
|
||||
await _drain_async_messages_stream(client, model, marker)
|
||||
_assert_messages_error(
|
||||
object_value(object_value(raised.value.body)["error"]), "rate_limit_error", RATE_LIMIT_MESSAGE
|
||||
)
|
||||
_assert_one_attempt(wire, marker)
|
||||
|
||||
|
||||
def test_sync_anthropic_sdk_stream_raises_the_top_level_server_error_message(gateway: Gateway) -> None:
|
||||
marker: Final = uuid.uuid4().hex
|
||||
stream: Final = openai_error_stream(_identity(), "server_error", _SERVER_MESSAGE)
|
||||
with wire_server(serve(stream, AZURE_TARGET)) as wire, gateway.scenario() as scenario:
|
||||
model: Final = _azure_model(scenario, wire)
|
||||
client: Final = anthropic.Anthropic(base_url=str(gateway.client.base_url), api_key=gateway.key, max_retries=0)
|
||||
with pytest.raises(anthropic.APIStatusError) as raised:
|
||||
_drain_messages_stream(client, model, marker)
|
||||
_assert_messages_error(object_value(object_value(raised.value.body)["error"]), "api_error", _SERVER_MESSAGE)
|
||||
_assert_one_attempt(wire, marker)
|
||||
|
|
@ -371,6 +371,7 @@ class TestChatGPTResponsesAPITransformation:
|
|||
"type": "error",
|
||||
"error": {"message": "ChatGPT upstream failed"},
|
||||
},
|
||||
{"type": "error", "code": "server_error", "message": "ChatGPT upstream failed"},
|
||||
],
|
||||
)
|
||||
def test_chatgpt_non_stream_sse_response_raises_openai_error(self, error_chunk):
|
||||
|
|
@ -392,5 +393,5 @@ class TestChatGPTResponsesAPITransformation:
|
|||
logging_obj=logging_obj,
|
||||
)
|
||||
|
||||
assert "ChatGPT upstream failed" in str(exc_info.value)
|
||||
assert exc_info.value.message == "ChatGPT upstream failed"
|
||||
assert exc_info.value.status_code == 502
|
||||
|
|
|
|||
|
|
@ -3119,7 +3119,7 @@ class TestNativeWebSocketEncryptedContentAffinity:
|
|||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize(
|
||||
"failure_frame, expected_status",
|
||||
"failure_frame, expected_status, expected_message",
|
||||
[
|
||||
(
|
||||
{
|
||||
|
|
@ -3131,6 +3131,7 @@ class TestNativeWebSocketEncryptedContentAffinity:
|
|||
},
|
||||
},
|
||||
400,
|
||||
"The encrypted content for item rs_1 could not be verified.",
|
||||
),
|
||||
(
|
||||
{
|
||||
|
|
@ -3142,11 +3143,17 @@ class TestNativeWebSocketEncryptedContentAffinity:
|
|||
},
|
||||
},
|
||||
500,
|
||||
"upstream blew up",
|
||||
),
|
||||
(
|
||||
{"type": "error", "code": "rate_limit_exceeded", "message": "Rate limit reached for gpt-5.6"},
|
||||
429,
|
||||
"Rate limit reached for gpt-5.6",
|
||||
),
|
||||
],
|
||||
)
|
||||
async def test_backend_to_client_books_failure_frames_as_failures(
|
||||
self, failure_frame: dict[str, object], expected_status: int
|
||||
self, failure_frame: dict[str, object], expected_status: int, expected_message: str
|
||||
):
|
||||
import asyncio
|
||||
from unittest.mock import AsyncMock
|
||||
|
|
@ -3183,7 +3190,7 @@ class TestNativeWebSocketEncryptedContentAffinity:
|
|||
logging_obj.dispatch_failure_handlers.assert_awaited_once()
|
||||
exception = logging_obj.dispatch_failure_handlers.await_args[0][0]
|
||||
assert exception.status_code == expected_status
|
||||
assert failure_frame.get("error", failure_frame.get("response", {}).get("error"))["message"] in str(exception)
|
||||
assert expected_message in str(exception)
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_backend_to_client_bills_completed_turns_before_a_failure(self):
|
||||
|
|
|
|||
|
|
@ -587,3 +587,24 @@ def test_maybe_raise_for_error_event_maps_vector_store_timeout_to_retriable_504(
|
|||
with pytest.raises(MidStreamFallbackError) as exc_info:
|
||||
iterator._maybe_raise_for_error_event(chunk)
|
||||
assert exc_info.value.status_code == 504
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("code", "status_code", "exception_type"),
|
||||
[
|
||||
("rate_limit_exceeded", 429, litellm.RateLimitError),
|
||||
("server_error", 500, litellm.InternalServerError),
|
||||
(429, 429, litellm.RateLimitError),
|
||||
],
|
||||
)
|
||||
def test_openai_top_level_error_event_keeps_the_provider_code_and_message(
|
||||
code: str | int, status_code: int, exception_type: type[Exception]
|
||||
):
|
||||
chunk = ErrorEvent(
|
||||
type=ResponsesAPIStreamEvents.ERROR, sequence_number=3, code=code, message=f"upstream said {code}"
|
||||
)
|
||||
with pytest.raises(MidStreamFallbackError) as exc_info:
|
||||
_make_iterator()._maybe_raise_for_error_event(chunk)
|
||||
assert exc_info.value.status_code == status_code
|
||||
assert type(exc_info.value.original_exception) is exception_type
|
||||
assert f"upstream said {code}" in str(exc_info.value.original_exception)
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue