scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-10-03 02:22:24 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 80

test(e2e): move live-provider legacy tests into tests/e2e (#44120)

Browse source 
* test(e2e): move live-provider legacy tests into tests/e2e

Port legacy tests that exercise real providers into the tests/e2e suites that own them, using the harness (/model/new plus deferred cleanup) and asserting on what the caller receives. Delete legacy tests already covered at equal or stronger strength by e2e, integration or unit tests, and drop the now empty ocr_testing CircleCI job

* test(e2e): address review on the live-provider test move

Assert the SSE error frame a client actually receives when a post_call guardrail blocks a stream, and require a tool call for every requested city before checking the answer. Restore the OCR matrix and its CircleCI job, the Claude Agent SDK streaming test, and test_async_create_batch, since their SDK-level and callback assertions have no equivalent in tests/e2e

* test(e2e): accept both guardrail block shapes on a blocked stream

A post_call block before the first chunk reaches the client as HTTP 400 with either a JSON error body or a single SSE error frame, depending on whether the block surfaced as an exception or an error chunk. Assert the policy message is present and the blocked output is absent in both

* test(realtime): restore direct SDK realtime tests against OpenAI

The e2e realtime tests go through the proxy and the remaining SDK tests either mock the upstream or assert less, so keep the direct litellm._arealtime tests with and without intent, and TestOpenAIRealtime::test_realtime_connection, in place

* test: make realtime and Nova stream checks deterministic

The direct SDK realtime tests now fail on a refused connection instead of skipping. The with-intent test asserts OpenAI rejects the exact intent value sent, which only happens when the intent is forwarded. The Nova /v1/messages stream test asserts stream structure, stop reason and usage instead of model wording

* test(realtime): own intent forwarding with a unit test instead of a live rejection

Assert litellm._arealtime passes the intent query param into the OpenAI realtime websocket URL, which is the behavior LiteLLM owns, and drop the live test that depended on OpenAI's rejection wording
This commit is contained in:
yuneng-jiang 2026-10-02 00:02:18 -07:00 • committed by GitHub
parent e32b25817f
commit 7d50a31eb5
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
45 changed files with 1016 additions and 2060 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

35
tests/audio_tests/test_audio_speech.py
View file

@ -667,38 +667,3 @@ async def test_aws_polly_tts_with_ssml():
assert request_body["VoiceId"] == "Joanna"
@pytest.mark.asyncio
async def test_aws_polly_tts_real_api():
"""
Test AWS Polly TTS with real API request.
Requires AWS credentials to be configured.
"""
speech_file_path = Path(__file__).parent / "aws_polly_speech_generative.mp3"
response = await litellm.aspeech(
model="aws_polly/generative",
voice="Joanna",
input="Hello, this is a test of AWS Polly text to speech integration with LiteLLM.",
aws_region_name="us-east-1",
)
from litellm.types.llms.openai import HttpxBinaryResponseContent
assert isinstance(response, HttpxBinaryResponseContent)
binary_content = response.content
assert len(binary_content) > 0
# MP3 files start with ID3 tag or MPEG sync word
assert (
binary_content[:3] == b"ID3"
or binary_content[:2] == b"\xff\xfb"
or binary_content[:2] == b"\xff\xf3"
)
response.stream_to_file(speech_file_path)
assert speech_file_path.exists()
assert speech_file_path.stat().st_size > 0
print(f"AWS Polly TTS audio saved to: {speech_file_path}")

16
tests/audio_tests/test_whisper.py
View file

@ -61,22 +61,6 @@ async def _run_transcription(
assert transcript.text is not None
@pytest.mark.parametrize(
"response_format, timestamp_granularities",
[("json", None), ("vtt", None), ("verbose_json", ["word"])],
)
@pytest.mark.asyncio
@pytest.mark.flaky(retries=3, delay=1)
async def test_transcription_openai_whisper(response_format, timestamp_granularities):
await _run_transcription(
model="whisper-1",
api_key=None,
api_base=None,
response_format=response_format,
timestamp_granularities=timestamp_granularities,
)
@pytest.mark.parametrize(
"response_format, timestamp_granularities",
[("json", None), ("vtt", None), ("verbose_json", ["word"])],

78
tests/batches_tests/test_openai_batches_and_files.py
View file

@ -118,84 +118,6 @@ async def cancel_batch_unless_already_terminal(batch_id: str, provider: str) ->
print("cancel_batch_response=", cancel_batch_response)
@pytest.mark.parametrize("provider", ["openai"]) # , "azure"
@pytest.mark.asyncio
@skip_if_no_openai_network
async def test_create_batch(provider, tmp_path):
"""
1. Create File for Batch completion
2. Create Batch Request
3. Retrieve the specific batch
"""
if provider == "azure":
# Don't have anymore Azure Quota
return
file_name = "openai_batch_completions.jsonl"
_current_dir = os.path.dirname(os.path.abspath(__file__))
file_path = os.path.join(_current_dir, file_name)
with open(file_path, "rb") as batch_file:
file_obj = await litellm.acreate_file(
file=batch_file,
purpose="batch",
custom_llm_provider=provider,
)
print("Response from creating file=", file_obj)
batch_input_file_id = file_obj.id
assert (
batch_input_file_id is not None
), "Failed to create file, expected a non null file_id but got {batch_input_file_id}"
await asyncio.sleep(1)
create_batch_response = await litellm.acreate_batch(
completion_window="24h",
endpoint="/v1/chat/completions",
input_file_id=batch_input_file_id,
custom_llm_provider=provider,
metadata={"key1": "value1", "key2": "value2"},
)
print("response from litellm.create_batch=", create_batch_response)
await asyncio.sleep(6)
assert (
create_batch_response.id is not None
), f"Failed to create batch, expected a non null batch_id but got {create_batch_response.id}"
assert (
create_batch_response.endpoint == "/v1/chat/completions"
or create_batch_response.endpoint == "/chat/completions"
), f"Failed to create batch, expected endpoint to be /v1/chat/completions but got {create_batch_response.endpoint}"
assert (
create_batch_response.input_file_id == batch_input_file_id
), f"Failed to create batch, expected input_file_id to be {batch_input_file_id} but got {create_batch_response.input_file_id}"
retrieved_batch = await litellm.aretrieve_batch(
batch_id=create_batch_response.id, custom_llm_provider=provider
)
print("retrieved batch=", retrieved_batch)
# just assert that we retrieved a non None batch
assert retrieved_batch.id == create_batch_response.id
# list all batches
list_batches = await litellm.alist_batches(custom_llm_provider=provider, limit=2)
print("list_batches=", list_batches)
file_content = await litellm.afile_content(
file_id=batch_input_file_id, custom_llm_provider=provider
)
result = file_content.content
result_file_path = tmp_path / "batch_job_results_furniture.jsonl"
result_file_path.write_bytes(result)
await cancel_batch_unless_already_terminal(batch_id=create_batch_response.id, provider=provider)
pass
class TestCustomLogger(CustomLogger):
def __init__(self):
super().__init__()

2
tests/e2e/coverage_registry/guardrail.yaml
View file

@ -6,7 +6,7 @@
- {id: guardrail.presidio.post_call.spend_log_stores_masked_output, module: guardrail, tier: P0, hook_point: post_call, assertions: [masks], exercised_on: [chat_completions, chat_completions_stream, messages, anthropic_messages_stream, responses], source: "guardrail_hooks/presidio.py", fail_before_fix: proven, rationale: "When an output guardrail masks the response, the spend log stores the masked text the caller received rather than the raw model output, on every endpoint and both stream modes (LIT-8325)"}
- {id: guardrail.presidio.logging_only.masks, module: guardrail, tier: P0, hook_point: logging_only, assertions: [masks], exercised_on: [chat_completions, messages], source: "guardrail_hooks/presidio.py", rationale: "Redact in logs without blocking"}
- {id: guardrail.presidio.pre_call.logs_masked_entities, module: guardrail, tier: P0, hook_point: pre_call, assertions: [logs_masked_entities], exercised_on: [chat_completions], source: "guardrail_hooks/presidio.py", rationale: "A masking run must record itself on the spend log: the dashboard's guardrail panel renders the masked-entity counts and per-entity scores straight off metadata.guardrail_information, so a run that masks but records nothing leaves an operator unable to audit it"}
- {id: guardrail.bedrock.pre_call.blocks, module: guardrail, tier: P0, hook_point: pre_call, assertions: [blocks], exercised_on: [chat_completions], source: "guardrail_hooks/bedrock_guardrails.py", rationale: "AWS content guardrail blocks harmful input"}
- {id: guardrail.bedrock.pre_call.blocks, module: guardrail, tier: P0, hook_point: pre_call, assertions: [blocks], exercised_on: [chat_completions, messages, responses], source: "guardrail_hooks/bedrock_guardrails.py", rationale: "AWS content guardrail blocks harmful input"}
- {id: guardrail.litellm_content_filter.pre_call.blocks, module: guardrail, tier: P0, hook_point: pre_call, assertions: [blocks], exercised_on: [chat_completions], source: "test_team_disable_global_guardrail_e2e.py", rationale: "Local content-filter default-on blocks banned keyword pre-call"}
- {id: guardrail.litellm_content_filter.pre_call.blocks_video, module: guardrail, tier: P0, hook_point: pre_call, assertions: [blocks], exercised_on: [videos], source: "test_key_guardrail_video_e2e.py", fail_before_fix: proven, rationale: "A content-filter guardrail attached to a key (metadata.guardrails) blocks a banned prompt on POST /v1/videos before the provider is called; before the fix the route's call type was unknown to the unified guardrail hook and the prompt went to the provider unscanned (LIT-6685)"}
- {id: guardrail.litellm_content_filter.pre_call.allows, module: guardrail, tier: P0, hook_point: pre_call, assertions: [allows], exercised_on: [chat_completions], source: "test_team_disable_global_guardrail_e2e.py", rationale: "Team disable_global_guardrails bypasses default-on content filter"}

93
tests/e2e/guardrails/test_bedrock_guardrail_e2e.py
View file

@ -21,11 +21,12 @@ from typing import Final
import pytest
from e2e_config import unique_marker
from e2e_http import UnknownApiError
from e2e_http import StreamingResponse, UnknownApiError
from guardrails_client import (
BedrockGuardrailParamsBody,
GuardrailsClient,
poll_until_blocked,
poll_until_blocked_stream,
)
from lifecycle import ResourceManager
from pydantic import JsonValue, TypeAdapter
@ -135,3 +136,93 @@ class TestBedrockGuardrail:
)
case _:
pytest.fail(f"bedrock post_call guardrail did not block denied model output; got {result}")
@pytest.mark.covers("guardrail.bedrock.pre_call.blocks", exercised_on=["messages"])
def test_bedrock_pre_call_blocks_on_messages(
self, client: GuardrailsClient, resources: ResourceManager, scoped_key: str
) -> None:
name = _register_pre_call(client, resources, "e2e-bedrock-messages")
result = poll_until_blocked_stream(
lambda: client.messages_raw(scoped_key, MODEL, BLOCKED_PROMPT, guardrails=[name])
)
_assert_policy_block(result, "/v1/messages")
@pytest.mark.covers("guardrail.bedrock.pre_call.blocks", exercised_on=["responses"])
def test_bedrock_pre_call_blocks_on_responses(
self, client: GuardrailsClient, resources: ResourceManager, scoped_key: str
) -> None:
name = _register_pre_call(client, resources, "e2e-bedrock-responses")
result = poll_until_blocked_stream(
lambda: client.responses(scoped_key, MODEL, BLOCKED_PROMPT, guardrails=[name])
)
_assert_policy_block(result, "/v1/responses")
@pytest.mark.covers("guardrail.bedrock.post_call.blocks", exercised_on=["chat_completions"])
def test_bedrock_post_call_blocks_denied_streamed_output_and_passes_clean_streams(
self, client: GuardrailsClient, resources: ResourceManager, scoped_key: str
) -> None:
blocked_word = os.environ.get("BEDROCK_GUARDRAIL_BLOCKED_WORD", "FORBIDDENWORD")
name = f"e2e-bedrock-post-stream-{unique_marker()}"
guardrail_id = client.register(
name,
BedrockGuardrailParamsBody(
mode="post_call",
default_on=False,
guardrailIdentifier=os.environ["BEDROCK_GUARDRAIL_IDENTIFIER"],
guardrailVersion=os.environ["BEDROCK_GUARDRAIL_VERSION"],
),
)
resources.defer(lambda: client.delete_guardrail(guardrail_id))
prompt = f"Reply with exactly this one word and nothing else: {blocked_word}"
blocked = poll_until_blocked_stream(
lambda: client.chat_stream_raw(scoped_key, MODEL, prompt, guardrails=[name], max_tokens=128)
)
_assert_policy_block(blocked, "streamed /chat/completions")
error = _blocked_stream_error(blocked)
assert isinstance(error, dict) and set(error) == {"error"}, (
f"a blocked stream must return only an error, not model content: {blocked.body[:400]}"
)
assert "violated guardrail policy" in json.dumps(error["error"]).lower(), (
f"the error must name the guardrail verdict; got: {blocked.body[:400]}"
)
assert blocked_word not in json.dumps(_without_assessments(error)), (
f"the blocked model output must not leak into the error; got: {blocked.body[:400]}"
)
clean = client.chat_stream_raw(
scoped_key, MODEL, "Reply with exactly this one word and nothing else: hello", guardrails=[name]
)
assert clean.ok and clean.is_streaming, f"a clean output must stream through the guardrail: {clean.body[:400]}"
assert clean.stream_events and clean.stream_error is None, f"clean stream carried no content: {clean!r}"
def _register_pre_call(client: GuardrailsClient, resources: ResourceManager, prefix: str) -> str:
name = f"{prefix}-{unique_marker()}"
guardrail_id = client.create_bedrock_guardrail(
name,
identifier=os.environ["BEDROCK_GUARDRAIL_IDENTIFIER"],
version=os.environ["BEDROCK_GUARDRAIL_VERSION"],
)
resources.defer(lambda: client.delete_guardrail(guardrail_id))
return name
def _blocked_stream_error(result: StreamingResponse) -> JsonValue:
if not result.is_streaming:
return _JSON.validate_json(result.body)
payloads = (line.removeprefix("data:").strip() for line in result.body.splitlines() if line.startswith("data:"))
frames = tuple(payload for payload in payloads if payload != "[DONE]")
assert len(frames) == 1, f"a blocked SSE response must carry exactly one error frame, got: {result.body[:400]}"
return _JSON.validate_json(frames[0])
def _assert_policy_block(result: StreamingResponse, surface: str) -> None:
assert result.status_code == 400, (
f"{surface}: a Bedrock policy block must be HTTP 400, got {result.status_code}: {result.body[:400]}"
)
assert "violated guardrail policy" in result.body.lower(), (
f"{surface}: block body must name the guardrail verdict; got: {result.body[:400]}"
)

24
tests/e2e/llm_translation/structured_output.py Normal file
View file

@ -0,0 +1,24 @@
from __future__ import annotations
from typing import Final
from pydantic import TypeAdapter
SENTIMENT_PROMPT: Final = "Classify the sentiment of this review: 'The battery died after two days.'"
SENTIMENT_LABELS: Final = frozenset({"positive", "negative", "neutral"})
SENTIMENT_OUTPUT_FORMAT: Final[dict[str, object]] = {
"type": "json_schema",
"schema": {
"type": "object",
"properties": {"sentiment": {"type": "string", "enum": sorted(SENTIMENT_LABELS)}},
"required": ["sentiment"],
"additionalProperties": False,
},
}
_SENTIMENT_JSON: Final = TypeAdapter(dict[str, str])
def assert_sentiment_json(text: str) -> None:
parsed = _SENTIMENT_JSON.validate_json(text)
assert set(parsed) == {"sentiment"}, f"output_format schema not enforced, extra or missing keys: {parsed}"
assert parsed["sentiment"] in SENTIMENT_LABELS, f"sentiment outside the schema enum: {parsed}"

30
tests/e2e/llm_translation/test_audio_speech_e2e.py
View file

@ -139,3 +139,33 @@ class TestAudioSpeech:
json=_OptionalSpeechBody(model=model, input="", voice="alloy"),
)
assert_client_error(result, "speech empty input")
MP3_PREFIXES = (b"ID3", b"\xff\xfb", b"\xff\xf3", b"\xff\xf2")
class TestAwsPollySpeech:
def test_polly_generative_voice_returns_mp3(
self, proxy: ProxyClient, resources: ResourceManager, sdk: SdkClients
) -> None:
model = f"e2e-speech-polly-{unique_marker()}"
model_id = proxy.create_model(
model,
LiteLLMParamsBody(
model="aws_polly/generative",
aws_access_key_id="os.environ/AWS_ACCESS_KEY_ID",
aws_secret_access_key="os.environ/AWS_SECRET_ACCESS_KEY",
aws_region_name="os.environ/AWS_REGION",
),
)
resources.defer(lambda: proxy.delete_model(model_id))
client = sdk.openai(resources.key())
response = client.audio.speech.with_raw_response.create(
model=model, voice="alloy", input="Hello from the gateway.", response_format="mp3"
)
content_type = response_header(response.headers, "content-type")
assert "audio" in (content_type or ""), f"polly speech content-type is not audio: {content_type!r}"
assert response.content.startswith(MP3_PREFIXES), (
f"polly speech body is not MP3 audio: {response.content[:16]!r}"
)

61
tests/e2e/llm_translation/test_audio_transcriptions_e2e.py
View file

@ -17,11 +17,11 @@ from typing import Final
import pytest
from e2e_config import unique_marker
from e2e_http import UnknownApiError
from e2e_http import UnknownApiError, unwrap
from lifecycle import ResourceManager
from models import LiteLLMParamsBody
from proxy_client import ProxyClient
from pydantic import BaseModel
from pydantic import BaseModel, Field
from sdk_clients import SdkClients
pytestmark = pytest.mark.e2e
@ -119,3 +119,60 @@ class TestAudioTranscriptions:
)
case other:
pytest.fail(f"missing model expected a model-specific 400, got {other!r}")
class _WhisperForm(BaseModel):
model: str
response_format: str
timestamp_granularities: str | None = Field(default=None, serialization_alias="timestamp_granularities[]")
class _TranscriptWord(BaseModel):
word: str
start: float
end: float
class _VerboseTranscription(BaseModel):
text: str
words: list[_TranscriptWord] = []
class TestWhisperTranscriptionFormats:
def _upload[R: BaseModel](
self, proxy: ProxyClient, resources: ResourceManager, form: _WhisperForm, response_type: type[R]
) -> R:
model_id = proxy.create_model(
form.model, LiteLLMParamsBody(model="openai/whisper-1", api_key="os.environ/OPENAI_API_KEY")
)
resources.defer(lambda: proxy.delete_model(model_id))
return unwrap(
proxy.transport.upload(
"/v1/audio/transcriptions",
headers=proxy.transport.bearer(resources.key()),
form=form,
filename=WEATHER_WAV.name,
content=WEATHER_WAV.read_bytes(),
file_content_type="audio/wav",
response_type=response_type,
)
)
def test_vtt_format_returns_webvtt_transcript(self, proxy: ProxyClient, resources: ResourceManager) -> None:
form = _WhisperForm(model=f"e2e-whisper-vtt-{unique_marker()}", response_format="vtt")
transcript = self._upload(proxy, resources, form, _TranscriptionResult)
assert transcript.text.lstrip().startswith("WEBVTT"), f"vtt transcript is not WebVTT: {transcript.text[:200]!r}"
assert "weather" in transcript.text.lower(), f"vtt transcript lost the spoken words: {transcript.text!r}"
def test_verbose_json_returns_word_timestamps(self, proxy: ProxyClient, resources: ResourceManager) -> None:
form = _WhisperForm(
model=f"e2e-whisper-verbose-{unique_marker()}",
response_format="verbose_json",
timestamp_granularities="word",
)
transcript = self._upload(proxy, resources, form, _VerboseTranscription)
assert "weather" in transcript.text.lower(), f"verbose transcript lost the spoken words: {transcript.text!r}"
assert transcript.words, f"word timestamps were requested but none came back: {transcript!r}"
assert all(word.start <= word.end for word in transcript.words), (
f"word timings out of order: {transcript.words}"
)

94
tests/e2e/llm_translation/test_chat_completions_regression_e2e.py
View file

@ -52,6 +52,14 @@ AZURE_FOUNDRY_BACKEND: Final = "azure_ai/claude-haiku-4-5"
OPENAI_BACKEND = "openai/gpt-5.6"
ANTHROPIC_BACKEND = "anthropic/claude-haiku-4-5-20251001"
BEDROCK_CONVERSE_BACKEND = "bedrock/us.anthropic.claude-haiku-4-5-20251001-v1:0"
BEDROCK_NOVA_BACKEND: Final = "bedrock/us.amazon.nova-2-lite-v1:0"
VERTEX_PARTNER_BACKENDS: Final = ("vertex_ai/mistral-small-2503", "vertex_ai/openai/gpt-oss-120b-maas")
PDF_DOCUMENT_URL: Final = (
"https://cdn.jsdelivr.net/gh/BerriAI/litellm"
"@d769e81c90d453240c61fc572cdb27fae06a89d0"
"/tests/llm_translation/fixtures/dummy.pdf"
)
PDF_DOCUMENT_TEXT: Final = "test pdf file"
class _StreamToolCallFunction(BaseModel):
@ -982,6 +990,92 @@ class TestBedrockConverseChatCompletions:
response = unwrap(client.proxy.chat(key, ChatBody(model=model, messages=_vision_messages(), max_tokens=32)))
_assert_describes_cat(response)
def test_bedrock_converse_reads_a_pdf_sent_by_url(
self, client: PassthroughClient, resources: ResourceManager
) -> None:
model = f"e2e-bedrock-document-{unique_marker()}"
model_id = client.proxy.create_model(
model, _bedrock_params().model_copy(update={"model": BEDROCK_NOVA_BACKEND})
)
resources.defer(lambda: client.proxy.delete_model(model_id))
key = resources.key()
response = unwrap(
client.proxy.chat(
key,
ChatBody(
model=model,
messages=[
ChatMessage(
role="user",
content=[
TextContentPart(text="What title text is in this document? Reply with it only."),
ImageContentPart(image_url=ImageUrl(url=PDF_DOCUMENT_URL)),
],
)
],
max_tokens=64,
),
)
)
message = response.choices[0].message if response.choices else None
content = (message.content if message else "") or ""
assert PDF_DOCUMENT_TEXT in content.lower(), f"model did not read the PDF document block: {response}"
class _PartnerDelta(BaseModel):
role: str | None = None
content: str | None = None
class _PartnerChoice(BaseModel):
delta: _PartnerDelta = _PartnerDelta()
finish_reason: str | None = None
class _PartnerChunk(BaseModel):
choices: list[_PartnerChoice] = []
class TestVertexPartnerChatCompletions:
@pytest.mark.parametrize("backend", VERTEX_PARTNER_BACKENDS)
def test_vertex_partner_model_streams_openai_shaped_chunks(
self, client: PassthroughClient, resources: ResourceManager, backend: str
) -> None:
model = f"e2e-vertex-partner-{unique_marker()}"
model_id = client.proxy.create_model(
model,
LiteLLMParamsBody(
model=backend, vertex_project="os.environ/VERTEXAI_PROJECT", vertex_location="us-central1"
),
)
resources.defer(lambda: client.proxy.delete_model(model_id))
key = resources.key()
result = client.proxy.chat_stream(
key,
ChatBody(
model=model,
messages=[
ChatMessage(role="user", content=f"Count from 1 to 5, one number per line. {unique_marker()}")
],
max_tokens=256,
stream=True,
),
)
assert result.ok and result.is_streaming, f"stream was not established: {result}"
assert result.stream_error is None, f"stream carried an error event: {result.stream_error}"
assert result.stream_done, "stream must terminate with [DONE]"
chunks = tuple(_PartnerChunk.model_validate_json(event) for event in result.stream_events)
choices = tuple(choice for chunk in chunks for choice in chunk.choices)
assert choices and choices[0].delta.role == "assistant", (
f"first chunk must carry the assistant role: {chunks[:2]}"
)
terminal = tuple(index for index, choice in enumerate(choices) if choice.finish_reason is not None)
assert len(terminal) == 1, f"expected exactly one terminal choice: {[c.finish_reason for c in choices]}"
text = "".join(choice.delta.content or "" for choice in choices[: terminal[0] + 1])
assert "5" in text, f"streamed text lost the requested content: {text!r}"
class TestAnthropicChatCompletions:
"""Anthropic via the OpenAI-compatible /chat/completions path, the translation

160
tests/e2e/llm_translation/test_chat_tool_round_trip_e2e.py Normal file
View file

@ -0,0 +1,160 @@
from __future__ import annotations
from types import MappingProxyType
from typing import Final
import pytest
from e2e_config import unique_marker
from e2e_http import unwrap
from lifecycle import ResourceManager
from models import (
ChatAssistantTurn,
ChatBody,
ChatMessage,
ChatTool,
ChatToolFunction,
ChatToolResultTurn,
LiteLLMParamsBody,
OutMessage,
ThinkingParam,
ToolCall,
)
from passthrough_client import PassthroughClient
from pydantic import BaseModel
pytestmark = pytest.mark.e2e
GEMINI_BACKEND: Final = "gemini/gemini-3.5-flash-lite"
MISTRAL_BACKEND: Final = "mistral/mistral-medium-3.5"
ANTHROPIC_BACKEND: Final = "anthropic/claude-haiku-4-5"
BEDROCK_CONVERSE_BACKEND: Final = "bedrock/converse/us.anthropic.claude-sonnet-5-5"
PROMPT: Final = "What is the weather in Paris and in Tokyo? Use the get_weather tool for each city."
CITY_TEMPERATURES: Final = MappingProxyType({"paris": "22", "tokyo": "31"})
THINKING: Final = ThinkingParam(type="enabled", budget_tokens=1024)
WEATHER_TOOL: Final = ChatTool(
function=ChatToolFunction(
name="get_weather",
description="Get the current weather for a city",
parameters={
"type": "object",
"properties": {"location": {"type": "string"}},
"required": ["location"],
},
)
)
class _WeatherArgs(BaseModel):
location: str
def _api_key_params(backend: str, env: str) -> LiteLLMParamsBody:
return LiteLLMParamsBody(model=backend, api_key=f"os.environ/{env}")
def _bedrock_params() -> LiteLLMParamsBody:
return LiteLLMParamsBody(
model=BEDROCK_CONVERSE_BACKEND,
aws_access_key_id="os.environ/AWS_ACCESS_KEY_ID",
aws_secret_access_key="os.environ/AWS_SECRET_ACCESS_KEY",
aws_region_name="os.environ/AWS_REGION",
)
def _register(client: PassthroughClient, resources: ResourceManager, params: LiteLLMParamsBody) -> tuple[str, str]:
model = f"e2e-chat-tool-loop-{unique_marker()}"
model_id = client.proxy.create_model(model, params)
resources.defer(lambda: client.proxy.delete_model(model_id))
return model, resources.key()
def _choice(client: PassthroughClient, key: str, body: ChatBody) -> tuple[OutMessage, str | None]:
response = unwrap(client.proxy.chat(key, body))
choice = response.choices[0] if response.choices else None
assert choice is not None and choice.message is not None, f"chat returned no message: {response}"
return choice.message, choice.finish_reason
def _city_for(call: ToolCall) -> str:
location = _WeatherArgs.model_validate_json(call.function.arguments or "").location.lower()
city = next((city for city in CITY_TEMPERATURES if city in location), None)
assert city is not None, f"get_weather called for a city the prompt never named: {location!r}"
return city
def _assert_tool_results_reach_the_model(
client: PassthroughClient, key: str, model: str, *, thinking: ThinkingParam | None, tool_choice: str | None
) -> None:
first, finish_reason = _choice(
client,
key,
ChatBody(
model=model,
messages=[ChatMessage(role="user", content=PROMPT)],
tools=[WEATHER_TOOL],
tool_choice=tool_choice,
thinking=thinking,
max_tokens=2048,
),
)
calls = tuple(call for call in first.tool_calls or () if call.function.name == "get_weather")
assert calls and all(call.id for call in calls), f"model returned no addressable get_weather call: {first}"
assert finish_reason == "tool_calls", f"a tool-calling turn must finish with tool_calls, got {finish_reason!r}"
if thinking is not None:
assert first.thinking_blocks, f"thinking was enabled but no thinking blocks came back: {first}"
cities = tuple(_city_for(call) for call in calls)
assert set(cities) == set(CITY_TEMPERATURES), (
f"expected a get_weather call for every city {sorted(CITY_TEMPERATURES)}, got calls for {cities}"
)
temperatures = tuple(CITY_TEMPERATURES[city] for city in cities)
answer, _ = _choice(
client,
key,
ChatBody(
model=model,
messages=[
ChatMessage(role="user", content=PROMPT),
ChatAssistantTurn(
content=first.content, thinking_blocks=first.thinking_blocks, tool_calls=first.tool_calls
),
*(
ChatToolResultTurn(tool_call_id=call.id or "", content=f"{temperature} degrees C and sunny")
for call, temperature in zip(calls, temperatures)
),
],
tools=[WEATHER_TOOL],
thinking=thinking,
max_tokens=2048,
),
)
content = answer.content or ""
assert all(temperature in content for temperature in temperatures), (
f"the answer ignored the tool results {temperatures}: {content!r}"
)
class TestChatToolResultRoundTrip:
def test_gemini(self, client: PassthroughClient, resources: ResourceManager) -> None:
model, key = _register(client, resources, _api_key_params(GEMINI_BACKEND, "GEMINI_API_KEY"))
_assert_tool_results_reach_the_model(client, key, model, thinking=None, tool_choice="required")
def test_mistral(self, client: PassthroughClient, resources: ResourceManager) -> None:
model, key = _register(client, resources, _api_key_params(MISTRAL_BACKEND, "MISTRAL_API_KEY"))
_assert_tool_results_reach_the_model(client, key, model, thinking=None, tool_choice="required")
def test_bedrock_converse(self, client: PassthroughClient, resources: ResourceManager) -> None:
model, key = _register(client, resources, _bedrock_params())
_assert_tool_results_reach_the_model(client, key, model, thinking=None, tool_choice="required")
def test_anthropic_with_extended_thinking(self, client: PassthroughClient, resources: ResourceManager) -> None:
model, key = _register(client, resources, _api_key_params(ANTHROPIC_BACKEND, "ANTHROPIC_API_KEY"))
_assert_tool_results_reach_the_model(client, key, model, thinking=THINKING, tool_choice=None)
def test_bedrock_converse_with_extended_thinking(
self, client: PassthroughClient, resources: ResourceManager
) -> None:
model, key = _register(client, resources, _bedrock_params())
_assert_tool_results_reach_the_model(client, key, model, thinking=THINKING, tool_choice=None)

23
tests/e2e/llm_translation/test_containers_e2e.py
View file

@ -46,6 +46,7 @@ import os
from types import MappingProxyType
from typing import Final
import openai
import pytest
from e2e_config import REQUEST_TIMEOUT, unique_marker
from e2e_http import unwrap
@ -198,3 +199,25 @@ class TestAzureContainerFiles:
)
resources.defer(lambda: client.containers.delete(native_id, extra_query=AZURE_PROVIDER_QUERY))
_assert_file_round_trip(client, native_id, marker)
class TestOpenAIContainerFiles:
def test_container_file_lifecycle_through_the_gateway(self, resources: ResourceManager, sdk: SdkClients) -> None:
client: Final = sdk.openai(resources.key())
marker: Final = unique_marker()
container: Final = client.containers.create(
name=f"e2e-container-{marker}", expires_after={"anchor": "last_active_at", "minutes": 5}
)
resources.defer(lambda: client.containers.delete(container.id))
assert not client.containers.files.list(container.id).data, "a new container must start with no files"
payload: Final = f"e2e container payload {marker}".encode()
uploaded: Final = client.containers.files.create(container.id, file=(f"{marker}.txt", payload))
listed: Final = tuple(entry.id for entry in client.containers.files.list(container.id).data)
assert uploaded.id in listed, f"uploaded file {uploaded.id} missing from the container listing {listed}"
assert client.containers.files.content.retrieve(uploaded.id, container_id=container.id).read() == payload
client.containers.files.delete(uploaded.id, container_id=container.id)
with pytest.raises(openai.NotFoundError):
client.containers.files.retrieve(uploaded.id, container_id=container.id)

81
tests/e2e/llm_translation/test_embeddings_endpoint_e2e.py
View file

@ -10,6 +10,8 @@ SDK refuses to build stay on the shared transport.
from __future__ import annotations
from typing import Final
import pytest
from e2e_config import provider_edge_base, unique_marker
from e2e_http import assert_client_error
@ -17,16 +19,28 @@ from lifecycle import ResourceManager
from models import LiteLLMParamsBody
from proxy_client import ProxyClient
from pydantic import BaseModel
from sdk_clients import NO_PROXY_CACHE, SdkClients
from sdk_clients import NO_PROXY_CACHE, SdkClients, response_header
pytestmark = pytest.mark.e2e
VERTEX_TEXT_EMBEDDING: Final = "vertex_ai/text-embedding-005"
VERTEX_MULTIMODAL_EMBEDDING: Final = "vertex_ai/multimodalembedding@001"
class _OptionalEmbeddingsBody(BaseModel):
model: str | None = None
input: str | list[str] | None = None
class _TokenEmbeddingsBody(BaseModel):
model: str
input: list[list[int]]
def _vertex_params(model: str) -> LiteLLMParamsBody:
return LiteLLMParamsBody(model=model, vertex_project="os.environ/VERTEXAI_PROJECT", vertex_location="us-central1")
def _openai_embeddings_params() -> LiteLLMParamsBody:
"""The OpenAI embeddings deployment, wired through the record/replay edge when a
fixture mode is active and straight at OpenAI otherwise (LIT-5974). Bedrock and
@ -116,6 +130,71 @@ class TestEmbeddingsEndpoint:
),
)
def test_mistral_embeddings_returns_vector(
self, proxy: ProxyClient, resources: ResourceManager, sdk: SdkClients
) -> None:
_assert_embedding_vector(
proxy,
resources,
sdk,
"e2e-embeddings-mistral",
LiteLLMParamsBody(model="mistral/mistral-embed", api_key="os.environ/MISTRAL_API_KEY"),
)
@pytest.mark.covers("llm.embeddings.vertex.basic.nonstream.works")
def test_vertex_embeddings_honor_requested_dimensions(
self, proxy: ProxyClient, resources: ResourceManager, sdk: SdkClients
) -> None:
model, key = _register(proxy, resources, "e2e-embeddings-vertex-dims", _vertex_params(VERTEX_TEXT_EMBEDDING))
embeddings = sdk.openai(key).embeddings.create(
model=model,
input="Say this is a test!",
dimensions=8,
extra_body={**NO_PROXY_CACHE, "task_type": "RETRIEVAL_QUERY", "auto_truncate": True},
)
assert len(embeddings.data[0].embedding) == 8, f"dimensions=8 was not honored: {embeddings!r}"
assert embeddings.usage.prompt_tokens > 0, f"vertex embeddings reported no prompt usage: {embeddings.usage!r}"
def test_vertex_multimodal_embeddings_honor_dimensions_and_are_costed(
self, proxy: ProxyClient, resources: ResourceManager, sdk: SdkClients
) -> None:
model, key = _register(
proxy, resources, "e2e-embeddings-vertex-mm", _vertex_params(VERTEX_MULTIMODAL_EMBEDDING)
)
raw = sdk.openai(key).embeddings.with_raw_response.create(
model=model, input="Say this is a test!", dimensions=128, extra_body=NO_PROXY_CACHE
)
embeddings = raw.parse()
assert len(embeddings.data[0].embedding) == 128, f"dimensions=128 was not honored: {embeddings!r}"
cost = response_header(raw.headers, "x-litellm-response-cost")
assert cost is not None and float(cost) > 0, f"multimodal embedding was not costed: {cost!r}"
def test_bedrock_titan_rejects_token_array_input_as_bad_request(
self, proxy: ProxyClient, resources: ResourceManager
) -> None:
model, key = _register(
proxy,
resources,
"e2e-embeddings-titan-tokens",
LiteLLMParamsBody(
model="bedrock/amazon.titan-embed-text-v2:0",
aws_access_key_id="os.environ/AWS_ACCESS_KEY_ID",
aws_secret_access_key="os.environ/AWS_SECRET_ACCESS_KEY",
aws_region_name="os.environ/AWS_REGION",
),
)
result = proxy.transport.send(
"/embeddings",
headers=proxy.transport.bearer(key),
json=_TokenEmbeddingsBody(model=model, input=[[1]]),
)
assert result.status_code == 400, (
f"titan cannot embed token arrays, so the caller must get a 400, got {result.status_code}: "
f"{result.body[:300]}"
)
@pytest.mark.replayable
@pytest.mark.covers("llm.embeddings.openai.basic.nonstream.works")
def test_array_input_returns_vectors(self, proxy: ProxyClient, resources: ResourceManager, sdk: SdkClients) -> None:

15
tests/e2e/llm_translation/test_messages_azure_foundry_e2e.py
View file

@ -18,6 +18,7 @@ from lifecycle import ResourceManager
from models import LiteLLMParamsBody
from proxy_client import ProxyClient
from sdk_clients import NO_PROXY_CACHE, SdkClients
from structured_output import SENTIMENT_OUTPUT_FORMAT, SENTIMENT_PROMPT, assert_sentiment_json
pytestmark = pytest.mark.e2e
@ -120,3 +121,17 @@ class TestAzureFoundryMessages:
event.type == "content_block_start" and event.content_block.type == "tool_use" for event in events
), "stream carried no tool_use block"
assert "message_stop" in event_types, "stream never reached message_stop"
def test_output_format_returns_schema_json(
self, proxy: ProxyClient, resources: ResourceManager, sdk: SdkClients
) -> None:
model = self._register(proxy, resources)
client = sdk.anthropic(resources.key(models=[model]))
message = client.messages.create(
model=model,
max_tokens=128,
messages=[{"role": "user", "content": SENTIMENT_PROMPT}],
extra_body={**NO_PROXY_CACHE, "output_format": SENTIMENT_OUTPUT_FORMAT},
)
assert_sentiment_json("".join(block.text for block in message.content if block.type == "text"))

81
tests/e2e/llm_translation/test_messages_bedrock_e2e.py Normal file
View file

@ -0,0 +1,81 @@
from __future__ import annotations
from typing import Final
import pytest
from anthropic.types import RawContentBlockDeltaEvent, RawMessageDeltaEvent, TextBlock, TextDelta
from e2e_config import unique_marker
from lifecycle import ResourceManager
from models import LiteLLMParamsBody
from proxy_client import ProxyClient
from sdk_clients import NO_PROXY_CACHE, SdkClients
from structured_output import SENTIMENT_OUTPUT_FORMAT, SENTIMENT_PROMPT, assert_sentiment_json
pytestmark = pytest.mark.e2e
CONVERSE_CLAUDE_BACKEND: Final = "bedrock/converse/us.anthropic.claude-haiku-4-5-20251001-v1:0"
NOVA_BACKEND: Final = "bedrock/us.amazon.nova-2-lite-v1:0"
def _register(proxy: ProxyClient, resources: ResourceManager, backend: str) -> str:
model = f"e2e-messages-bedrock-{unique_marker()}"
model_id = proxy.create_model(
model,
LiteLLMParamsBody(
model=backend,
aws_access_key_id="os.environ/AWS_ACCESS_KEY_ID",
aws_secret_access_key="os.environ/AWS_SECRET_ACCESS_KEY",
aws_region_name="os.environ/AWS_REGION",
),
)
resources.defer(lambda: proxy.delete_model(model_id))
return model
class TestBedrockMessages:
def test_converse_output_format_returns_schema_json_text(
self, proxy: ProxyClient, resources: ResourceManager, sdk: SdkClients
) -> None:
model = _register(proxy, resources, CONVERSE_CLAUDE_BACKEND)
client = sdk.anthropic(resources.key())
message = client.messages.create(
model=model,
max_tokens=128,
messages=[{"role": "user", "content": SENTIMENT_PROMPT}],
extra_body={**NO_PROXY_CACHE, "output_format": SENTIMENT_OUTPUT_FORMAT},
)
texts = tuple(block.text for block in message.content if isinstance(block, TextBlock))
assert len(texts) == len(message.content), f"structured output came back as non-text blocks: {message!r}"
assert_sentiment_json("".join(texts))
@pytest.mark.covers("llm.messages.bedrock_converse.basic.stream.works")
def test_nova_stream_relays_text_usage_and_stop(
self, proxy: ProxyClient, resources: ResourceManager, sdk: SdkClients
) -> None:
model = _register(proxy, resources, NOVA_BACKEND)
client = sdk.anthropic(resources.key())
events = tuple(
client.messages.create(
model=model,
max_tokens=64,
stream=True,
messages=[{"role": "user", "content": "Say hello in one short sentence."}],
extra_body=NO_PROXY_CACHE,
)
)
types = tuple(event.type for event in events)
text = "".join(
event.delta.text
for event in events
if isinstance(event, RawContentBlockDeltaEvent) and isinstance(event.delta, TextDelta)
)
assert text.strip(), f"streamed Nova reply carried no text: {types}"
assert types[0] == "message_start" and types[-1] == "message_stop", (
f"stream must open with message_start and end with message_stop: {types}"
)
deltas = tuple(event for event in events if isinstance(event, RawMessageDeltaEvent))
assert len(deltas) == 1, f"expected exactly one message_delta: {types}"
assert deltas[0].delta.stop_reason is not None, f"message_delta carried no stop_reason: {deltas[0]!r}"
assert deltas[0].usage.output_tokens > 0, f"message_delta reported no output tokens: {deltas[0]!r}"

17
tests/e2e/llm_translation/test_messages_e2e.py
View file

@ -49,6 +49,7 @@ from provider_edge_bedrock import bedrock_signer
from proxy_client import ProxyClient
from pydantic import BaseModel, ConfigDict, JsonValue, TypeAdapter, ValidationError
from sdk_clients import NO_PROXY_CACHE, SdkClients, response_header
from structured_output import SENTIMENT_OUTPUT_FORMAT, SENTIMENT_PROMPT, assert_sentiment_json
pytestmark = [pytest.mark.e2e, pytest.mark.replayable]
@ -106,6 +107,7 @@ def _user_turn(text: str) -> MessageParam:
return {"role": "user", "content": text}
class TestAnthropicMessages:
@pytest.mark.covers("llm.messages.anthropic.basic.nonstream.works")
def test_messages_returns_completion(self, proxy: ProxyClient, resources: ResourceManager, sdk: SdkClients) -> None:
@ -240,6 +242,21 @@ class TestAnthropicMessages:
f"model did not call the tool: {message.content!r}"
)
@pytest.mark.covers("llm.messages.anthropic.structured_output.nonstream.works")
def test_messages_output_format_returns_schema_json(
self, proxy: ProxyClient, resources: ResourceManager, sdk: SdkClients
) -> None:
model, key = _register(proxy, resources)
client = sdk.anthropic(key)
message = client.messages.create(
model=model,
max_tokens=128,
messages=[_user_turn(SENTIMENT_PROMPT)],
extra_body={**NO_PROXY_CACHE, "output_format": SENTIMENT_OUTPUT_FORMAT},
)
assert_sentiment_json(_text(message))
@pytest.mark.skip(
reason="stage red: product gap, /v1/messages 500s (anthropic_messages TypeError) on missing messages instead of 400"
)

62
tests/e2e/llm_translation/test_ocr_rust_e2e.py
View file

@ -118,6 +118,14 @@ class VertexOcr:
return LiteLLMParamsBody(model=self.model, vertex_location=self.location)
@dataclass(frozen=True, slots=True)
class CohereOcr:
model: str = "cohere/parse-v5.0"
def litellm_params(self) -> LiteLLMParamsBody:
return LiteLLMParamsBody(model=self.model, api_key="os.environ/COHERE_API_KEY")
@dataclass(frozen=True, slots=True)
class _OcrCase:
suffix: str
@ -150,6 +158,30 @@ RUST_OCR_CASES: tuple[_OcrCase, ...] = (
_CASE_IDS = tuple(case.suffix for case in RUST_OCR_CASES)
PDF_TEXT: Final = "test pdf file"
IMAGE_TEXT: Final = "litellm"
PDF_DOCUMENT: Final = OcrDocument(type="document_url", document_url=TEST_PDF_URL)
IMAGE_DOCUMENT: Final = OcrDocument(type="image_url", image_url=TEST_IMAGE_URL)
@dataclass(frozen=True, slots=True)
class _OcrContentCase:
suffix: str
provider: OcrProvider
document: OcrDocument
expected_text: str
OCR_CONTENT_CASES: Final = (
_OcrContentCase("mistral-pdf", MistralOcr(), PDF_DOCUMENT, PDF_TEXT),
_OcrContentCase("mistral-image", MistralOcr(), IMAGE_DOCUMENT, IMAGE_TEXT),
_OcrContentCase("azure-ai-image", AzureAiOcr("azure_ai/mistral-document-ai-2512"), IMAGE_DOCUMENT, IMAGE_TEXT),
_OcrContentCase(
"vertex-mistral-image", VertexOcr("vertex_ai/mistral-ocr-2505", "us-central1"), IMAGE_DOCUMENT, IMAGE_TEXT
),
_OcrContentCase("cohere-image", CohereOcr(), IMAGE_DOCUMENT, IMAGE_TEXT),
)
def _assert_ocr_document(response: OcrResponse) -> None:
assert response.object == "ocr", f"expected object='ocr', got {response.object!r}"
@ -198,3 +230,33 @@ class TestRustOcrGateway:
json=_OptionalOcrBody(model=model),
)
assert_client_error(result, "ocr missing document")
class TestOcrDocumentContent:
@pytest.mark.parametrize("case", OCR_CONTENT_CASES, ids=tuple(case.suffix for case in OCR_CONTENT_CASES))
def test_ocr_reads_the_document_and_bills_its_pages(
self, proxy: ProxyClient, resources: ResourceManager, case: _OcrContentCase
) -> None:
model = f"ocr-content-{case.suffix}-{unique_marker()}"
model_id = proxy.create_model(model, case.provider.litellm_params())
resources.defer(lambda: proxy.delete_model(model_id))
result = proxy.transport.send(
"/v1/ocr",
headers=proxy.transport.bearer(resources.key()),
json=OcrBody(model=model, document=case.document),
)
assert result.status_code == 200, f"{model}: /v1/ocr failed with {result.status_code}: {result.body[:300]}"
response = OcrResponse.model_validate_json(result.body)
assert response.object == "ocr", f"expected object='ocr', got {response.object!r}"
assert [page.index for page in response.pages] == list(range(len(response.pages))), (
f"page indexes are not contiguous from 0: {[page.index for page in response.pages]}"
)
text = " ".join(" ".join(page.markdown for page in response.pages).split()).lower()
assert case.expected_text in text, f"{model}: OCR text lost the document content: {text[:300]!r}"
assert response.usage_info is not None and response.usage_info.pages_processed == len(response.pages), (
f"usage_info.pages_processed disagrees with the returned pages: {response.usage_info!r}"
)
assert result.response_cost is not None and result.response_cost > 0, (
f"{model}: OCR call was not costed: x-litellm-response-cost={result.response_cost!r}"
)

132
tests/e2e/llm_translation/test_responses_e2e.py
View file

@ -24,15 +24,23 @@ from e2e_http import assert_client_error
from lifecycle import ResourceManager
from models import ChatBody, ChatMessage, LiteLLMParamsBody
from openai.types.responses import (
FunctionShellToolParam,
FunctionToolParam,
Response,
ResponseCompletedEvent,
ResponseFormatTextJSONSchemaConfigParam,
ResponseFunctionShellToolCall,
ResponseFunctionShellToolCallOutput,
ResponseFunctionToolCall,
ResponseInputItemParam,
ResponseInputParam,
ResponseOutputItemDoneEvent,
ResponseReasoningItem,
)
from provider_edge import LiveEdge, start_provider_edge
from provider_edge_bedrock import bedrock_signer
from proxy_client import ProxyClient
from pydantic import BaseModel
from pydantic import BaseModel, TypeAdapter
from sdk_clients import NO_PROXY_CACHE, SdkClients
pytestmark = pytest.mark.e2e
@ -464,3 +472,125 @@ class TestResponses:
json=_OptionalResponsesBody(model=model, input=""),
)
assert_client_error(result, "responses empty input")
REASONING_BACKEND: Final = "openai/gpt-5.4-mini"
SHELL_BACKEND: Final = "openai/gpt-5.5"
TOOL_DATE: Final = "2025-01-15"
GET_TODAY_TOOL: FunctionToolParam = {
"type": "function",
"name": "get_today",
"description": "Return today's date",
"parameters": {"type": "object", "properties": {}, "additionalProperties": False},
"strict": True,
}
TODAY_REPORT_FORMAT: ResponseFormatTextJSONSchemaConfigParam = {
"type": "json_schema",
"name": "today_report",
"strict": True,
"schema": {
"type": "object",
"properties": {"today": {"type": "string"}, "number_of_r": {"type": "string"}},
"required": ["today", "number_of_r"],
"additionalProperties": False,
},
}
SHELL_TOOL: FunctionShellToolParam = {"type": "shell", "environment": {"type": "container_auto"}}
_INPUT_ITEMS: Final = TypeAdapter(list[ResponseInputItemParam])
class TodayReport(BaseModel):
today: str
number_of_r: str
class TestResponsesOpenAIHostedFeatures:
def test_reasoning_items_replay_into_structured_output_after_tool_call(
self, proxy: ProxyClient, resources: ResourceManager, sdk: SdkClients
) -> None:
model = _register(
proxy,
resources,
LiteLLMParamsBody(model=REASONING_BACKEND, api_key="os.environ/OPENAI_API_KEY"),
prefix="e2e-responses-reasoning",
)
client = sdk.openai(resources.key())
question: ResponseInputItemParam = {
"role": "user",
"content": (
"How many r are in strrawberrry? Call get_today first, then report today exactly as get_today "
f"returned it and the count of r. {unique_marker()}"
),
}
first = client.responses.create(
model=model,
input=[question],
tools=[GET_TODAY_TOOL],
tool_choice={"type": "function", "name": "get_today"},
reasoning={"effort": "medium", "summary": "auto"},
text={"format": TODAY_REPORT_FORMAT},
extra_body=NO_PROXY_CACHE,
)
assert any(isinstance(item, ResponseReasoningItem) for item in first.output), (
f"reasoning model returned no reasoning item: {first.output!r}"
)
call = next((call for call in _function_calls(first) if call.name == "get_today"), None)
assert call is not None, f"forced get_today call missing: {first.output!r}"
replayed = _INPUT_ITEMS.validate_python([item.model_dump(exclude_none=True) for item in first.output])
tool_result: ResponseInputItemParam = {
"type": "function_call_output",
"call_id": call.call_id,
"output": TOOL_DATE,
}
second = client.responses.create(
model=model,
input=[question, *replayed, tool_result],
tools=[GET_TODAY_TOOL],
reasoning={"effort": "medium", "summary": "auto"},
text={"format": TODAY_REPORT_FORMAT},
extra_body=NO_PROXY_CACHE,
)
assert second.status == "completed", f"second turn did not complete: {second.status} {second.output!r}"
report = TodayReport.model_validate_json(second.output_text)
assert TOOL_DATE in report.today, f"structured output ignored the tool result: {report!r}"
@pytest.mark.provider_live
def test_shell_tool_stream_surfaces_shell_call_and_its_output(
self, proxy: ProxyClient, resources: ResourceManager, sdk: SdkClients
) -> None:
model = _register(
proxy,
resources,
LiteLLMParamsBody(model=SHELL_BACKEND, api_key="os.environ/OPENAI_API_KEY"),
prefix="e2e-responses-shell",
)
client = sdk.openai(resources.key())
stream = client.responses.create(
model=model,
input="Run `python --version` in the shell and reply with what it printed.",
tools=[SHELL_TOOL],
tool_choice="required",
max_output_tokens=1024,
stream=True,
extra_body=NO_PROXY_CACHE,
)
events = tuple(stream)
completed = events[-1] if events else None
assert isinstance(completed, ResponseCompletedEvent), (
f"shell stream did not end with response.completed: {[event.type for event in events]}"
)
streamed_items = tuple(event.item for event in events if isinstance(event, ResponseOutputItemDoneEvent))
assert any(isinstance(item, ResponseFunctionShellToolCall) for item in streamed_items), (
f"no shell_call item reached the stream: {[item.type for item in streamed_items]}"
)
outputs = tuple(
item for item in completed.response.output if isinstance(item, ResponseFunctionShellToolCallOutput)
)
assert outputs, f"completed response carries no shell_call_output: {completed.response.output!r}"

102
tests/e2e/llm_translation/test_responses_retrieve_e2e.py
View file

@ -6,17 +6,30 @@ Creates a stored response, retrieves it by id, and pins invalid-id error handlin
from __future__ import annotations
import time
from typing import Final
import openai
import pytest
from e2e_config import POLL_INTERVAL, POLL_TIMEOUT, unique_marker
from e2e_http import NoBody, Success, UnknownApiError, unwrap
from lifecycle import ResourceManager
from models import LiteLLMParamsBody
from openai.types.responses import (
ResponseCreatedEvent,
ResponseInputMessageItem,
ResponseInputText,
ResponseQueuedEvent,
)
from proxy_client import ProxyClient
from pydantic import BaseModel
from sdk_clients import NO_PROXY_CACHE, SdkClients
pytestmark = pytest.mark.e2e
OPENAI_BACKEND: Final = "openai/gpt-5.5"
LONG_TASK: Final = "Write a numbered list counting from 1 to 400, one number per line, with a short word after each."
CANCELLABLE_STATUSES: Final = frozenset({"queued", "in_progress"})
class ResponsesCreateBody(BaseModel):
model: str
@ -105,3 +118,92 @@ class TestResponsesRetrieve:
return
case other:
pytest.fail(f"invalid response id expected 404, got {other!r}")
def _register_openai(proxy: ProxyClient, resources: ResourceManager, prefix: str) -> str:
model = f"{prefix}-{unique_marker()}"
model_id = proxy.create_model(model, LiteLLMParamsBody(model=OPENAI_BACKEND, api_key="os.environ/OPENAI_API_KEY"))
resources.defer(lambda: proxy.delete_model(model_id))
return model
def _input_texts(item: object) -> tuple[str, ...]:
if not isinstance(item, ResponseInputMessageItem):
return ()
return tuple(part.text for part in item.content if isinstance(part, ResponseInputText))
@pytest.mark.provider_live
class TestStoredResponseLifecycle:
def test_input_items_list_the_stored_prompt(
self, proxy: ProxyClient, resources: ResourceManager, sdk: SdkClients
) -> None:
model = _register_openai(proxy, resources, "e2e-resp-items")
client = sdk.openai(resources.key())
marker = unique_marker()
created = client.responses.create(
model=model, input=f"Reply with one word. {marker}", store=True, extra_body=NO_PROXY_CACHE
)
items = client.responses.input_items.list(created.id, limit=20, order="desc").data
texts = tuple(text for item in items for text in _input_texts(item))
assert any(marker in text for text in texts), f"input_items did not list the stored prompt: {items!r}"
def test_deleted_response_is_no_longer_retrievable(
self, proxy: ProxyClient, resources: ResourceManager, sdk: SdkClients
) -> None:
model = _register_openai(proxy, resources, "e2e-resp-delete")
client = sdk.openai(resources.key())
created = client.responses.create(
model=model, input=f"Reply with one word. {unique_marker()}", store=True, extra_body=NO_PROXY_CACHE
)
retrieved = client.responses.retrieve(created.id)
assert retrieved.status == "completed", f"stored response not retrievable as completed: {retrieved!r}"
client.responses.delete(created.id)
with pytest.raises(openai.APIStatusError) as gone:
client.responses.retrieve(created.id)
assert 400 <= gone.value.status_code < 500, f"retrieve after delete expected a 4xx: {gone.value!r}"
@pytest.mark.provider_live
class TestBackgroundResponseCancel:
def test_cancel_background_response(
self, proxy: ProxyClient, resources: ResourceManager, sdk: SdkClients
) -> None:
model = _register_openai(proxy, resources, "e2e-resp-cancel")
client = sdk.openai(resources.key())
created = client.responses.create(
model=model, input=f"{LONG_TASK} {unique_marker()}", background=True, extra_body=NO_PROXY_CACHE
)
assert created.status in CANCELLABLE_STATUSES, f"background response was not queued: {created.status}"
cancelled = client.responses.cancel(created.id)
assert cancelled.status == "cancelled", f"cancel did not stop the response: {cancelled.status}"
def test_cancel_background_streaming_response_by_streamed_id(
self, proxy: ProxyClient, resources: ResourceManager, sdk: SdkClients
) -> None:
model = _register_openai(proxy, resources, "e2e-resp-cancel-stream")
client = sdk.openai(resources.key())
stream = client.responses.create(
model=model,
input=f"{LONG_TASK} {unique_marker()}",
background=True,
stream=True,
extra_body=NO_PROXY_CACHE,
)
response_id = next(
(event.response.id for event in stream if isinstance(event, (ResponseCreatedEvent, ResponseQueuedEvent))),
None,
)
stream.close()
assert response_id, "background stream advertised no response id before the first output"
cancelled = client.responses.cancel(response_id)
assert cancelled.status == "cancelled", f"cancel by streamed id did not stop the response: {cancelled.status}"

14
tests/e2e/models.py
View file

@ -296,10 +296,18 @@ class ToolCall(BaseModel):
function: ToolCallFunction = ToolCallFunction()
class ThinkingBlock(BaseModel):
type: str
thinking: str | None = None
signature: str | None = None
data: str | None = None
class ChatAssistantTurn(BaseModel):
role: Literal["assistant"] = "assistant"
content: str | None = None
reasoning_content: str | None = None
thinking_blocks: list[ThinkingBlock] | None = None
tool_calls: list[ToolCall] | None = None
@ -422,6 +430,7 @@ class OutMessage(BaseModel):
role: str | None = None
content: str | None = None
reasoning_content: str | None = None
thinking_blocks: list[ThinkingBlock] | None = None
tool_calls: list[ToolCall] | None = None
provider_specific_fields: McpResponseMetadata | None = None
@ -817,10 +826,15 @@ class OcrPage(BaseModel):
markdown: str
class OcrUsageInfo(BaseModel):
pages_processed: int | None = None
class OcrResponse(BaseModel):
object: str | None = None
model: str | None = None
pages: list[OcrPage] = []
usage_info: OcrUsageInfo | None = None
# ---------- completions ----------

185
tests/guardrails_tests/test_bedrock_guardrails.py
View file

@ -7,7 +7,6 @@ from litellm.proxy.guardrails.guardrail_hooks.bedrock_guardrails import (
_redact_pii_matches,
)
from litellm.proxy._types import UserAPIKeyAuth
from litellm.caching import DualCache
from unittest.mock import MagicMock, AsyncMock, patch
@ -101,190 +100,6 @@ async def test_bedrock_guardrails_pii_masking_content_list():
)
@pytest.mark.asyncio
async def test_bedrock_guardrails_block_messages_api():
"""
Test that guardrails block messages API requests containing 'coffee' and raise the expected exception.
"""
from fastapi import HTTPException
# Create proper mock objects
mock_user_api_key_dict = UserAPIKeyAuth()
guardrail = BedrockGuardrail(
guardrailIdentifier="ff6ujrregl1q",
guardrailVersion="DRAFT",
)
request_data = {
"model": "claude-sonnet-4-5-20250929",
"messages": [
{
"role": "user",
"content": [
{
"type": "text",
"text": "Hello, my phone number is +1 412 555 1212",
},
{"type": "text", "text": "what time is it?"},
],
},
{"role": "user", "content": "tell me about coffee"},
],
}
with pytest.raises(HTTPException) as exc_info:
await guardrail.async_pre_call_hook(
data=request_data,
user_api_key_dict=mock_user_api_key_dict,
call_type="anthropic_messages",
cache=MagicMock(spec=DualCache),
)
exception = exc_info.value
assert exception.status_code == 400
detail = exception.detail
assert isinstance(detail, dict)
assert detail["error"] == "Violated guardrail policy"
assert (
detail["bedrock_guardrail_response"]
== "Sorry, the model cannot answer this question. coffee guardrail applied "
)
@pytest.mark.asyncio
async def test_bedrock_guardrails_block_responses_api():
"""
Test that guardrails block responses API requests containing 'coffee' and raise the expected exception.
"""
from fastapi import HTTPException
# Create proper mock objects
mock_user_api_key_dict = UserAPIKeyAuth()
guardrail = BedrockGuardrail(
guardrailIdentifier="ff6ujrregl1q",
guardrailVersion="DRAFT",
)
request_data = {
"model": "gpt-4.1",
"input": "Tell me a three sentence bedtime story about a unicorn drinking coffee",
"stream": False,
}
with pytest.raises(HTTPException) as exc_info:
await guardrail.async_pre_call_hook(
data=request_data,
user_api_key_dict=mock_user_api_key_dict,
call_type="responses",
cache=MagicMock(spec=DualCache),
)
exception = exc_info.value
assert exception.status_code == 400
detail = exception.detail
assert isinstance(detail, dict)
assert detail["error"] == "Violated guardrail policy"
assert (
detail["bedrock_guardrail_response"]
== "Sorry, the model cannot answer this question. coffee guardrail applied "
)
@pytest.mark.asyncio
async def test_bedrock_guardrails_with_streaming():
from fastapi import HTTPException
from litellm.proxy.utils import ProxyLogging
from litellm.types.guardrails import GuardrailEventHooks
# Create proper mock objects
mock_user_api_key_cache = MagicMock(spec=DualCache)
mock_user_api_key_dict = UserAPIKeyAuth()
async def _stream_through_guardrail():
proxy_logging_obj = ProxyLogging(
user_api_key_cache=mock_user_api_key_cache,
premium_user=True,
)
guardrail = BedrockGuardrail(
guardrailIdentifier="ff6ujrregl1q",
guardrailVersion="DRAFT",
supported_event_hooks=[GuardrailEventHooks.post_call],
guardrail_name="bedrock-post-guard",
)
litellm.callbacks.append(guardrail)
request_data = {
"model": "gpt-5.5",
"messages": [{"role": "user", "content": "Hi I like coffee"}],
"stream": True,
"metadata": {"guardrails": ["bedrock-post-guard"]},
}
response = await litellm.acompletion(
**request_data,
)
response = proxy_logging_obj.async_post_call_streaming_iterator_hook(
user_api_key_dict=mock_user_api_key_dict,
response=response,
request_data=request_data,
)
async for chunk in response:
print(chunk)
with pytest.raises(HTTPException):
await _stream_through_guardrail()
@pytest.mark.asyncio
async def test_bedrock_guardrails_with_streaming_no_violation():
from litellm.proxy.utils import ProxyLogging
from litellm.types.guardrails import GuardrailEventHooks
# Create proper mock objects
mock_user_api_key_cache = MagicMock(spec=DualCache)
mock_user_api_key_dict = UserAPIKeyAuth()
proxy_logging_obj = ProxyLogging(
user_api_key_cache=mock_user_api_key_cache,
premium_user=True,
)
guardrail = BedrockGuardrail(
guardrailIdentifier="ff6ujrregl1q",
guardrailVersion="DRAFT",
supported_event_hooks=[GuardrailEventHooks.post_call],
guardrail_name="bedrock-post-guard",
)
litellm.callbacks.append(guardrail)
request_data = {
"model": "gpt-5.5",
"messages": [{"role": "user", "content": "hi"}],
"stream": True,
"metadata": {"guardrails": ["bedrock-post-guard"]},
}
response = await litellm.acompletion(
**request_data,
)
response = proxy_logging_obj.async_post_call_streaming_iterator_hook(
user_api_key_dict=mock_user_api_key_dict,
response=response,
request_data=request_data,
)
async for chunk in response:
print(chunk)
@pytest.mark.asyncio
async def test_bedrock_guardrails_streaming_request_body_mock():
"""Test that the exact request body sent to Bedrock matches expected format when using streaming"""

187
tests/llm_responses_api_testing/base_responses_api.py
View file

@ -105,10 +105,6 @@ class BaseResponsesAPITest(ABC):
"""Must return the base completion call args"""
pass
def get_base_completion_reasoning_call_args(self) -> dict:
"""Must return the base completion reasoning call args"""
return None
def get_advanced_model_for_shell_tool(self) -> Optional[str]:
"""If specified, overrides the model used by test_responses_api_shell_tool_streaming_sees_shell_output (e.g. openai/gpt-5.2 for shell support)."""
return None
@ -351,32 +347,6 @@ class BaseResponsesAPITest(ABC):
else:
raise ValueError("response is not a ResponsesAPIResponse")
@pytest.mark.asyncio
@pytest.mark.flaky(retries=3, delay=2)
async def test_basic_openai_list_input_items_endpoint(self):
"""Test that calls the OpenAI List Input Items endpoint"""
litellm._turn_on_debug()
response = await litellm.aresponses(
model="gpt-5.5",
input="Tell me a three sentence bedtime story about a unicorn.",
)
print("Initial response=", json.dumps(response, indent=4, default=str))
response_id = response.get("id")
assert response_id is not None, "Response should have an ID"
print(f"Got response_id: {response_id}")
list_items_response = await litellm.alist_input_items(
response_id=response_id,
limit=20,
order="desc",
)
print(
"List items response=",
json.dumps(list_items_response, indent=4, default=str),
)
@pytest.mark.asyncio
async def test_multiturn_responses_api(self):
litellm._turn_on_debug()
@ -477,99 +447,6 @@ class BaseResponsesAPITest(ABC):
else:
assert len(response["output"]) > 0
@pytest.mark.asyncio
async def test_responses_api_multi_turn_with_reasoning_and_structured_output(self):
"""
Test multi-turn conversation with reasoning, structured output, and tool calls.
This test validates:
- First call: Model uses reasoning to process a question and makes a tool call
- Tool call handling: Function call output is properly processed
- Second call: Model produces structured output incorporating tool results
- Structured output: Response conforms to defined Pydantic model schema
"""
from pydantic import BaseModel
litellm._turn_on_debug()
litellm.set_verbose = True
base_completion_call_args = self.get_base_completion_reasoning_call_args()
if base_completion_call_args is None:
pytest.skip("Skipping test due to no base completion reasoning call args")
# Define tools for the conversation
tools = [{"type": "function", "name": "get_today"}]
# Define structured output schema
class Output(BaseModel):
today: str
number_of_r: str
# Initial conversation input
input_messages = [
{
"role": "user",
"content": "How many r in strrawberrry? While you're thinking, you should call tool get_today. Then you output the today and number of r",
}
]
# First call - should trigger reasoning and tool call
response = await litellm.aresponses(
input=input_messages,
tools=tools,
reasoning={"effort": "low", "summary": "detailed"},
text_format=Output,
**base_completion_call_args,
)
print("First call output:")
print(json.dumps(response.output, indent=4, default=str))
# Validate first response structure
validate_responses_api_response(response, final_chunk=True)
assert response.output is not None
assert len(response.output) > 0
# Extend input with first response output
input_messages.extend(response.output)
# Process any tool calls and add function outputs
function_outputs = []
for item in response.output:
if hasattr(item, "type") and item.type in [
"function_call",
"custom_tool_call",
]:
if hasattr(item, "name") and item.name == "get_today":
function_outputs.append(
{
"type": "function_call_output",
"call_id": item.call_id,
"output": "2025-01-15",
}
)
# Add function outputs to conversation
input_messages.extend(function_outputs)
print("Second call input:")
print(json.dumps(input_messages, indent=4, default=str))
# Second call - should produce structured output
final_response = await litellm.aresponses(
input=input_messages,
tools=tools,
reasoning={"effort": "low", "summary": "detailed"},
text_format=Output,
**base_completion_call_args,
)
print("Second call output:")
print(json.dumps(final_response.output, indent=4, default=str))
# Validate final response structure
validate_responses_api_response(final_response, final_chunk=True)
assert final_response.output is not None
def test_openai_responses_api_dict_input_filtering(self):
"""
Test that regular dict inputs with status fields are properly filtered
@ -779,67 +656,3 @@ class BaseResponsesAPITest(ABC):
assert response.get("id") is not None
assert response.get("status") is not None
@pytest.mark.asyncio
async def test_responses_api_shell_tool_streaming_sees_shell_output(self):
"""
E2E streaming call with Shell tool; validate we can see shell output in the stream.
Calls aresponses(..., tools=[shell], stream=True), then iterates the stream and
asserts at least one event is shell-related or response output contains shell_call.
Skips when model does not support shell (e.g. gpt-5.5).
"""
base_completion_call_args = self.get_base_completion_call_args()
model = (
self.get_advanced_model_for_shell_tool()
or base_completion_call_args.get("model")
or "openai/gpt-5.2"
)
if "openai/" not in str(model):
pytest.skip(
"Shell tool streaming e2e is only run for OpenAI/Azure Responses API"
)
tools = [{"type": "shell", "environment": {"type": "container_auto"}}]
input_msg = "List files in /mnt/data and run python --version."
stream = await litellm.aresponses(
**{**base_completion_call_args, "model": model},
input=input_msg,
max_output_tokens=512,
tools=tools,
tool_choice="auto",
stream=True,
)
event_types_seen = []
output_items_with_shell = []
async for event in stream:
print("event=", json.dumps(event, indent=4, default=str))
event_type = getattr(event, "type", None) or (
event.get("type") if isinstance(event, dict) else None
)
if event_type is not None:
event_types_seen.append(str(event_type))
if "shell" in str(event_type or "").lower():
output_items_with_shell.append(event_type)
response_obj = getattr(event, "response", None) or (
event.get("response") if isinstance(event, dict) else None
)
if response_obj is not None:
output = getattr(response_obj, "output", None) or (
response_obj.get("output")
if isinstance(response_obj, dict)
else None
)
if isinstance(output, list):
for item in output:
item_type = getattr(item, "type", None) or (
item.get("type") if isinstance(item, dict) else None
)
if item_type and "shell" in str(item_type).lower():
output_items_with_shell.append(item_type)
assert len(event_types_seen) > 0, "Expected at least one stream event"
assert (
len(output_items_with_shell) > 0
), f"Expected to see shell output in stream; event types seen: {event_types_seen!r}"

5
tests/llm_responses_api_testing/test_openai_responses_api.py
View file

@ -28,11 +28,6 @@ class TestOpenAIResponsesAPITest(BaseResponsesAPITest):
"model": "openai/gpt-5.5",
}
def get_base_completion_reasoning_call_args(self):
return {
"model": "openai/gpt-5-mini",
}
def get_advanced_model_for_shell_tool(self):
return "openai/gpt-5.2"

7
tests/llm_translation/realtime/base_realtime_tests.py
View file

@ -217,13 +217,6 @@ class BaseRealtimeTest(ABC):
f"exception: {type(caught_exception).__name__}: {caught_exception}"
)
# Skip on transient connection failures
if (
not websocket_client.connection_successful
and websocket_client.close_code is not None
):
pytest.skip(f"Transient connection failure: {'; '.join(error_details)}")
# Assertions
assert (
websocket_client.connection_successful

183
tests/llm_translation/realtime/test_openai_realtime.py
View file

@ -19,9 +19,6 @@ async def test_openai_realtime_direct_call_no_intent():
End-to-end test calling the actual OpenAI realtime endpoint via LiteLLM SDK
without intent parameter. This should succeed without "Invalid intent" error.
Uses real websocket connection to OpenAI.
Note: This test may be skipped on transient connection failures since it depends
on external OpenAI API availability.
"""
import asyncio
import json
@ -125,16 +122,6 @@ async def test_openai_realtime_direct_call_no_intent():
f"exception: {type(caught_exception).__name__}: {caught_exception}"
)
# Skip test on transient connection failures (e.g., WebSocket connection rejected)
# These are not regressions, just external API availability issues
if (
not websocket_client.connection_successful
and websocket_client.close_code is not None
):
pytest.skip(
f"Skipping due to transient connection failure: close_code={websocket_client.close_code}, close_reason={websocket_client.close_reason}"
)
assert (
websocket_client.connection_successful
), f"Failed to establish connection. Debug info: {'; '.join(error_details)}"
@ -154,176 +141,6 @@ async def test_openai_realtime_direct_call_no_intent():
assert "model" in session_message["session"], "Session object missing model field"
@pytest.mark.asyncio
@pytest.mark.skipif(
os.environ.get("OPENAI_API_KEY", None) is None,
reason="No OpenAI API key provided",
)
async def test_openai_realtime_direct_call_with_intent():
"""
End-to-end test calling the actual OpenAI realtime endpoint via LiteLLM SDK
with explicit intent parameter. This should include the intent in the URL.
Uses real websocket connection to OpenAI.
Note: This test may be skipped on transient connection failures since it depends
on external OpenAI API availability.
"""
import asyncio
import json
class RealTimeWebSocketClient:
def __init__(self):
self.messages_sent = []
self.messages_received = []
self.received_session_created = False
self.connection_successful = False
self._receive_called = False
self.intent_error_received = None
self.close_code = None
self.close_reason = None
async def accept(self):
pass
async def send_text(self, message):
self.messages_sent.append(message)
try:
if isinstance(message, bytes):
message_str = message.decode("utf-8")
else:
message_str = message
msg_data = json.loads(message_str)
msg_type = msg_data.get("type", "unknown")
if msg_type == "error":
error_info = msg_data.get("error", {})
error_code = error_info.get("code", "unknown")
error_message = error_info.get("message", "unknown")
if error_code == "invalid_intent":
self.intent_error_received = {
"code": error_code,
"message": error_message,
}
# Don't fail on other errors, just record them
self.messages_received.append(msg_data)
return
if msg_type == "session.created" and not self.received_session_created:
self.messages_received.append(msg_data)
self.received_session_created = True
self.connection_successful = True
except (json.JSONDecodeError, UnicodeDecodeError):
# Non-JSON messages are acceptable
pass
async def receive_text(self):
if not self._receive_called:
self._receive_called = True
max_wait = 60.0
check_interval = 0.1
waited = 0.0
while waited < max_wait:
if self.connection_successful:
break
await asyncio.sleep(check_interval)
waited += check_interval
if not self.connection_successful:
await asyncio.sleep(3.0)
raise ConnectionClosedOK(None, None)
async def close(self, code=1000, reason=""):
self.close_code = code
self.close_reason = reason
@property
def headers(self):
return {}
websocket_client = RealTimeWebSocketClient()
caught_exception = None
# OpenAI shut down the gpt-4o-realtime-preview family (incl. the undated
# alias) on 2026-05-07; gpt-realtime is the GA successor.
query_params: RealtimeQueryParams = {
"model": "openai/gpt-realtime",
"intent": "chat",
}
try:
await litellm._arealtime(
model="openai/gpt-realtime",
websocket=websocket_client,
api_key=os.environ.get("OPENAI_API_KEY"),
query_params=query_params,
timeout=60,
)
except (ConnectionClosedOK, ConnectionClosedError):
pass
except Exception as e:
caught_exception = e
if "invalid_intent" in str(e).lower():
pytest.fail(f"Unexpected invalid intent error: {e}")
# Other exceptions are recorded but don't fail immediately
if websocket_client.intent_error_received:
websocket_client.connection_successful = True
# Build detailed error message for debugging
error_details = []
error_details.append(f"messages_sent count: {len(websocket_client.messages_sent)}")
error_details.append(
f"messages_received count: {len(websocket_client.messages_received)}"
)
error_details.append(f"close_code: {websocket_client.close_code}")
error_details.append(f"close_reason: {websocket_client.close_reason}")
if caught_exception:
error_details.append(
f"exception: {type(caught_exception).__name__}: {caught_exception}"
)
# Skip test on transient connection failures (e.g., WebSocket connection rejected)
# These are not regressions, just external API availability issues
if (
not websocket_client.connection_successful
and websocket_client.close_code is not None
):
pytest.skip(
f"Skipping due to transient connection failure: close_code={websocket_client.close_code}, close_reason={websocket_client.close_reason}"
)
assert (
websocket_client.connection_successful
), f"Failed to establish connection or verify intent parameter pass-through. Debug info: {'; '.join(error_details)}"
if websocket_client.received_session_created:
assert len(websocket_client.messages_received) > 0, "No messages received"
session_message = websocket_client.messages_received[0]
assert (
session_message["type"] == "session.created"
), f"Expected session.created, got {session_message.get('type')}"
assert (
"session" in session_message
), "session.created response missing session object"
assert "id" in session_message["session"], "Session object missing id field"
assert (
"model" in session_message["session"]
), "Session object missing model field"
elif websocket_client.intent_error_received:
# invalid_intent error confirms intent parameter was passed through
pass
else:
pytest.fail(
f"Unexpected test state: connection_successful={websocket_client.connection_successful}, "
f"received_session_created={websocket_client.received_session_created}, "
f"intent_error_received={websocket_client.intent_error_received}"
)
def test_realtime_query_params_construction():
"""
Test that query params are constructed correctly by the proxy server logic

81
tests/llm_translation/test_bedrock_completion.py
View file

@ -2506,43 +2506,6 @@ def test_bedrock_error_handling_streaming(exception_type, expected_status_code):
assert e.value.status_code == expected_status_code
@pytest.mark.parametrize(
"image_url",
[
"https://www.w3.org/WAI/ER/tests/xhtml/testfiles/resources/pdf/dummy.pdf",
# "https://raw.githubusercontent.com/datasets/gdp/master/data/gdp.csv",
"https://www.cmu.edu/blackboard/files/evaluate/tests-example.xls",
# "https://raw.githubusercontent.com/datasets/sample-data/master/README.txt", # invalid url
"https://raw.githubusercontent.com/mdn/content/main/README.md",
],
)
@pytest.mark.flaky(retries=6, delay=2)
@pytest.mark.asyncio
async def test_bedrock_document_understanding(image_url):
from litellm import acompletion
litellm._turn_on_debug()
model = "bedrock/us.amazon.nova-pro-v1:0"
image_content = [
{"type": "text", "text": f"What's this file about?"},
{
"type": "image_url",
"image_url": image_url,
},
]
try:
response = await acompletion(
model=model,
messages=[{"role": "user", "content": image_content}],
)
assert response is not None
assert response.choices[0].message.content != ""
except litellm.ServiceUnavailableError as e:
pytest.skip("Skipping test due to ServiceUnavailableError")
def test_bedrock_custom_proxy():
from litellm.llms.custom_httpx.http_handler import HTTPHandler
@ -3092,50 +3055,6 @@ def test_bedrock_meta_llama_function_calling():
print(response)
@pytest.mark.asyncio
@pytest.mark.parametrize("sync_mode", [True, False])
async def test_bedrock_passthrough(sync_mode: bool):
import litellm
litellm._turn_on_debug()
data = {
"max_tokens": 512,
"messages": [{"role": "user", "content": "Hey"}],
"system": [
{
"type": "text",
"text": "Analyze if this message indicates a new conversation topic. If it does, extract a 2-3 word title that captures the new topic. Format your response as a JSON object with two fields: 'isNewTopic' (boolean) and 'title' (string, or null if isNewTopic is false). Only include these fields, no other text.",
}
],
"temperature": 0,
"metadata": {
"user_id": "5dd07c33da27e6d2968d94ea20bf47a7b090b6b158b82328d54da2909a108e84"
},
"anthropic_version": "bedrock-2023-05-31",
"anthropic_beta": ["claude-code-20250219"],
}
if sync_mode:
response = litellm.llm_passthrough_route(
model="bedrock/us.anthropic.claude-haiku-4-5-20251001-v1:0",
method="POST",
endpoint="/model/us.anthropic.claude-haiku-4-5-20251001-v1:0/invoke",
data=data,
)
else:
response = await litellm.allm_passthrough_route(
model="bedrock/us.anthropic.claude-haiku-4-5-20251001-v1:0",
method="POST",
endpoint="/model/us.anthropic.claude-haiku-4-5-20251001-v1:0/invoke",
data=data,
)
print(response.text)
assert response.status_code == 200
@pytest.mark.asyncio
async def test_bedrock_passthrough_router():
"""

110
tests/llm_translation/test_containers_api.py
View file

@ -1,110 +0,0 @@
"""
E2E Test for Container Files API.
Tests the container files endpoints using LiteLLM SDK methods.
"""
import os
import time
import pytest
from litellm.containers import (
create_container,
delete_container,
)
from litellm.containers.endpoint_factory import (
list_container_files,
retrieve_container_file,
retrieve_container_file_content,
delete_container_file,
)
@pytest.mark.skipif(not os.getenv("OPENAI_API_KEY"), reason="OPENAI_API_KEY not set")
def test_container_files_api():
"""
Test container files API: list, retrieve, delete.
Flow:
1. Create a container
2. List files (should be empty)
3. Try retrieve file (should error - no files)
4. Try delete file (should error - no files)
5. Cleanup: delete container
"""
api_key = os.getenv("OPENAI_API_KEY")
# 1. Create container
print("\n1. Creating container...")
container = create_container(
name=f"test-files-api-{int(time.time())}",
custom_llm_provider="openai",
api_key=api_key,
expires_after={"anchor": "last_active_at", "minutes": 5},
)
print(f" Created: {container.id}")
try:
# 2. List files
print("2. Listing container files...")
files = list_container_files(
container_id=container.id,
custom_llm_provider="openai",
api_key=api_key,
)
assert files.object == "list"
assert isinstance(files.data, list)
assert len(files.data) == 0 # New container has no files
print(f" Files found: {len(files.data)} ✓")
# 3. Try retrieve non-existent file metadata (should raise error)
print("3. Testing retrieve_container_file (expect error)...")
with pytest.raises(Exception, match=r"(?i)not found|invalid"):
retrieve_container_file(
container_id=container.id,
file_id="cfile_nonexistent",
custom_llm_provider="openai",
api_key=api_key,
)
# 3b. Try retrieve non-existent file content (should raise error)
print("3b. Testing retrieve_container_file_content (expect error)...")
try:
retrieve_container_file_content(
container_id=container.id,
file_id="cfile_nonexistent",
custom_llm_provider="openai",
api_key=api_key,
)
pytest.fail("Should have raised error for non-existent file content")
except Exception as e:
print(f" Got expected error ✓")
# 4. Try delete non-existent file (should raise error)
print("4. Testing delete_container_file (expect error)...")
try:
delete_container_file(
container_id=container.id,
file_id="cfile_nonexistent",
custom_llm_provider="openai",
api_key=api_key,
)
pytest.fail("Should have raised error for non-existent file")
except Exception as e:
# Delete returns 400 for non-existent files
print(f" Got expected error ✓")
finally:
# 5. Cleanup
print("5. Deleting container...")
result = delete_container(
container_id=container.id,
custom_llm_provider="openai",
api_key=api_key,
)
assert result.deleted is True
print(f" Deleted ✓")
print("\nAll container files API tests passed! ✓")

2
tests/llm_translation/test_mistral_api.py
View file

@ -24,6 +24,8 @@ from base_llm_unit_tests import BaseLLMChatTest
@pytest.mark.flaky(retries=3, delay=2)
class TestMistralCompletion(BaseLLMChatTest):
test_basic_tool_calling = None
def get_base_completion_call_args(self) -> dict:
litellm.set_verbose = True
return {"model": "mistral/mistral-medium-latest"}

2
tests/llm_translation/test_openai.py
View file

@ -273,6 +273,8 @@ async def test_vision_with_custom_model():
class TestOpenAIChatCompletion(BaseLLMChatTest):
test_basic_tool_calling = None
def get_base_completion_call_args(self) -> dict:
return {"model": "gpt-4o-mini"}

2
tests/llm_translation/test_together_ai.py
View file

@ -15,6 +15,8 @@ import pytest
class TestTogetherAI(BaseLLMChatTest):
test_basic_tool_calling = None
def get_base_completion_call_args(self) -> dict:
litellm.set_verbose = True
return {

265
tests/local_testing/test_amazing_vertex_completion.py
View file

@ -694,93 +694,6 @@ def test_gemini_pro_grounding(value_in_dict):
# @pytest.mark.skip(reason="exhausted vertex quota. need to refactor to mock the call")
@pytest.mark.parametrize(
"model", ["vertex_ai_beta/gemini-2.5-flash-lite"]
) # "vertex_ai",
@pytest.mark.parametrize("sync_mode", [True]) # "vertex_ai",
@pytest.mark.asyncio
@pytest.mark.flaky(retries=6, delay=2)
async def test_gemini_pro_function_calling_httpx(model, sync_mode):
try:
load_vertex_ai_credentials()
litellm.set_verbose = True
messages = [
{
"role": "system",
"content": "Your name is Litellm Bot, you are a helpful assistant",
},
# User asks for their name and weather in San Francisco
{
"role": "user",
"content": "Hello, what is your name and can you tell me the weather?",
},
]
tools = [
{
"type": "function",
"function": {
"name": "get_weather",
"description": "Get the current weather in a given location",
"parameters": {
"type": "object",
"properties": {
"location": {
"type": "string",
"description": "The city and state, e.g. San Francisco, CA",
}
},
"required": ["location"],
},
},
}
]
data = {
"model": model,
"messages": messages,
"tools": tools,
"tool_choice": "required",
"timeout": 60, # Add explicit timeout
}
print(f"Model for call - {model}")
if sync_mode:
response = litellm.completion(**data)
else:
response = await litellm.acompletion(**data)
print(f"response: {response}")
assert response.choices[0].message.tool_calls[0].function.arguments is not None
assert isinstance(
response.choices[0].message.tool_calls[0].function.arguments, str
)
except litellm.RateLimitError as e:
pytest.skip(f"Rate limit exceeded: {str(e)}")
except litellm.ServiceUnavailableError as e:
pytest.skip(f"Service unavailable: {str(e)}")
except litellm.Timeout as e:
pytest.skip(f"Request timeout: {str(e)}")
except Exception as e:
error_msg = str(e)
# Skip test for known transient API issues
if any(
x in error_msg
for x in [
"429 Quota exceeded",
"503",
"Service unavailable",
"timeout",
"Timeout",
"UNAVAILABLE",
]
):
pytest.skip(f"Transient API error: {error_msg}")
else:
pytest.fail(f"An unexpected exception occurred - {error_msg}")
from test_completion import response_format_tests
@ -854,68 +767,6 @@ async def test_partner_models_httpx(model, region, sync_mode):
pytest.fail("An unexpected exception occurred - {}".format(str(e)))
@pytest.mark.parametrize(
"model,region",
[
# vertex_ai/meta/llama-4-scout-17b-16e-instruct-maas removed - consistently returns 400 BadRequest on Vertex AI
# vertex_ai/qwen/qwen3-coder-480b-a35b-instruct-maas removed - us-south1 endpoint unavailable in CI
(
"vertex_ai/mistral-small-2503",
"us-central1",
), # critical - we had this issue: https://github.com/BerriAI/litellm/issues/13888
("vertex_ai/openai/gpt-oss-20b-maas", "us-central1"),
],
)
@pytest.mark.parametrize(
"sync_mode",
[True, False], #
) #
@pytest.mark.asyncio
@pytest.mark.flaky(retries=3, delay=1)
async def test_partner_models_httpx_streaming(model, region, sync_mode):
try:
load_vertex_ai_credentials()
litellm._turn_on_debug()
messages = [
{
"role": "system",
"content": "Your name is Litellm Bot, you are a helpful assistant",
},
# User asks for their name and weather in San Francisco
{
"role": "user",
"content": "Hello, what is your name and can you tell me the weather?",
},
]
data = {
"model": model,
"messages": messages,
"stream": True,
"vertex_ai_location": region,
}
if sync_mode:
response = litellm.completion(**data)
for idx, chunk in enumerate(response):
streaming_format_tests(idx=idx, chunk=chunk)
else:
response = await litellm.acompletion(**data)
idx = 0
async for chunk in response:
streaming_format_tests(idx=idx, chunk=chunk)
idx += 1
print(f"response: {response}")
except litellm.RateLimitError as e:
pass
except Exception as e:
if "429 Quota exceeded" in str(e):
pass
else:
pytest.fail("An unexpected exception occurred - {}".format(str(e)))
def vertex_httpx_mock_reject_prompt_post(*args, **kwargs):
mock_response = MagicMock()
mock_response.status_code = 200
@ -1706,73 +1557,6 @@ async def test_gemini_pro_function_calling(provider, sync_mode):
# gemini_pro_function_calling()
@pytest.mark.parametrize("sync_mode", [True])
@pytest.mark.asyncio
@pytest.mark.flaky(retries=3, delay=1)
async def test_gemini_pro_function_calling_streaming(sync_mode):
load_vertex_ai_credentials()
litellm.set_verbose = True
data = {
"model": "vertex_ai/gemini-2.5-flash-lite",
"messages": [
{
"role": "user",
"content": "Call the submit_cities function with San Francisco and New York",
}
],
"tools": [
{
"type": "function",
"function": {
"name": "submit_cities",
"description": "Submits a list of cities",
"parameters": {
"type": "object",
"properties": {
"cities": {"type": "array", "items": {"type": "string"}}
},
"required": ["cities"],
},
},
}
],
"tool_choice": "auto",
"n": 1,
"stream": True,
"temperature": 0.1,
}
chunks = []
try:
if sync_mode == True:
response = litellm.completion(**data)
print(f"completion: {response}")
for chunk in response:
chunks.append(chunk)
assert isinstance(chunk, litellm.ModelResponseStream)
else:
response = await litellm.acompletion(**data)
print(f"completion: {response}")
assert isinstance(response, litellm.CustomStreamWrapper)
async for chunk in response:
print(f"chunk: {chunk}")
chunks.append(chunk)
assert isinstance(chunk, litellm.ModelResponseStream)
complete_response = litellm.stream_chunk_builder(chunks=chunks)
assert (
complete_response.choices[0].message.content is not None
or len(complete_response.choices[0].message.tool_calls) > 0
)
print(f"complete_response: {complete_response}")
except litellm.APIError as e:
pass
except litellm.RateLimitError as e:
pass
# asyncio.run(gemini_pro_async_function_calling())
@ -2061,55 +1845,6 @@ async def test_vertexai_multimodal_embedding_base64image_in_input():
print("Response:", response)
def test_vertexai_multimodalembedding_embedding_latest():
try:
import requests, base64
load_vertex_ai_credentials()
litellm._turn_on_debug()
response = embedding(
model="vertex_ai/multimodalembedding@001",
input=["hi"],
dimensions=128,
auto_truncate=True,
task_type="RETRIEVAL_QUERY",
)
print(f"response.usage: {response.usage}")
assert response.usage is not None
assert response.usage.prompt_tokens_details is not None
assert response._hidden_params["response_cost"] > 0
print(f"response:", response)
except litellm.RateLimitError as e:
pass
except Exception as e:
pytest.fail(f"Error occurred: {e}")
def test_vertexai_embedding_embedding_latest():
try:
load_vertex_ai_credentials()
litellm.set_verbose = True
response = embedding(
model="vertex_ai/text-embedding-004",
input=["hi"],
dimensions=1,
auto_truncate=True,
task_type="RETRIEVAL_QUERY",
)
assert len(response.data[0]["embedding"]) == 1
assert response.usage.prompt_tokens > 0
print(f"response:", response)
except litellm.RateLimitError as e:
pass
except Exception as e:
pytest.fail(f"Error occurred: {e}")
@pytest.mark.skip(reason="need to get gecko permissions on vertex ai to run this test")
@pytest.mark.flaky(retries=3, delay=1)
def test_vertexai_embedding_embedding_latest_input_type():

26
tests/local_testing/test_embedding.py
View file

@ -1,6 +1,5 @@
import json
import os
import re
import traceback
import httpx
@ -537,31 +536,6 @@ def test_bedrock_embedding_cohere():
# test_bedrock_embedding_cohere()
def test_demo_tokens_as_input_to_embeddings_fails_for_titan():
litellm.set_verbose = True
with pytest.raises(
litellm.BadRequestError,
match=re.escape(
'litellm.BadRequestError: BedrockException - {"message":"Malformed input request: '
'expected type: String, found: JSONArray, please reformat your input and try again."}'
),
):
litellm.embedding(model="amazon.titan-embed-text-v1", input=[[1]])
with pytest.raises(
litellm.BadRequestError,
match=re.escape(
'litellm.BadRequestError: BedrockException - {"message":"Malformed input request: '
'expected type: String, found: Integer, please reformat your input and try again."}'
),
):
litellm.embedding(
model="amazon.titan-embed-text-v1",
input=[1],
)
# comment out hf tests - since hf endpoints are unstable
def test_hf_embedding():
try:

220
tests/local_testing/test_function_calling.py
View file

@ -36,229 +36,9 @@ def get_current_weather(location, unit="fahrenheit"):
# In production, this could be your backend API or an external API
@pytest.mark.parametrize(
"model",
[
"gpt-6-luna",
"mistral/mistral-large-latest",
"claude-haiku-4-5-20251001",
"gemini/gemini-2.5-flash-lite",
"us.anthropic.claude-sonnet-4-5-20250929-v1:0",
],
)
@pytest.mark.flaky(retries=3, delay=1)
def test_aaparallel_function_call(model):
try:
litellm.set_verbose = True
litellm.modify_params = True
# Step 1: send the conversation and available functions to the model
messages = [
{
"role": "user",
"content": "What's the weather like in San Francisco, Tokyo, and Paris? - give me 3 responses",
}
]
tools = [
{
"type": "function",
"function": {
"name": "get_current_weather",
"description": "Get the current weather in a given location",
"parameters": {
"type": "object",
"properties": {
"location": {
"type": "string",
"description": "The city and state",
},
"unit": {
"type": "string",
"enum": ["celsius", "fahrenheit"],
},
},
"required": ["location"],
},
},
}
]
response = litellm.completion(
model=model,
messages=messages,
tools=tools,
tool_choice="auto", # auto is default, but we'll be explicit
)
print("Response\n", response)
response_message = response.choices[0].message
tool_calls = response_message.tool_calls
print("Expecting there to be 3 tool calls")
assert (
len(tool_calls) > 0
) # this has to call the function for SF, Tokyo and paris
# Step 2: check if the model wanted to call a function
print(f"tool_calls: {tool_calls}")
if tool_calls:
# Step 3: call the function
# Note: the JSON response may not always be valid; be sure to handle errors
available_functions = {
"get_current_weather": get_current_weather,
} # only one function in this example, but you can have multiple
messages.append(
response_message
) # extend conversation with assistant's reply
print("Response message\n", response_message)
# Step 4: send the info for each function call and function response to the model
for tool_call in tool_calls:
function_name = tool_call.function.name
if function_name not in available_functions:
# the model called a function that does not exist in available_functions - don't try calling anything
return
function_to_call = available_functions[function_name]
function_args = json.loads(tool_call.function.arguments)
function_response = function_to_call(
location=function_args.get("location"),
unit=function_args.get("unit"),
)
messages.append(
{
"tool_call_id": tool_call.id,
"role": "tool",
"name": function_name,
"content": function_response,
}
) # extend conversation with function response
print(f"messages: {messages}")
second_response = litellm.completion(
model=model,
messages=messages,
temperature=0.2,
seed=22,
# tools=tools,
drop_params=True,
) # get a new response from the model where it can see the function response
print("second response\n", second_response)
except litellm.InternalServerError as e:
print(e)
except litellm.RateLimitError as e:
print(e)
except Exception as e:
pytest.fail(f"Error occurred: {e}")
# test_parallel_function_call()
@pytest.mark.parametrize(
"model",
[
"anthropic/claude-haiku-4-5-20251001",
"bedrock/us.anthropic.claude-sonnet-4-5-20250929-v1:0",
],
)
@pytest.mark.flaky(retries=3, delay=1)
def test_aaparallel_function_call_with_anthropic_thinking(model):
try:
litellm._turn_on_debug()
litellm.modify_params = True
# Step 1: send the conversation and available functions to the model
messages = [
{
"role": "user",
"content": "What's the weather like in San Francisco, Tokyo, and Paris? - give me 3 responses",
}
]
tools = [
{
"type": "function",
"function": {
"name": "get_current_weather",
"description": "Get the current weather in a given location",
"parameters": {
"type": "object",
"properties": {
"location": {
"type": "string",
"description": "The city and state",
},
"unit": {
"type": "string",
"enum": ["celsius", "fahrenheit"],
},
},
"required": ["location"],
},
},
}
]
response = litellm.completion(
model=model,
messages=messages,
tools=tools,
tool_choice="auto", # auto is default, but we'll be explicit
thinking={"type": "enabled", "budget_tokens": 1024},
)
print("Response\n", response)
response_message = response.choices[0].message
tool_calls = response_message.tool_calls
print("Expecting there to be 3 tool calls")
assert (
len(tool_calls) > 0
) # this has to call the function for SF, Tokyo and paris
# Step 2: check if the model wanted to call a function
print(f"tool_calls: {tool_calls}")
if tool_calls:
# Step 3: call the function
# Note: the JSON response may not always be valid; be sure to handle errors
available_functions = {
"get_current_weather": get_current_weather,
} # only one function in this example, but you can have multiple
messages.append(
response_message
) # extend conversation with assistant's reply
print("Response message\n", response_message)
# Step 4: send the info for each function call and function response to the model
for tool_call in tool_calls:
function_name = tool_call.function.name
if function_name not in available_functions:
# the model called a function that does not exist in available_functions - don't try calling anything
return
function_to_call = available_functions[function_name]
function_args = json.loads(tool_call.function.arguments)
function_response = function_to_call(
location=function_args.get("location"),
unit=function_args.get("unit"),
)
messages.append(
{
"tool_call_id": tool_call.id,
"role": "tool",
"name": function_name,
"content": function_response,
}
) # extend conversation with function response
print(f"messages: {messages}")
second_response = litellm.completion(
model=model,
messages=messages,
seed=22,
# tools=tools,
drop_params=True,
thinking={"type": "enabled", "budget_tokens": 1024},
) # get a new response from the model where it can see the function response
print("second response\n", second_response)
## THIRD RESPONSE
except litellm.InternalServerError as e:
print(e)
except litellm.RateLimitError as e:
print(e)
except Exception as e:
pytest.fail(f"Error occurred: {e}")
from litellm.types.utils import ChatCompletionMessageToolCall, Function, Message
_PARALLEL_TOOL_HISTORY_MESSAGES = [

91
tests/logging_callback_tests/test_token_counting.py
View file

@ -1,4 +1,3 @@
import os
import traceback
from litellm._uuid import uuid
import pytest
@ -156,93 +155,3 @@ async def test_stream_token_counting_with_redaction():
assert actual_usage.total_tokens == custom_logger.recorded_usage.total_tokens
@pytest.mark.asyncio
async def test_stream_token_counting_anthropic_with_include_usage():
""" """
from anthropic import Anthropic
anthropic_client = Anthropic(api_key=os.getenv("ANTHROPIC_API_KEY"))
litellm._turn_on_debug()
custom_logger = TestCustomLogger()
litellm.logging_callback_manager.add_litellm_callback(custom_logger)
input_text = "Respond in just 1 word. Say ping"
response = await litellm.acompletion(
model="claude-sonnet-4-5-20250929",
messages=[{"role": "user", "content": input_text}],
max_tokens=4096,
stream=True,
)
actual_usage = None
output_text = ""
async for chunk in response:
output_text += chunk["choices"][0]["delta"]["content"] or ""
pass
await asyncio.sleep(1)
print("\n\n\n\n\n")
print(
"recorded_usage",
json.dumps(custom_logger.recorded_usage, indent=4, default=str),
)
print("\n\n\n\n\n")
# print making the same request with anthropic client
anthropic_response = anthropic_client.messages.create(
model="claude-sonnet-4-5-20250929",
max_tokens=4096,
messages=[{"role": "user", "content": input_text}],
stream=True,
)
usage = None
all_anthropic_usage_chunks = []
for chunk in anthropic_response:
print("chunk", json.dumps(chunk, indent=4, default=str))
if hasattr(chunk, "message"):
if chunk.message.usage:
print(
"USAGE BLOCK",
json.dumps(chunk.message.usage, indent=4, default=str),
)
all_anthropic_usage_chunks.append(chunk.message.usage)
elif hasattr(chunk, "usage"):
print("USAGE BLOCK", json.dumps(chunk.usage, indent=4, default=str))
all_anthropic_usage_chunks.append(chunk.usage)
print(
"all_anthropic_usage_chunks",
json.dumps(all_anthropic_usage_chunks, indent=4, default=str),
)
# Get the most recent value of input tokens (iterate backwards to find last non-zero value)
anthropic_api_input_tokens = 0
for usage in reversed(all_anthropic_usage_chunks):
if getattr(usage, "input_tokens", 0) > 0:
anthropic_api_input_tokens = getattr(usage, "input_tokens", 0)
break
anthropic_api_output_tokens = 0
for usage in reversed(all_anthropic_usage_chunks):
if getattr(usage, "output_tokens", 0) > 0:
anthropic_api_output_tokens = getattr(usage, "output_tokens", 0)
break
print("input_tokens_anthropic_api", anthropic_api_input_tokens)
print("output_tokens_anthropic_api", anthropic_api_output_tokens)
print("input_tokens_litellm", custom_logger.recorded_usage.prompt_tokens)
print("output_tokens_litellm", custom_logger.recorded_usage.completion_tokens)
## Assert Accuracy of token counting
# input tokens should be exactly the same
assert anthropic_api_input_tokens == custom_logger.recorded_usage.prompt_tokens
# output tokens can have at max abs diff of 10. We can't guarantee the response from two api calls will be exactly the same
assert (
abs(
anthropic_api_output_tokens - custom_logger.recorded_usage.completion_tokens
)
<= 10
)

37
tests/openai_endpoints_tests/test_bedrock_batches_api.py
View file

@ -1,37 +0,0 @@
from openai import OpenAI
import pytest
client = OpenAI(
base_url="http://0.0.0.0:4000",
api_key="sk-1234",
)
BEDROCK_BATCH_MODEL = "bedrock/batch-us.anthropic.claude-haiku-4-5-20251001-v1:0"
@pytest.mark.asyncio
async def test_bedrock_batches_api():
"""
Test bedrock batches api
E2E Test Creating a File and a Batch on Bedrock
"""
# Upload file
batch_input_file = client.files.create(
file=open("tests/openai_endpoints_tests/bedrock_batch_completions.jsonl", "rb"),
purpose="batch",
extra_body={"target_model_names": BEDROCK_BATCH_MODEL},
)
print(batch_input_file)
# Create batch
batch = client.batches.create(
input_file_id=batch_input_file.id,
endpoint="/v1/chat/completions",
completion_window="24h",
metadata={"description": "Test batch job"},
)
print(batch)
assert batch.id is not None

83
tests/openai_endpoints_tests/test_e2e_openai_responses_api.py
View file

@ -77,27 +77,6 @@ def validate_stream_chunk(chunk):
assert isinstance(chunk.created, int)
@pytest.mark.flaky(retries=3, delay=2)
def test_basic_response():
client = get_test_client()
response = client.responses.create(
model="gpt-5.5", input="just respond with the word 'ping'"
)
print("basic response=", response)
# get the response
response = client.responses.retrieve(response.id)
print("GET response=", response)
# delete the response
delete_response = client.responses.delete(response.id)
print("DELETE response=", delete_response)
# expect an error when getting the response again since it was deleted
with pytest.raises(APIStatusError):
get_response = client.responses.retrieve(response.id)
def test_streaming_response():
client = get_test_client()
stream = client.responses.create(
@ -127,39 +106,6 @@ def test_bad_request_bad_param_error():
)
def test_anthropic_with_responses_api() -> None:
client: Final = get_test_client()
response: Final = client.responses.create(
model="anthropic/claude-sonnet-5",
input="just respond with the word 'ping'",
)
assert response.status == "completed"
assert response.output_text.strip()
def test_cancel_response():
try:
client = get_test_client()
from litellm.types.llms.openai import ResponsesAPIResponse
response = client.responses.create(
model="gpt-5.5", input="just respond with the word 'ping'", background=True
)
print("basic response=", response)
# cancel the response
cancel_response = client.responses.cancel(response.id)
print("CANCEL response=", cancel_response)
# verify cancel response structure
assert hasattr(cancel_response, "id")
except Exception as e:
if "Cannot cancel a completed response" in str(e):
pass
else:
raise e
def admitted_response_id(chunk: ResponseStreamEvent) -> str | None:
response: Final = getattr(chunk, "response", None)
return None if response is None else response.id
@ -175,35 +121,6 @@ def events_until_admission(stream: Stream[ResponseStreamEvent], started: float)
return
def test_cancel_streaming_response():
client: Final = get_test_client()
started: Final = time.monotonic()
stream: Final = client.responses.create(
model="gpt-5.5",
input="count from 1 to 500, one number per line",
stream=True,
background=True,
timeout=BACKGROUND_STREAM_ADMISSION_DEADLINE_SECONDS,
)
with stream:
events: Final = tuple(events_until_admission(stream, started))
elapsed: Final = time.monotonic() - started
keepalive_events: Final = sum(1 for chunk in events if chunk.type == "keepalive")
response_id: Final = next((rid for rid in map(admitted_response_id, events) if rid is not None), None)
if response_id is None and keepalive_events:
pytest.skip(
f"OpenAI held the background stream in keepalive for {elapsed:.0f}s "
f"({keepalive_events} keepalive events) without creating the response"
)
assert response_id is not None, f"no response event within {elapsed:.0f}s of streaming a background response"
cancel_response: Final = client.responses.cancel(response_id)
print("CANCEL streaming response=", cancel_response)
assert cancel_response.status == "cancelled"
def test_cancel_invalid_response_id():
client = get_test_client()
with pytest.raises(APIStatusError):

85
tests/openai_endpoints_tests/test_openai_batches_endpoint.py
View file

@ -6,7 +6,6 @@ import aiohttp, openai
from openai import OpenAI, AsyncOpenAI
from typing import Optional, List, Union
from test_openai_files_endpoints import upload_file, delete_file
import os
import sys
import time
from unittest.mock import patch, MagicMock, AsyncMock
@ -19,54 +18,6 @@ API_KEY = "sk-1234" # Replace with your actual API key
client = OpenAI(base_url=BASE_URL, api_key=API_KEY)
@pytest.mark.asyncio
async def test_batches_operations():
_current_dir = os.path.dirname(os.path.abspath(__file__))
input_file_path = os.path.join(_current_dir, "input.jsonl")
file_obj = client.files.create(
file=open(input_file_path, "rb"),
purpose="batch",
)
batch = client.batches.create(
input_file_id=file_obj.id,
endpoint="/v1/chat/completions",
completion_window="24h",
)
assert batch.id is not None
# Test get batch
_retrieved_batch = client.batches.retrieve(batch_id=batch.id)
print("response from get batch", _retrieved_batch)
assert _retrieved_batch.id == batch.id
assert _retrieved_batch.input_file_id == file_obj.id
# Test list batches
_list_batches = client.batches.list()
print("response from list batches", _list_batches)
assert _list_batches is not None
assert len(_list_batches.data) > 0
# Clean up
# Test cancel batch
_canceled_batch = client.batches.cancel(batch_id=batch.id)
print("response from cancel batch", _canceled_batch)
assert _canceled_batch.status is not None
assert (
_canceled_batch.status == "cancelling" or _canceled_batch.status == "cancelled"
)
# finally delete the file
_deleted_file = client.files.delete(file_id=file_obj.id)
print("response from delete file", _deleted_file)
assert _deleted_file.deleted is True
def create_batch_oai_sdk(filepath: str, custom_llm_provider: str) -> str:
batch_input_file = client.files.create(
file=open(filepath, "rb"),
@ -153,42 +104,6 @@ def get_any_completed_batch_id_azure():
return None
@pytest.mark.parametrize("custom_llm_provider", ["openai"])
def test_e2e_batches_files(custom_llm_provider):
"""
[PROD Test] Ensures OpenAI Batches + files work with OpenAI SDK
"""
input_path = (
"input.jsonl" if custom_llm_provider == "openai" else "input_azure.jsonl"
)
output_path = "out.jsonl" if custom_llm_provider == "openai" else "out_azure.jsonl"
_current_dir = os.path.dirname(os.path.abspath(__file__))
input_file_path = os.path.join(_current_dir, input_path)
output_file_path = os.path.join(_current_dir, output_path)
print("running e2e batches files with custom_llm_provider=", custom_llm_provider)
batch_id = create_batch_oai_sdk(
filepath=input_file_path, custom_llm_provider=custom_llm_provider
)
if custom_llm_provider == "azure":
# azure takes very long to complete a batch
return
else:
response_batch_id = await_batch_completion(
batch_id=batch_id, custom_llm_provider=custom_llm_provider
)
if response_batch_id is None:
return
write_content_to_file(
batch_id=batch_id,
output_path=output_file_path,
custom_llm_provider=custom_llm_provider,
)
read_jsonl(output_file_path)
@pytest.mark.skip(reason="Local only test to verify if things work well")
def test_vertex_batches_endpoint():
"""

12
tests/pass_through_unit_tests/messages_api_structured_output/__init__.py
View file

@ -1,12 +0,0 @@
"""
Anthropic Messages API Structured Outputs Test Suite
E2E tests for structured outputs functionality across different providers:
- Direct Anthropic API
- Azure AI Foundry Anthropic models
- AWS Bedrock Invoke API
- AWS Bedrock Converse API
All tests validate that the output_format parameter works correctly
and returns valid JSON instead of Markdown text.
"""

135
tests/pass_through_unit_tests/messages_api_structured_output/base_anthropic_messages_structured_output_test.py
View file

@ -1,135 +0,0 @@
"""
Base test class for Anthropic Messages API structured outputs E2E tests.
Tests that structured outputs work correctly via litellm.anthropic.messages interface
by making actual API calls and validating JSON response format.
"""
import json
from abc import ABC, abstractmethod
from typing import Any, Dict, List, Optional
import pytest
import litellm
class BaseAnthropicMessagesStructuredOutputTest(ABC):
"""
Base test class for structured outputs E2E tests across different providers.
Subclasses must implement:
- get_model(): Returns the model string to use for tests
Subclasses may optionally implement:
- get_api_base(): Returns the API base URL (for Azure, etc.)
- get_api_key(): Returns the API key (for Azure, etc.)
"""
@abstractmethod
def get_model(self) -> str:
"""
Returns the model string to use for tests.
"""
pass
def get_api_base(self) -> Optional[str]:
"""
Returns the API base URL. Override for providers like Azure.
"""
return None
def get_api_key(self) -> Optional[str]:
"""
Returns the API key. Override for providers like Azure.
"""
return None
def get_output_format_schema(self) -> Dict[str, Any]:
"""
Returns a simple JSON schema for testing structured outputs.
"""
return {
"type": "json_schema",
"schema": {
"type": "object",
"properties": {
"sentiment": {
"type": "string",
"enum": ["positive", "negative", "neutral"],
}
},
"required": ["sentiment"],
"additionalProperties": False,
},
}
def get_test_messages(self) -> List[Dict[str, Any]]:
"""
Returns test messages for structured output testing.
"""
return [
{
"role": "user",
"content": "What is the sentiment of this text: 'This product is amazing!' Return only the sentiment.",
}
]
@pytest.mark.asyncio
async def test_structured_output_e2e(self):
"""
E2E test: Make actual API call with structured output and validate JSON response.
"""
litellm._turn_on_debug()
messages = self.get_test_messages()
output_format = self.get_output_format_schema()
# Build kwargs with optional api_base and api_key
kwargs: Dict[str, Any] = {
"model": self.get_model(),
"messages": messages,
"max_tokens": 100,
"output_format": output_format,
}
api_base = self.get_api_base()
if api_base:
kwargs["api_base"] = api_base
api_key = self.get_api_key()
if api_key:
kwargs["api_key"] = api_key
response = await litellm.anthropic.messages.acreate(**kwargs)
print(f"Response: {response}")
# Validate response structure - handle both dict and object responses
if isinstance(response, dict):
assert "content" in response
content_list = response["content"]
else:
assert hasattr(response, "content")
content_list = response.content
assert len(content_list) > 0
content = content_list[0]
# Handle both dict and object content blocks
if isinstance(content, dict):
assert "text" in content
response_text = content["text"]
else:
assert hasattr(content, "text")
response_text = content.text
print(f"Response text: {response_text}")
# The response should be valid JSON
parsed_json = json.loads(response_text)
print(f"Parsed JSON: {parsed_json}")
# Validate the JSON structure
assert "sentiment" in parsed_json
assert parsed_json["sentiment"] in ["positive", "negative", "neutral"]

26
tests/pass_through_unit_tests/messages_api_structured_output/test_anthropic_api_structured_output.py
View file

@ -1,26 +0,0 @@
"""
E2E Test suite for Anthropic API structured outputs via litellm.anthropic.messages.
Tests that structured outputs work correctly with direct Anthropic API calls
by making actual API calls and validating JSON response format.
Requires ANTHROPIC_API_KEY environment variable.
"""
from .base_anthropic_messages_structured_output_test import (
BaseAnthropicMessagesStructuredOutputTest,
)
class TestAnthropicAPIStructuredOutput(BaseAnthropicMessagesStructuredOutputTest):
"""
E2E tests for structured outputs with direct Anthropic API.
Uses Claude Sonnet 4.5 which supports structured outputs with the
'anthropic-beta: structured-outputs-2025-11-13' header.
"""
def get_model(self) -> str:
return "claude-sonnet-4-5-20250929"

34
tests/pass_through_unit_tests/messages_api_structured_output/test_azure_anthropic_structured_output.py
View file

@ -1,34 +0,0 @@
"""
E2E Test suite for Azure Anthropic structured outputs via litellm.anthropic.messages.
Tests that structured outputs work correctly with Azure AI Foundry Anthropic models
by making actual API calls and validating JSON response format.
Requires Azure AI credentials and model deployment.
"""
import os
from typing import Optional
from .base_anthropic_messages_structured_output_test import (
BaseAnthropicMessagesStructuredOutputTest,
)
class TestAzureAnthropicStructuredOutput(BaseAnthropicMessagesStructuredOutputTest):
"""
E2E tests for structured outputs with Azure AI Foundry Anthropic models.
Uses the azure_ai/ prefix which routes through Azure AI Foundry
while maintaining the Anthropic Messages API format.
"""
def get_model(self) -> str:
return "azure_ai/claude-opus-4-5"
def get_api_base(self) -> Optional[str]:
return "https://krris-mnb3t0vd-swedencentral.services.ai.azure.com"
def get_api_key(self) -> Optional[str]:
return os.environ.get("AZURE_ANTHROPIC_API_KEY")

26
tests/pass_through_unit_tests/messages_api_structured_output/test_bedrock_converse_structured_output.py
View file

@ -1,26 +0,0 @@
"""
E2E Test suite for Bedrock Converse API structured outputs via litellm.anthropic.messages.
Tests that structured outputs work correctly with Bedrock Converse API
by making actual API calls and validating JSON response format.
Requires AWS credentials and Bedrock model access.
"""
from .base_anthropic_messages_structured_output_test import (
BaseAnthropicMessagesStructuredOutputTest,
)
class TestBedrockConverseStructuredOutput(BaseAnthropicMessagesStructuredOutputTest):
"""
E2E tests for structured outputs with Bedrock Converse API.
Uses the bedrock/converse/ prefix which routes through litellm.completion()
and the AmazonConverseConfig transformation.
"""
def get_model(self) -> str:
return "bedrock/converse/us.anthropic.claude-haiku-4-5-20251001-v1:0"

29
tests/pass_through_unit_tests/messages_api_structured_output/test_bedrock_invoke_structured_output.py
View file

@ -1,29 +0,0 @@
"""
E2E Test suite for Bedrock Invoke API structured outputs via litellm.anthropic.messages.
Tests that structured outputs work correctly with Bedrock Invoke API (native Anthropic format)
by making actual API calls and validating JSON response format.
Requires AWS credentials and Bedrock model access.
"""
import pytest
from .base_anthropic_messages_structured_output_test import (
BaseAnthropicMessagesStructuredOutputTest,
)
@pytest.mark.skip(reason="Skipping Bedrock Invoke structured output tests")
class TestBedrockInvokeStructuredOutput(BaseAnthropicMessagesStructuredOutputTest):
"""
E2E tests for structured outputs with Bedrock Invoke API.
Uses the bedrock/invoke/ prefix which routes through the native
Anthropic Messages API format on Bedrock.
"""
def get_model(self) -> str:
return "bedrock/invoke/us.anthropic.claude-haiku-4-5-20251001-v1:0"

16
tests/test_fallbacks.py
View file

@ -82,22 +82,6 @@ async def chat_completion(
return await response.json()
@pytest.mark.asyncio
async def test_chat_completion():
"""
make chat completion call with prompt > context window. expect it to work with fallback
"""
async with aiohttp.ClientSession() as session:
model = "gpt-3.5-turbo"
messages = [
{"role": "system", "content": text},
{"role": "user", "content": "Who was Alexander?"},
]
await chat_completion(
session=session, key="sk-1234", model=model, messages=messages
)
@pytest.mark.parametrize("has_access", [True, False])
@pytest.mark.asyncio
async def test_chat_completion_client_fallbacks(has_access: bool) -> None:

82
tests/test_openai_endpoints.py
View file

@ -458,71 +458,6 @@ async def test_chat_completion_anthropic_structured_output():
print(message.parsed.events)
@pytest.mark.asyncio
async def test_completion():
"""
- Create key
Make chat completion call
- Create user
make chat completion call
"""
async with aiohttp.ClientSession() as session:
key_gen = await generate_key(session=session)
key = key_gen["key"]
await completion(session=session, key=key)
key_gen = await new_user(session=session)
key_2 = key_gen["key"]
# response = await completion(session=session, key=key_2)
## validate openai format ##
client = OpenAI(api_key=key_2, base_url="http://0.0.0.0:4000")
client.completions.create(
model="gpt-4",
prompt="Say this is a test",
max_tokens=7,
temperature=0,
)
@pytest.mark.asyncio
async def test_embeddings():
"""
- Create key
Make embeddings call
- Create user
make embeddings call
"""
async with aiohttp.ClientSession() as session:
key_gen = await generate_key(session=session)
key = key_gen["key"]
await embeddings(session=session, key=key)
key_gen = await new_user(session=session)
key_2 = key_gen["key"]
await embeddings(session=session, key=key_2)
# embedding request with non OpenAI model
await embeddings(session=session, key=key, model="mistral-embed")
@pytest.mark.flaky(retries=5, delay=1)
@pytest.mark.asyncio
async def test_image_generation():
"""
- Create key
Make embeddings call
- Create user
make embeddings call
"""
async with aiohttp.ClientSession() as session:
key_gen = await generate_key(session=session)
key = key_gen["key"]
await image_generation(session=session, key=key)
key_gen = await new_user(session=session)
key_2 = key_gen["key"]
await image_generation(session=session, key=key_2)
@pytest.mark.asyncio
async def test_proxy_all_models():
"""
@ -566,20 +501,3 @@ async def test_batch_chat_completions():
assert isinstance(response, list)
@pytest.mark.asyncio
async def test_moderations_endpoint():
"""
- Make chat completion call using
"""
async with aiohttp.ClientSession() as session:
# call chat/completions with a model that the key was not created for + the model is not on the config.yaml
response = await moderation(
session=session,
key="sk-1234",
)
print(f"response: {response}")
assert "results" in response

25
tests/unit/realtime_api/test_main.py
View file

@ -25,6 +25,9 @@ class FakeLogging:
def update_from_kwargs(self, **kwargs):
pass
def pre_call(self, **kwargs):
pass
def test_resolves_top_level_session_model():
resolved = _with_resolved_session_model({"model": "alias/gpt-realtime"}, "gpt-realtime")
@ -574,3 +577,25 @@ async def test_arealtime_keeps_gemini_live_on_the_vertex_realtime_websocket(monk
async def test_realtime_health_check_names_the_batch_mode_for_chirp_models():
with pytest.raises(ValueError, match="mode audio_transcription"):
await realtime_main._realtime_health_check(model="chirp_3", custom_llm_provider="vertex_ai", api_key=None)
class _ClosableGaClientWebSocket:
def __init__(self) -> None:
self.scope: Final = {"headers": ()}
async def close(self, code: int = 1000, reason: str = "") -> None:
return None
@pytest.mark.asyncio
async def test_arealtime_openai_forwards_the_intent_query_param_to_the_upstream_url():
connect: Final = _ConnectThatStopsAfterCapturingTheUrl()
with patch("websockets.connect", connect):
await realtime_main._arealtime.__wrapped__(
model="openai/gpt-realtime",
websocket=_ClosableGaClientWebSocket(),
api_key="fake-key",
query_params={"model": "openai/gpt-realtime", "intent": "chat"},
litellm_logging_obj=FakeLogging(),
)
assert connect.url == "wss://api.openai.com/v1/realtime?model=gpt-realtime&intent=chat"