scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-09-25 01:02:15 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 81

test(e2e): replace custom endpoints_client with provider SDK clients

Browse source 
The llm_translation suite drove /v1/responses, /v1/messages, /embeddings,
/v1/images/generations, /v1/moderations and /v1/audio/* through a bespoke
endpoints_client wrapper that no customer runs. Tests now call the proxy the
way customers do: the OpenAI SDK for the OpenAI-compatible surface and the
Anthropic SDK for /v1/messages, wired through a session-scoped sdk fixture
(sdk_clients.py) that points both SDKs at the proxy with a virtual key.

Endpoints no official SDK covers keep the shared typed transport: rerank moves
onto ProxyClient (RerankBody/RerankResponse in models.py) and the passthrough
header test parses with the shared AnthropicMessagesResponse model. Files that
only used endpoints_client for model registration now use the proxy fixture
directly. endpoints_client.py is deleted; anthropic joins the e2e-dev
dependency group so the lint env resolves the SDK imports.

Resolves LIT-4577
This commit is contained in:
mateo-berri 2026-07-23 03:43:01 +00:00
parent eb2dce8771
commit 76bf0cd579
No known key found for this signature in database
26 changed files with 828 additions and 1200 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

1
pyproject.toml
View file

@ -198,6 +198,7 @@ e2e-dev = [
"playwright==1.61.0", "playwright==1.61.0",
"websockets>=15.0.1,<16.0", "websockets>=15.0.1,<16.0",
"locust==2.45.0", "locust==2.45.0",
"anthropic==0.84.0",
] ]
proxy-dev = [ proxy-dev = [
"prisma==0.11.0", "prisma==0.11.0",

2
tests/e2e/CLAUDE.md
View file

@ -58,6 +58,8 @@ That snippet only conveys intent. What you actually write uses the real harness:
Every HTTP call goes through the shared transport, never through `requests.*` in a test. `e2e_http.py` is the only module permitted to call `requests.*`, and that is enforced in CI by `tests/code_coverage_tests/check_e2e_no_raw_requests.py`. A test that imports requests will fail the check Every HTTP call goes through the shared transport, never through `requests.*` in a test. `e2e_http.py` is the only module permitted to call `requests.*`, and that is enforced in CI by `tests/code_coverage_tests/check_e2e_no_raw_requests.py`. A test that imports requests will fail the check
One deliberate exception: LLM-endpoint calls in `llm_translation/` go through the real provider SDKs (OpenAI, Anthropic) via the suite's `sdk` fixture (`llm_translation/sdk_clients.py`), because that is what customers actually run against the proxy (LIT-4577). The SDKs raise their own typed exceptions on failure, which is exactly the customer-observable contract; management routes (model/key CRUD, spend read-back) and endpoints no official SDK covers (e.g. `/v1/rerank`, `/v1/ocr`, custom passthrough paths) stay on the shared transport. Raw HTTP client imports remain banned either way
The shape is layered so tests stay declarative The shape is layered so tests stay declarative
`transport.py` exposes a `Transport` Protocol with `post`, `get`, `delete`, `send`, `stream`, `probe`, plus `bearer(key)` and the `master` header. `HttpTransport` fulfils it, and `SplitTransport` routes each call by path to the data plane or the control plane so a split control-plane/data-plane deployment works without any change in the test `transport.py` exposes a `Transport` Protocol with `post`, `get`, `delete`, `send`, `stream`, `probe`, plus `bearer(key)` and the `master` header. `HttpTransport` fulfils it, and `SplitTransport` routes each call by path to the data plane or the control plane so a split control-plane/data-plane deployment works without any change in the test

2
tests/e2e/CONTRIBUTING.md
View file

@ -122,6 +122,8 @@ That snippet only conveys intent. What you actually write uses the real harness:
Every HTTP call goes through the shared transport, never through `requests.*` in a test. `e2e_http.py` is the only module permitted to call `requests.*`, and that is enforced in CI by `tests/code_coverage_tests/check_e2e_no_raw_requests.py`. A test that imports requests will fail the check Every HTTP call goes through the shared transport, never through `requests.*` in a test. `e2e_http.py` is the only module permitted to call `requests.*`, and that is enforced in CI by `tests/code_coverage_tests/check_e2e_no_raw_requests.py`. A test that imports requests will fail the check
One deliberate exception: LLM-endpoint calls in `llm_translation/` go through the real provider SDKs (OpenAI, Anthropic) via the suite's `sdk` fixture (`llm_translation/sdk_clients.py`), because that is what customers actually run against the proxy (LIT-4577). Management routes and endpoints no official SDK covers stay on the shared transport, and raw HTTP client imports remain banned either way
The shape is layered so tests stay declarative The shape is layered so tests stay declarative
`transport.py` exposes a `Transport` Protocol with `post`, `get`, `delete`, `send`, `stream`, `probe`, plus `bearer(key)` and the `master` header. `HttpTransport` fulfils it, and `SplitTransport` routes each call by path to the data plane or the control plane so a split control-plane/data-plane deployment works without any change in the test `transport.py` exposes a `Transport` Protocol with `post`, `get`, `delete`, `send`, `stream`, `probe`, plus `bearer(key)` and the `master` header. `HttpTransport` fulfils it, and `SplitTransport` routes each call by path to the data plane or the control plane so a split control-plane/data-plane deployment works without any change in the test

10
tests/e2e/llm_translation/conftest.py
View file

@ -2,14 +2,16 @@
The shared lifecycle (resources/scoped_key), proxy liveness gate, and e2e marker The shared lifecycle (resources/scoped_key), proxy liveness gate, and e2e marker
live in the parent tests/e2e/conftest.py. PassthroughClient holds the shared live in the parent tests/e2e/conftest.py. PassthroughClient holds the shared
ProxyClient, so the `resources` fixture cleans up keys this suite creates. ProxyClient, so the `resources` fixture cleans up keys this suite creates. The
`sdk` fixture hands tests real provider SDK clients (OpenAI, Anthropic) pointed
at the proxy, the way customers actually call it.
""" """
import pytest import pytest
from endpoints_client import EndpointsClient, build_endpoints_client
from passthrough_client import PassthroughClient, build_client from passthrough_client import PassthroughClient, build_client
from proxy_client import ProxyClient from proxy_client import ProxyClient
from sdk_clients import SdkClients, build_sdk_clients
def pytest_configure(config: pytest.Config) -> None: def pytest_configure(config: pytest.Config) -> None:
@ -25,5 +27,5 @@ def client(proxy: ProxyClient) -> PassthroughClient:
@pytest.fixture(scope="session") @pytest.fixture(scope="session")
def endpoints_client(proxy: ProxyClient) -> EndpointsClient: def sdk() -> SdkClients:
return build_endpoints_client(proxy) return build_sdk_clients()

385
tests/e2e/llm_translation/endpoints_client.py
View file

@ -1,385 +0,0 @@
"""Client for the non-chat inference endpoints (responses, messages, rerank,
embeddings, audio speech, image generation).
Each test registers the deployment it needs through /model/new (deleted on
teardown), so nothing is hardcoded into the gateway config, then drives the
endpoint with `send` and parses the provider-native body with a suite-local model
so the assertion is on real content, not just a 200.
"""
from __future__ import annotations
from dataclasses import dataclass
from typing import Literal
from pydantic import BaseModel
from proxy_client import ProxyClient
from e2e_http import BinaryStream, Result, StreamingResponse
from models import CacheControl, ChatMessage, LiteLLMParamsBody, RichMessage, TextBlock
__all__ = [
"CacheControl",
"RichMessage",
"TextBlock",
]
class FunctionParameterProperty(BaseModel):
type: str
description: str | None = None
class FunctionParameters(BaseModel):
type: Literal["object"] = "object"
properties: dict[str, FunctionParameterProperty]
required: list[str] = []
class ResponsesFunctionTool(BaseModel):
type: Literal["function"] = "function"
name: str
description: str | None = None
parameters: FunctionParameters
class ResponsesInputTextPart(BaseModel):
type: Literal["input_text"] = "input_text"
text: str
class ResponsesInputImagePart(BaseModel):
type: Literal["input_image"] = "input_image"
image_url: str
ResponsesInputContentPart = ResponsesInputTextPart | ResponsesInputImagePart
class ResponsesInputMessage(BaseModel):
role: Literal["user", "assistant", "system"] = "user"
content: list[ResponsesInputContentPart]
ResponsesInput = str | list[ResponsesInputMessage]
class ResponsesRequest(BaseModel):
model: str
input: ResponsesInput
instructions: str | None = None
stream: bool = False
tools: list[ResponsesFunctionTool] | None = None
class MessagesRequest(BaseModel):
model: str
max_tokens: int
messages: list[ChatMessage]
class RichMessagesRequest(BaseModel):
model: str
max_tokens: int = 64
system: list[TextBlock]
messages: list[RichMessage]
class EmbeddingsRequest(BaseModel):
model: str
input: str
class RerankRequest(BaseModel):
model: str
query: str
documents: list[str]
top_n: int
class SpeechRequest(BaseModel):
model: str
input: str
voice: str
class ImageRequest(BaseModel):
model: str
prompt: str
n: int = 1
size: str = "1024x1024"
class TranscriptionForm(BaseModel):
model: str
response_format: str = "json"
class ModerationRequest(BaseModel):
model: str
input: str
class ResponsesOutputContent(BaseModel):
type: str | None = None
text: str | None = None
class ResponsesOutputItem(BaseModel):
type: str | None = None
content: list[ResponsesOutputContent] = []
name: str | None = None
arguments: str | None = None
call_id: str | None = None
class ResponsesResult(BaseModel):
id: str | None = None
status: str | None = None
model: str | None = None
output: list[ResponsesOutputItem] = []
@property
def text(self) -> str:
return "".join(
content.text or "" for item in self.output for content in item.content
)
@property
def function_calls(self) -> tuple[ResponsesOutputItem, ...]:
return tuple(
item
for item in self.output
if item.type == "function_call"
and item.name is not None
and item.arguments is not None
)
class ResponsesStreamEvent(BaseModel):
event_id: str | None = None
class ResponsesStreamEventType(BaseModel):
type: str
class ResponsesOutputTextDeltaEvent(ResponsesStreamEvent):
type: Literal["response.output_text.delta"]
delta: str
class AnthropicContentBlock(BaseModel):
type: str | None = None
text: str | None = None
class MessagesUsage(BaseModel):
input_tokens: int = 0
output_tokens: int = 0
cache_creation_input_tokens: int = 0
cache_read_input_tokens: int = 0
class MessagesResult(BaseModel):
id: str | None = None
role: str | None = None
model: str | None = None
content: list[AnthropicContentBlock] = []
usage: MessagesUsage = MessagesUsage()
@property
def text(self) -> str:
return "".join(block.text or "" for block in self.content)
class EmbeddingItem(BaseModel):
embedding: list[float] = []
class EmbeddingsResult(BaseModel):
data: list[EmbeddingItem] = []
@property
def first_vector(self) -> tuple[float, ...]:
return tuple(self.data[0].embedding) if self.data else ()
class RerankItem(BaseModel):
index: int | None = None
relevance_score: float | None = None
class RerankResult(BaseModel):
results: list[RerankItem] = []
class ImageItem(BaseModel):
url: str | None = None
b64_json: str | None = None
class ImagesResult(BaseModel):
data: list[ImageItem] = []
class TranscriptionResult(BaseModel):
text: str = ""
class ModerationResultItem(BaseModel):
flagged: bool
categories: dict[str, bool] = {}
@property
def flagged_categories(self) -> tuple[str, ...]:
return tuple(name for name, hit in self.categories.items() if hit)
class ModerationResult(BaseModel):
results: list[ModerationResultItem] = []
@property
def first(self) -> ModerationResultItem | None:
return self.results[0] if self.results else None
@dataclass(frozen=True, slots=True)
class EndpointsClient:
proxy: ProxyClient
def create_model(self, model_name: str, litellm_params: LiteLLMParamsBody) -> str:
return self.proxy.create_model(model_name, litellm_params)
def delete_model(self, model_id: str) -> None:
self.proxy.delete_model(model_id)
def _send(
self, path: str, key: str, body: BaseModel, *, stream: bool = False
) -> StreamingResponse:
return self.proxy.transport.send(
path,
headers=self.proxy.transport.bearer(key),
json=body,
stream=stream,
)
def responses(
self, key: str, model: str, text: str, *, stream: bool = False
) -> StreamingResponse:
return self._send(
"/v1/responses",
key,
ResponsesRequest(
model=model,
input=text,
instructions="You are a helpful assistant",
stream=stream,
),
stream=stream,
)
def responses_vision(
self, key: str, model: str, text: str, image_url: str
) -> StreamingResponse:
return self._send(
"/v1/responses",
key,
ResponsesRequest(
model=model,
input=[
ResponsesInputMessage(
content=[
ResponsesInputTextPart(text=text),
ResponsesInputImagePart(image_url=image_url),
]
)
],
instructions="You are a helpful assistant",
),
)
def responses_with_tools(
self, key: str, model: str, text: str, tools: list[ResponsesFunctionTool]
) -> StreamingResponse:
return self._send(
"/v1/responses",
key,
ResponsesRequest(
model=model,
input=text,
instructions="You are a helpful assistant",
tools=tools,
),
)
def messages(
self, key: str, model: str, text: str, *, max_tokens: int = 64
) -> StreamingResponse:
return self._send(
"/v1/messages",
key,
MessagesRequest(
model=model,
max_tokens=max_tokens,
messages=[ChatMessage(role="user", content=text)],
),
)
def embeddings(self, key: str, model: str, text: str) -> StreamingResponse:
return self._send("/embeddings", key, EmbeddingsRequest(model=model, input=text))
def rerank(
self, key: str, model: str, query: str, documents: list[str], top_n: int
) -> StreamingResponse:
return self._send(
"/v1/rerank",
key,
RerankRequest(model=model, query=query, documents=documents, top_n=top_n),
)
def audio_speech(
self, key: str, model: str, text: str, *, voice: str = "alloy"
) -> StreamingResponse:
return self._send(
"/v1/audio/speech", key, SpeechRequest(model=model, input=text, voice=voice)
)
def audio_speech_stream(
self, key: str, model: str, text: str, *, voice: str = "alloy"
) -> BinaryStream:
return self.proxy.transport.stream_binary(
"/v1/audio/speech",
headers=self.proxy.transport.bearer(key),
json=SpeechRequest(model=model, input=text, voice=voice),
)
def transcribe(
self, key: str, model: str, *, filename: str, content: bytes
) -> Result[TranscriptionResult]:
return self.proxy.transport.upload(
"/v1/audio/transcriptions",
headers=self.proxy.transport.bearer(key),
form=TranscriptionForm(model=model),
filename=filename,
content=content,
file_content_type="audio/wav",
response_type=TranscriptionResult,
)
def moderations(self, key: str, model: str, text: str) -> Result[ModerationResult]:
return self.proxy.transport.post(
"/v1/moderations",
headers=self.proxy.transport.bearer(key),
json=ModerationRequest(model=model, input=text),
response_type=ModerationResult,
)
def images(self, key: str, model: str, prompt: str) -> StreamingResponse:
return self._send(
"/v1/images/generations", key, ImageRequest(model=model, prompt=prompt)
)
def build_endpoints_client(proxy: ProxyClient) -> EndpointsClient:
return EndpointsClient(proxy=proxy)

53
tests/e2e/llm_translation/sdk_clients.py Normal file
View file

@ -0,0 +1,53 @@
"""Real provider SDK clients pointed at the proxy, connected the way customers
connect (LIT-4577).
The OpenAI SDK drives the OpenAI-compatible surface (/responses, /embeddings,
/images/generations, /moderations, /audio/*) and the Anthropic SDK drives
/v1/messages, each authenticated with a litellm virtual key. Errors surface as
the SDK's own exceptions, exactly what an end user sees. Retries are disabled
so a proxy fault fails the test instead of being papered over, and the timeout
matches the shared transport's request budget.
"""
from __future__ import annotations
from collections.abc import Mapping
from dataclasses import dataclass
from anthropic import Anthropic
from openai import OpenAI
from e2e_config import PROXY_BASE_URL, REQUEST_TIMEOUT
def response_header(headers: Mapping[str, str], name: str) -> str | None:
"""Typed read of an SDK response header: httpx.Headers.get returns Any and
httpx itself is a banned import in suite code, so tests read headers through
the Mapping[str, str] interface Headers fulfils."""
return headers[name] if name in headers else None
@dataclass(frozen=True, slots=True)
class SdkClients:
base_url: str
request_timeout: float
def openai(self, key: str) -> OpenAI:
return OpenAI(
base_url=self.base_url,
api_key=key,
timeout=self.request_timeout,
max_retries=0,
)
def anthropic(self, key: str) -> Anthropic:
return Anthropic(
base_url=self.base_url,
api_key=key,
timeout=self.request_timeout,
max_retries=0,
)
def build_sdk_clients() -> SdkClients:
return SdkClients(base_url=PROXY_BASE_URL, request_timeout=REQUEST_TIMEOUT)

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

@ -1,9 +1,10 @@
"""Live e2e: POST /v1/audio/speech returns audio, non-streamed and streamed. """Live e2e: POST /v1/audio/speech returns audio, non-streamed and streamed.
The non-streamed call asserts an audio (not JSON) body. The streamed call consumes Both calls go through the real OpenAI SDK (LIT-4577). The non-streamed call
the response the way a player would and asserts customer-observable streaming: asserts an audio (not JSON) body. The streamed call consumes the response the
chunked transfer encoding (a buffered body would carry a content-length) with way a player would and asserts customer-observable streaming: chunked transfer
non-zero audio bytes. encoding (a buffered body would carry a content-length) with non-zero audio
bytes.
""" """
from __future__ import annotations from __future__ import annotations
@ -11,68 +12,70 @@ from __future__ import annotations
import pytest import pytest
from e2e_config import unique_marker from e2e_config import unique_marker
from e2e_http import require_successful_call
from endpoints_client import EndpointsClient
from lifecycle import ResourceManager from lifecycle import ResourceManager
from models import LiteLLMParamsBody from models import LiteLLMParamsBody
from proxy_client import ProxyClient
from sdk_clients import SdkClients, response_header
pytestmark = pytest.mark.e2e pytestmark = pytest.mark.e2e
def _register(proxy: ProxyClient, resources: ResourceManager, prefix: str) -> str:
model = f"{prefix}-{unique_marker()}"
model_id = proxy.create_model(
model,
LiteLLMParamsBody(model="openai/gpt-4o-mini-tts", api_key="os.environ/OPENAI_API_KEY"),
)
resources.defer(lambda: proxy.delete_model(model_id))
return model
class TestAudioSpeech: class TestAudioSpeech:
@pytest.mark.covers("llm.audio_speech.openai.basic.nonstream.works") @pytest.mark.covers("llm.audio_speech.openai.basic.nonstream.works")
def test_audio_speech_returns_audio( def test_audio_speech_returns_audio(
self, endpoints_client: EndpointsClient, resources: ResourceManager self, proxy: ProxyClient, resources: ResourceManager, sdk: SdkClients
) -> None: ) -> None:
model = f"e2e-speech-{unique_marker()}" model = _register(proxy, resources, "e2e-speech")
model_id = endpoints_client.create_model( client = sdk.openai(resources.key())
model,
LiteLLMParamsBody(
model="openai/gpt-4o-mini-tts", api_key="os.environ/OPENAI_API_KEY"
),
)
resources.defer(lambda: endpoints_client.delete_model(model_id))
key = resources.key()
result = endpoints_client.audio_speech(key, model, "Hello!") response = client.audio.speech.with_raw_response.create(
require_successful_call(result) model=model, voice="alloy", input="Hello!"
assert "audio" in (result.content_type or ""), (
f"/audio/speech content-type is not audio: {result.content_type!r}"
) )
assert result.body, "/audio/speech returned an empty body" content_type = response_header(response.headers, "content-type")
assert "audio" in (content_type or ""), (
f"/audio/speech content-type is not audio: {content_type!r}"
)
assert response.content, "/audio/speech returned an empty body"
@pytest.mark.covers("llm.audio_speech.openai.basic.stream.works") @pytest.mark.covers("llm.audio_speech.openai.basic.stream.works")
def test_audio_speech_streams_audio_chunks( def test_audio_speech_streams_audio_chunks(
self, endpoints_client: EndpointsClient, resources: ResourceManager self, proxy: ProxyClient, resources: ResourceManager, sdk: SdkClients
) -> None: ) -> None:
model = f"e2e-speech-stream-{unique_marker()}" model = _register(proxy, resources, "e2e-speech-stream")
model_id = endpoints_client.create_model( client = sdk.openai(resources.key())
model,
LiteLLMParamsBody(
model="openai/gpt-4o-mini-tts", api_key="os.environ/OPENAI_API_KEY"
),
)
resources.defer(lambda: endpoints_client.delete_model(model_id))
key = resources.key()
result = endpoints_client.audio_speech_stream( with client.audio.speech.with_streaming_response.create(
key, model=model,
model, voice="alloy",
"Streaming speech should arrive in several audio chunks so a client can " input=(
"begin playback well before the whole clip has finished generating.", "Streaming speech should arrive in several audio chunks so a client can "
"begin playback well before the whole clip has finished generating."
),
) as response:
content_type = response_header(response.headers, "content-type")
transfer_encoding = response_header(response.headers, "transfer-encoding")
content_length = response_header(response.headers, "content-length")
total_bytes = sum(len(chunk) for chunk in response.iter_bytes(chunk_size=8192))
assert "audio" in (content_type or ""), (
f"/audio/speech content-type is not audio: {content_type!r}"
) )
assert result.ok, ( assert "chunked" in (transfer_encoding or ""), (
f"/audio/speech stream failed (status {result.status_code}); body={result.error_body}" f"/audio/speech did not stream: transfer-encoding={transfer_encoding!r}, "
f"content-length={content_length!r} (a buffered body is not a stream)"
) )
assert "audio" in (result.content_type or ""), ( assert content_length is None, (
f"/audio/speech content-type is not audio: {result.content_type!r}" f"/audio/speech advertised content-length={content_length!r} on a "
)
assert result.chunked, (
f"/audio/speech did not stream: transfer-encoding={result.transfer_encoding!r}, "
f"content-length={result.content_length!r} (a buffered body is not a stream)"
)
assert result.content_length is None, (
f"/audio/speech advertised content-length={result.content_length!r} on a "
f"streamed response (a buffered body is not a stream)" f"streamed response (a buffered body is not a stream)"
) )
assert result.total_bytes > 0, "/audio/speech stream returned no audio bytes" assert total_bytes > 0, "/audio/speech stream returned no audio bytes"

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

@ -1,8 +1,9 @@
"""Live e2e: POST /v1/audio/transcriptions turns speech into text. """Live e2e: POST /v1/audio/transcriptions turns speech into text.
Registers an OpenAI speech-to-text deployment at runtime and uploads a spoken Registers an OpenAI speech-to-text deployment at runtime and uploads a spoken
weather question (the realtime suite's 24kHz WAV fixture) as multipart, asserting weather question (the realtime suite's 24kHz WAV fixture) through the real
the returned transcript is non-empty and mentions the word it was asked about. OpenAI SDK (LIT-4577), asserting the returned transcript is non-empty and
mentions the word it was asked about.
""" """
from __future__ import annotations from __future__ import annotations
@ -12,10 +13,10 @@ from pathlib import Path
import pytest import pytest
from e2e_config import unique_marker from e2e_config import unique_marker
from e2e_http import unwrap
from endpoints_client import EndpointsClient
from lifecycle import ResourceManager from lifecycle import ResourceManager
from models import LiteLLMParamsBody from models import LiteLLMParamsBody
from proxy_client import ProxyClient
from sdk_clients import SdkClients
pytestmark = pytest.mark.e2e pytestmark = pytest.mark.e2e
@ -27,24 +28,22 @@ WEATHER_WAV = (
class TestAudioTranscriptions: class TestAudioTranscriptions:
@pytest.mark.covers("llm.audio_transcriptions.openai.basic.nonstream.works") @pytest.mark.covers("llm.audio_transcriptions.openai.basic.nonstream.works")
def test_audio_transcriptions_returns_text( def test_audio_transcriptions_returns_text(
self, endpoints_client: EndpointsClient, resources: ResourceManager self, proxy: ProxyClient, resources: ResourceManager, sdk: SdkClients
) -> None: ) -> None:
model = f"e2e-transcribe-{unique_marker()}" model = f"e2e-transcribe-{unique_marker()}"
model_id = endpoints_client.create_model( model_id = proxy.create_model(
model, model,
LiteLLMParamsBody( LiteLLMParamsBody(
model="openai/gpt-4o-mini-transcribe", api_key="os.environ/OPENAI_API_KEY" model="openai/gpt-4o-mini-transcribe", api_key="os.environ/OPENAI_API_KEY"
), ),
) )
resources.defer(lambda: endpoints_client.delete_model(model_id)) resources.defer(lambda: proxy.delete_model(model_id))
key = resources.key() client = sdk.openai(resources.key())
result = unwrap( transcription = client.audio.transcriptions.create(
endpoints_client.transcribe( model=model, file=(WEATHER_WAV.name, WEATHER_WAV.read_bytes(), "audio/wav")
key, model, filename=WEATHER_WAV.name, content=WEATHER_WAV.read_bytes()
)
) )
text = result.text.strip() text = transcription.text.strip()
assert text, "/audio/transcriptions returned an empty transcript" assert text, "/audio/transcriptions returned an empty transcript"
assert "weather" in text.lower(), ( assert "weather" in text.lower(), (
f"transcript of a spoken weather question does not mention weather: {text!r}" f"transcript of a spoken weather question does not mention weather: {text!r}"

5
tests/e2e/llm_translation/test_cache_control.py
View file

@ -15,7 +15,7 @@ service_tier lives in test_provider_features_e2e.py.
The provider-native cache_control request shape is not expressible with the The provider-native cache_control request shape is not expressible with the
shared ``ChatBody`` (whose content is a plain string), so the cacheable body is shared ``ChatBody`` (whose content is a plain string), so the cacheable body is
built from the typed content blocks shared in ``endpoints_client.py``. built from the typed content blocks shared in ``models.py``.
""" """
from __future__ import annotations from __future__ import annotations
@ -27,9 +27,8 @@ from pydantic import BaseModel
from e2e_config import unique_marker from e2e_config import unique_marker
from e2e_http import Result, unwrap from e2e_http import Result, unwrap
from endpoints_client import CacheControl, RichMessage, TextBlock
from lifecycle import ResourceManager from lifecycle import ResourceManager
from models import ChatResponse, LiteLLMParamsBody, Usage from models import CacheControl, ChatResponse, LiteLLMParamsBody, RichMessage, TextBlock, Usage
from passthrough_client import PassthroughClient from passthrough_client import PassthroughClient
import os import os

31
tests/e2e/llm_translation/test_credential_messages_e2e.py
View file

@ -7,43 +7,48 @@ import os
import pytest import pytest
from e2e_config import unique_marker from e2e_config import unique_marker
from e2e_http import require_successful_call
from endpoints_client import EndpointsClient, MessagesResult
from lifecycle import ResourceManager from lifecycle import ResourceManager
from models import CredentialCreateBody, LiteLLMParamsBody from models import CredentialCreateBody, LiteLLMParamsBody
from proxy_client import ProxyClient
from sdk_clients import SdkClients
pytestmark = pytest.mark.e2e pytestmark = pytest.mark.e2e
class TestCredentialBackedMessages: class TestCredentialBackedMessages:
@pytest.mark.covers("mgmt.credential.new.serves_request") @pytest.mark.covers("mgmt.credential.new.serves_request")
def test_credential_backed_messages(self, endpoints_client: EndpointsClient, resources: ResourceManager) -> None: def test_credential_backed_messages(
self, proxy: ProxyClient, resources: ResourceManager, sdk: SdkClients
) -> None:
marker = unique_marker() marker = unique_marker()
credential_name = f"e2e-cred-{marker}" credential_name = f"e2e-cred-{marker}"
model = f"e2e-cred-messages-{marker}" model = f"e2e-cred-messages-{marker}"
anthropic_api_key = os.getenv("ANTHROPIC_API_KEY") anthropic_api_key = os.getenv("ANTHROPIC_API_KEY")
assert anthropic_api_key, "ANTHROPIC_API_KEY must be set for this live e2e test" assert anthropic_api_key, "ANTHROPIC_API_KEY must be set for this live e2e test"
endpoints_client.proxy.create_credential( proxy.create_credential(
CredentialCreateBody( CredentialCreateBody(
credential_name=credential_name, credential_name=credential_name,
credential_values={"api_key": anthropic_api_key}, credential_values={"api_key": anthropic_api_key},
) )
) )
resources.defer(lambda: endpoints_client.proxy.delete_credential(credential_name)) resources.defer(lambda: proxy.delete_credential(credential_name))
model_id = endpoints_client.create_model( model_id = proxy.create_model(
model, model,
LiteLLMParamsBody( LiteLLMParamsBody(
model="anthropic/claude-haiku-4-5", model="anthropic/claude-haiku-4-5",
litellm_credential_name=credential_name, litellm_credential_name=credential_name,
), ),
) )
resources.defer(lambda: endpoints_client.delete_model(model_id)) resources.defer(lambda: proxy.delete_model(model_id))
key = resources.key() client = sdk.anthropic(resources.key())
result = endpoints_client.messages(key, model, "reply with one word") message = client.messages.create(
require_successful_call(result) model=model,
parsed = MessagesResult.model_validate_json(result.body) max_tokens=64,
assert parsed.role == "assistant", f"unexpected role: {result.body[:300]}" messages=[{"role": "user", "content": "reply with one word"}],
assert parsed.text.strip(), f"/v1/messages returned no text: {result.body[:300]}" )
assert message.role == "assistant", f"unexpected role: {message.role!r}"
text = "".join(block.text for block in message.content if block.type == "text")
assert text.strip(), f"/v1/messages returned no text: {message.content!r}"

33
tests/e2e/llm_translation/test_custom_pricing_e2e.py
View file

@ -25,7 +25,6 @@ from pydantic import BaseModel, RootModel
from e2e_config import unique_marker from e2e_config import unique_marker
from proxy_client import ProxyClient from proxy_client import ProxyClient
from e2e_http import Success, unwrap from e2e_http import Success, unwrap
from endpoints_client import EndpointsClient
from lifecycle import ResourceManager from lifecycle import ResourceManager
from models import ( from models import (
ChatBody, ChatBody,
@ -71,7 +70,7 @@ def _approx_equal(actual: float, expected: float) -> bool:
def _provision( def _provision(
endpoints_client: EndpointsClient, proxy: ProxyClient,
resources: ResourceManager, resources: ResourceManager,
prefix: str, prefix: str,
*, *,
@ -84,7 +83,7 @@ def _provision(
marker keeps the name unique so concurrent runs on the shared proxy never marker keeps the name unique so concurrent runs on the shared proxy never
collide.""" collide."""
model_name = f"{prefix}-{unique_marker()}" model_name = f"{prefix}-{unique_marker()}"
model_id = endpoints_client.create_model( model_id = proxy.create_model(
model_name, model_name,
LiteLLMParamsBody( LiteLLMParamsBody(
model=BACKEND_MODEL, model=BACKEND_MODEL,
@ -93,15 +92,15 @@ def _provision(
output_cost_per_token=output_cost_per_token, output_cost_per_token=output_cost_per_token,
), ),
) )
resources.defer(lambda: endpoints_client.delete_model(model_id)) resources.defer(lambda: proxy.delete_model(model_id))
return model_name return model_name
def _provision_custom_priced( def _provision_custom_priced(
endpoints_client: EndpointsClient, resources: ResourceManager proxy: ProxyClient, resources: ResourceManager
) -> str: ) -> str:
return _provision( return _provision(
endpoints_client, proxy,
resources, resources,
"custom-priced-flash", "custom-priced-flash",
input_cost_per_token=CUSTOM_INPUT_RATE, input_cost_per_token=CUSTOM_INPUT_RATE,
@ -151,14 +150,14 @@ def _poll_breakdown_row(proxy: ProxyClient, key: str, response_id: str | None) -
class TestCustomPricing: class TestCustomPricing:
def test_custom_pricing_is_billed_at_configured_rate( def test_custom_pricing_is_billed_at_configured_rate(
self, self,
endpoints_client: EndpointsClient, proxy: ProxyClient,
resources: ResourceManager, resources: ResourceManager,
scoped_key: str, scoped_key: str,
) -> None: ) -> None:
model = _provision_custom_priced(endpoints_client, resources) model = _provision_custom_priced(proxy, resources)
chat = unwrap( chat = unwrap(
endpoints_client.proxy.chat( proxy.chat(
scoped_key, scoped_key,
ChatBody( ChatBody(
model=model, model=model,
@ -172,7 +171,7 @@ class TestCustomPricing:
) )
) )
row = _poll_breakdown_row(endpoints_client.proxy, scoped_key, chat.id) row = _poll_breakdown_row(proxy, scoped_key, chat.id)
assert row.metadata and row.metadata.cost_breakdown # guaranteed by the poll assert row.metadata and row.metadata.cost_breakdown # guaranteed by the poll
breakdown = row.metadata.cost_breakdown breakdown = row.metadata.cost_breakdown
@ -195,10 +194,10 @@ class TestCustomPricing:
) )
def test_model_info_reports_custom_pricing( def test_model_info_reports_custom_pricing(
self, endpoints_client: EndpointsClient, resources: ResourceManager self, proxy: ProxyClient, resources: ResourceManager
) -> None: ) -> None:
model = _provision_custom_priced(endpoints_client, resources) model = _provision_custom_priced(proxy, resources)
entry = _model_info_entry(endpoints_client.proxy.model_info(), model) entry = _model_info_entry(proxy.model_info(), model)
assert entry.litellm_params.input_cost_per_token == CUSTOM_INPUT_RATE, ( assert entry.litellm_params.input_cost_per_token == CUSTOM_INPUT_RATE, (
f"/model/info litellm_params input rate " f"/model/info litellm_params input rate "
@ -210,20 +209,20 @@ class TestCustomPricing:
) )
def test_custom_pricing_is_isolated_from_sibling_deployment( def test_custom_pricing_is_isolated_from_sibling_deployment(
self, endpoints_client: EndpointsClient, resources: ResourceManager self, proxy: ProxyClient, resources: ResourceManager
) -> None: ) -> None:
# Register the override first so its rate is in the backend cost map before # Register the override first so its rate is in the backend cost map before
# the sibling resolves; a leak (LIT-3897) would then poison the sibling. # the sibling resolves; a leak (LIT-3897) would then poison the sibling.
custom = _provision_custom_priced(endpoints_client, resources) custom = _provision_custom_priced(proxy, resources)
sibling = _provision( sibling = _provision(
endpoints_client, proxy,
resources, resources,
"base-flash", "base-flash",
input_cost_per_token=None, input_cost_per_token=None,
output_cost_per_token=None, output_cost_per_token=None,
) )
entries = {entry.model_name: entry for entry in endpoints_client.proxy.model_info()} entries = {entry.model_name: entry for entry in proxy.model_info()}
custom_entry = entries.get(custom) custom_entry = entries.get(custom)
sibling_entry = entries.get(sibling) sibling_entry = entries.get(sibling)
assert custom_entry is not None, f"{custom} absent from /model/info" assert custom_entry is not None, f"{custom} absent from /model/info"

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

@ -1,9 +1,10 @@
"""Live e2e: POST /embeddings returns a real vector across OpenAI, Bedrock, Vertex. """Live e2e: POST /embeddings returns a real vector across OpenAI, Bedrock, Vertex.
Each test registers the deployment it needs at runtime (deleted on teardown) and Each test registers the deployment it needs at runtime (deleted on teardown),
asserts a non-empty, non-zero vector came back. The LIT-3167 guard in drives the endpoint with the real OpenAI SDK (LIT-4577), and asserts a
tests/e2e/embeddings/ covers the Gemini embedding path; embeddings cost tracking is non-empty, non-zero vector came back. The LIT-3167 guard in
covered by tests/e2e/quota_management/spend_tracking/. tests/e2e/embeddings/ covers the Gemini embedding path; embeddings cost tracking
is covered by tests/e2e/quota_management/spend_tracking/.
""" """
from __future__ import annotations from __future__ import annotations
@ -11,79 +12,74 @@ from __future__ import annotations
import pytest import pytest
from e2e_config import unique_marker from e2e_config import unique_marker
from e2e_http import require_successful_call
from endpoints_client import EmbeddingsResult, EndpointsClient
from lifecycle import ResourceManager from lifecycle import ResourceManager
from models import LiteLLMParamsBody from models import LiteLLMParamsBody
from proxy_client import ProxyClient
from sdk_clients import SdkClients
pytestmark = pytest.mark.e2e pytestmark = pytest.mark.e2e
def _assert_embedding_vector(
proxy: ProxyClient,
resources: ResourceManager,
sdk: SdkClients,
prefix: str,
params: LiteLLMParamsBody,
) -> None:
model = f"{prefix}-{unique_marker()}"
model_id = proxy.create_model(model, params)
resources.defer(lambda: proxy.delete_model(model_id))
client = sdk.openai(resources.key())
embeddings = client.embeddings.create(model=model, input="Say this is a test!")
assert embeddings.data, f"/embeddings returned no data: {embeddings!r}"
vector = embeddings.data[0].embedding
assert vector, f"/embeddings returned no vector: {embeddings!r}"
assert any(component != 0.0 for component in vector), "embedding vector is all zeros"
class TestEmbeddingsEndpoint: class TestEmbeddingsEndpoint:
@pytest.mark.covers("llm.embeddings.openai.basic.nonstream.works") @pytest.mark.covers("llm.embeddings.openai.basic.nonstream.works")
def test_embeddings_returns_vector( def test_embeddings_returns_vector(
self, endpoints_client: EndpointsClient, resources: ResourceManager self, proxy: ProxyClient, resources: ResourceManager, sdk: SdkClients
) -> None: ) -> None:
model = f"e2e-embeddings-{unique_marker()}" _assert_embedding_vector(
model_id = endpoints_client.create_model( proxy,
model, resources,
sdk,
"e2e-embeddings",
LiteLLMParamsBody( LiteLLMParamsBody(
model="openai/text-embedding-3-small", api_key="os.environ/OPENAI_API_KEY" model="openai/text-embedding-3-small", api_key="os.environ/OPENAI_API_KEY"
), ),
) )
resources.defer(lambda: endpoints_client.delete_model(model_id))
key = resources.key()
result = endpoints_client.embeddings(key, model, "Say this is a test!")
require_successful_call(result)
parsed = EmbeddingsResult.model_validate_json(result.body)
assert parsed.first_vector, f"/embeddings returned no vector: {result.body[:300]}"
assert any(component != 0.0 for component in parsed.first_vector), (
f"embedding vector is all zeros: {result.body[:300]}"
)
@pytest.mark.covers("llm.embeddings.bedrock.basic.nonstream.works") @pytest.mark.covers("llm.embeddings.bedrock.basic.nonstream.works")
def test_bedrock_embeddings_returns_vector( def test_bedrock_embeddings_returns_vector(
self, endpoints_client: EndpointsClient, resources: ResourceManager self, proxy: ProxyClient, resources: ResourceManager, sdk: SdkClients
) -> None: ) -> None:
model = f"e2e-embeddings-bedrock-{unique_marker()}" _assert_embedding_vector(
model_id = endpoints_client.create_model( proxy,
model, resources,
sdk,
"e2e-embeddings-bedrock",
LiteLLMParamsBody( LiteLLMParamsBody(
model="bedrock/amazon.titan-embed-text-v2:0", aws_region_name="us-west-2" model="bedrock/amazon.titan-embed-text-v2:0", aws_region_name="us-west-2"
), ),
) )
resources.defer(lambda: endpoints_client.delete_model(model_id))
key = resources.key()
result = endpoints_client.embeddings(key, model, "Say this is a test!")
require_successful_call(result)
parsed = EmbeddingsResult.model_validate_json(result.body)
assert parsed.first_vector, f"/embeddings returned no vector: {result.body[:300]}"
assert any(component != 0.0 for component in parsed.first_vector), (
f"embedding vector is all zeros: {result.body[:300]}"
)
@pytest.mark.covers("llm.embeddings.vertex.basic.nonstream.works") @pytest.mark.covers("llm.embeddings.vertex.basic.nonstream.works")
def test_vertex_embeddings_returns_vector( def test_vertex_embeddings_returns_vector(
self, endpoints_client: EndpointsClient, resources: ResourceManager self, proxy: ProxyClient, resources: ResourceManager, sdk: SdkClients
) -> None: ) -> None:
model = f"e2e-embeddings-vertex-{unique_marker()}" _assert_embedding_vector(
model_id = endpoints_client.create_model( proxy,
model, resources,
sdk,
"e2e-embeddings-vertex",
LiteLLMParamsBody( LiteLLMParamsBody(
model="vertex_ai/gemini-embedding-2", model="vertex_ai/gemini-embedding-2",
vertex_project="os.environ/VERTEXAI_PROJECT", vertex_project="os.environ/VERTEXAI_PROJECT",
vertex_location="us-central1", vertex_location="us-central1",
), ),
) )
resources.defer(lambda: endpoints_client.delete_model(model_id))
key = resources.key()
result = endpoints_client.embeddings(key, model, "Say this is a test!")
require_successful_call(result)
parsed = EmbeddingsResult.model_validate_json(result.body)
assert parsed.first_vector, f"/embeddings returned no vector: {result.body[:300]}"
assert any(component != 0.0 for component in parsed.first_vector), (
f"embedding vector is all zeros: {result.body[:300]}"
)

73
tests/e2e/llm_translation/test_image_generation_e2e.py
View file

@ -1,8 +1,7 @@
"""Live e2e: POST /v1/images/generations returns an image. """Live e2e: POST /v1/images/generations returns an image.
Registers an OpenAI image deployment at runtime and asserts the response carries a Registers an image deployment at runtime, drives it through the real OpenAI SDK
generated image (url or base64). Migrated from (LIT-4577), and asserts the response carries a generated image (url or base64).
litellm-regression-tests/tests/test_inference_endpoints.py.
""" """
from __future__ import annotations from __future__ import annotations
@ -10,50 +9,58 @@ from __future__ import annotations
import pytest import pytest
from e2e_config import require_env, unique_marker from e2e_config import require_env, unique_marker
from e2e_http import require_successful_call
from endpoints_client import EndpointsClient, ImagesResult
from lifecycle import ResourceManager from lifecycle import ResourceManager
from models import LiteLLMParamsBody from models import LiteLLMParamsBody
from proxy_client import ProxyClient
from sdk_clients import SdkClients
pytestmark = pytest.mark.e2e pytestmark = pytest.mark.e2e
def _assert_image_returned(body: str) -> None: def _assert_image_returned(
parsed = ImagesResult.model_validate_json(body) proxy: ProxyClient,
assert parsed.data, f"/images/generations returned no data: {body[:300]}" resources: ResourceManager,
first = parsed.data[0] sdk: SdkClients,
assert first.b64_json or first.url, ( prefix: str,
f"generated image has neither b64_json nor url: {body[:300]}" params: LiteLLMParamsBody,
) ) -> None:
model = f"{prefix}-{unique_marker()}"
model_id = proxy.create_model(model, params)
resources.defer(lambda: proxy.delete_model(model_id))
client = sdk.openai(resources.key())
images = client.images.generate(model=model, prompt="Draw a cute cat", n=1, size="1024x1024")
data = images.data or []
assert data, f"/images/generations returned no data: {images!r}"
first = data[0]
assert first.b64_json or first.url, "generated image has neither b64_json nor url"
class TestImageGeneration: class TestImageGeneration:
@pytest.mark.covers("llm.images_generations.openai.basic.nonstream.works") @pytest.mark.covers("llm.images_generations.openai.basic.nonstream.works")
def test_image_generation_returns_image( def test_image_generation_returns_image(
self, endpoints_client: EndpointsClient, resources: ResourceManager self, proxy: ProxyClient, resources: ResourceManager, sdk: SdkClients
) -> None: ) -> None:
model = f"e2e-image-{unique_marker()}" _assert_image_returned(
model_id = endpoints_client.create_model( proxy,
model, resources,
LiteLLMParamsBody( sdk,
model="openai/gpt-image-1-mini", api_key="os.environ/OPENAI_API_KEY" "e2e-image",
), LiteLLMParamsBody(model="openai/gpt-image-1-mini", api_key="os.environ/OPENAI_API_KEY"),
) )
resources.defer(lambda: endpoints_client.delete_model(model_id))
key = resources.key()
result = endpoints_client.images(key, model, "Draw a cute cat") @pytest.mark.covers(
require_successful_call(result) "llm.images_generations.bedrock.basic.nonstream.works", exercised_on=["images_generations"]
_assert_image_returned(result.body) )
@pytest.mark.covers("llm.images_generations.bedrock.basic.nonstream.works", exercised_on=["images_generations"])
def test_bedrock_image_generation_returns_image( def test_bedrock_image_generation_returns_image(
self, endpoints_client: EndpointsClient, resources: ResourceManager self, proxy: ProxyClient, resources: ResourceManager, sdk: SdkClients
) -> None: ) -> None:
require_env("AWS_ACCESS_KEY_ID", "AWS_SECRET_ACCESS_KEY", "AWS_REGION") require_env("AWS_ACCESS_KEY_ID", "AWS_SECRET_ACCESS_KEY", "AWS_REGION")
model = f"e2e-bedrock-image-{unique_marker()}" _assert_image_returned(
model_id = endpoints_client.create_model( proxy,
model, resources,
sdk,
"e2e-bedrock-image",
LiteLLMParamsBody( LiteLLMParamsBody(
model="bedrock/amazon.titan-image-generator-v2:0", model="bedrock/amazon.titan-image-generator-v2:0",
aws_access_key_id="os.environ/AWS_ACCESS_KEY_ID", aws_access_key_id="os.environ/AWS_ACCESS_KEY_ID",
@ -61,9 +68,3 @@ class TestImageGeneration:
aws_region_name="os.environ/AWS_REGION", aws_region_name="os.environ/AWS_REGION",
), ),
) )
resources.defer(lambda: endpoints_client.delete_model(model_id))
key = resources.key()
result = endpoints_client.images(key, model, "Draw a cute cat")
require_successful_call(result)
_assert_image_returned(result.body)

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

@ -1,9 +1,9 @@
"""Live e2e: POST /v1/messages routed to Azure AI Foundry Anthropic deployments. """Live e2e: POST /v1/messages routed to Azure AI Foundry Anthropic deployments.
Registers `azure_ai/<claude>` deployments at runtime and drives the Messages Registers `azure_ai/<claude>` deployments at runtime and drives the Messages
endpoint through the gateway across the behaviors an Anthropic client relies on: endpoint through the gateway with the real Anthropic SDK (LIT-4577) across the
a basic completion, a streamed completion, and tool use (non-streaming and behaviors an Anthropic client relies on: a basic completion, a streamed
streaming). Auth is the Azure API key (`x-api-key`); the deployment reads completion, and tool use (non-streaming and streaming). The deployment reads
`AZURE_AI_API_BASE` / `AZURE_AI_API_KEY` from the proxy env, so no secret is `AZURE_AI_API_BASE` / `AZURE_AI_API_KEY` from the proxy env, so no secret is
sent in the request. sent in the request.
""" """
@ -11,60 +11,50 @@ sent in the request.
from __future__ import annotations from __future__ import annotations
import pytest import pytest
from anthropic.types import RawMessageStreamEvent, ToolParam
from e2e_config import EXPECT_RUST, unique_marker from e2e_config import EXPECT_RUST, unique_marker
from e2e_http import StreamingResponse, require_successful_call, unwrap
from endpoints_client import EndpointsClient
from lifecycle import ResourceManager from lifecycle import ResourceManager
from models import ( from models import LiteLLMParamsBody
AnthropicCustomTool, from proxy_client import ProxyClient
AnthropicMessagesBody, from sdk_clients import SdkClients
ChatMessage,
JsonSchemaProperty,
LiteLLMParamsBody,
ToolInputSchema,
)
pytestmark = pytest.mark.e2e pytestmark = pytest.mark.e2e
AZURE_FOUNDRY_MODEL = "azure_ai/claude-haiku-4-5" AZURE_FOUNDRY_MODEL = "azure_ai/claude-haiku-4-5"
WEATHER_TOOL = AnthropicCustomTool( WEATHER_TOOL: ToolParam = {
name="get_weather", "name": "get_weather",
description="Get the current weather for a city.", "description": "Get the current weather for a city.",
input_schema=ToolInputSchema( "input_schema": {
properties={"city": JsonSchemaProperty(type="string")}, "type": "object",
required=["city"], "properties": {"city": {"type": "string"}},
), "required": ["city"],
) },
}
def _assert_streamed_ok(result: StreamingResponse) -> None: def _assert_rust_served(headers: dict[str, str]) -> None:
require_successful_call(result) if not EXPECT_RUST:
assert result.is_streaming, f"response was not streamed: {result.headers}" return
assert not result.stream_error, f"stream errored: {result.stream_error}" assert headers.get("x-litellm-rust") == "true", (
assert result.stream_events, "stream produced no SSE events" "E2E_EXPECT_RUST is set, so this gateway must serve /v1/messages through the "
assert any("content_block_delta" in event for event in result.stream_events), ( "Rust path, but the response carried no x-litellm-rust marker. The request "
"stream carried no content deltas" "still succeeded, which is exactly the failure mode: a gateway whose native "
f"extension is unavailable falls back to Python silently. headers={headers}"
) )
assert any("message_stop" in event for event in result.stream_events), (
"stream never reached message_stop"
) def _assert_streamed_ok(event_types: list[str]) -> None:
if EXPECT_RUST: assert event_types, "stream produced no SSE events"
assert result.headers.get("x-litellm-rust") == "true", ( assert "content_block_delta" in event_types, "stream carried no content deltas"
"E2E_EXPECT_RUST is set, so this gateway must serve /v1/messages through the " assert "message_stop" in event_types, "stream never reached message_stop"
"Rust path, but the response carried no x-litellm-rust marker. The request "
"still succeeded, which is exactly the failure mode: a gateway whose native "
f"extension is unavailable falls back to Python silently. headers={result.headers}"
)
class TestAzureFoundryMessages: class TestAzureFoundryMessages:
def _register( def _register(self, proxy: ProxyClient, resources: ResourceManager) -> str:
self, endpoints_client: EndpointsClient, resources: ResourceManager
) -> tuple[str, str]:
model = f"e2e-azure-foundry-messages-{unique_marker()}" model = f"e2e-azure-foundry-messages-{unique_marker()}"
model_id = endpoints_client.create_model( model_id = proxy.create_model(
model, model,
LiteLLMParamsBody( LiteLLMParamsBody(
model=AZURE_FOUNDRY_MODEL, model=AZURE_FOUNDRY_MODEL,
@ -72,91 +62,78 @@ class TestAzureFoundryMessages:
api_key="os.environ/AZURE_AI_API_KEY", api_key="os.environ/AZURE_AI_API_KEY",
), ),
) )
resources.defer(lambda: endpoints_client.delete_model(model_id)) resources.defer(lambda: proxy.delete_model(model_id))
return model, resources.key(models=[model]) return model
@pytest.mark.covers("llm.messages.azure_foundry.basic.nonstream.works") @pytest.mark.covers("llm.messages.azure_foundry.basic.nonstream.works")
def test_basic_nonstream( def test_basic_nonstream(
self, endpoints_client: EndpointsClient, resources: ResourceManager self, proxy: ProxyClient, resources: ResourceManager, sdk: SdkClients
) -> None: ) -> None:
model, key = self._register(endpoints_client, resources) model = self._register(proxy, resources)
response = unwrap( client = sdk.anthropic(resources.key(models=[model]))
endpoints_client.proxy.messages(
key, message = client.messages.create(
AnthropicMessagesBody( model=model,
model=model, max_tokens=64,
max_tokens=64, messages=[{"role": "user", "content": "Reply with one word."}],
messages=[ChatMessage(role="user", content="Reply with one word.")],
),
)
) )
assert response.content, f"no content blocks in response: {response}" assert message.content, f"no content blocks in response: {message!r}"
text = "".join(block.text or "" for block in response.content if block.type == "text") text = "".join(block.text for block in message.content if block.type == "text")
assert text.strip(), f"/v1/messages returned no text: {response}" assert text.strip(), f"/v1/messages returned no text: {message.content!r}"
@pytest.mark.covers("llm.messages.azure_foundry.basic.stream.works") @pytest.mark.covers("llm.messages.azure_foundry.basic.stream.works")
def test_basic_stream( def test_basic_stream(
self, endpoints_client: EndpointsClient, resources: ResourceManager self, proxy: ProxyClient, resources: ResourceManager, sdk: SdkClients
) -> None: ) -> None:
model, key = self._register(endpoints_client, resources) model = self._register(proxy, resources)
result = endpoints_client.proxy.messages_stream( client = sdk.anthropic(resources.key(models=[model]))
key,
AnthropicMessagesBody( raw = client.messages.with_raw_response.create(
model=model, model=model,
max_tokens=64, max_tokens=64,
stream=True, stream=True,
messages=[ChatMessage(role="user", content="Count from one to three.")], messages=[{"role": "user", "content": "Count from one to three."}],
),
) )
_assert_streamed_ok(result) _assert_rust_served({name.lower(): value for name, value in raw.headers.items()})
_assert_streamed_ok([event.type for event in raw.parse()])
@pytest.mark.covers("llm.messages.azure_foundry.tool_use.nonstream.works") @pytest.mark.covers("llm.messages.azure_foundry.tool_use.nonstream.works")
def test_tool_use_nonstream( def test_tool_use_nonstream(
self, endpoints_client: EndpointsClient, resources: ResourceManager self, proxy: ProxyClient, resources: ResourceManager, sdk: SdkClients
) -> None: ) -> None:
model, key = self._register(endpoints_client, resources) model = self._register(proxy, resources)
response = unwrap( client = sdk.anthropic(resources.key(models=[model]))
endpoints_client.proxy.messages(
key, message = client.messages.create(
AnthropicMessagesBody( model=model,
model=model, max_tokens=256,
max_tokens=256, tools=[WEATHER_TOOL],
tools=[WEATHER_TOOL], messages=[{"role": "user", "content": "What is the weather in Paris? Use the tool."}],
messages=[
ChatMessage(role="user", content="What is the weather in Paris? Use the tool.")
],
),
)
) )
assert response.content, f"no content blocks in response: {response}" assert message.content, f"no content blocks in response: {message!r}"
assert any(block.type == "tool_use" for block in response.content), ( assert any(block.type == "tool_use" for block in message.content), (
f"model did not call the tool: {response}" f"model did not call the tool: {message.content!r}"
) )
@pytest.mark.covers("llm.messages.azure_foundry.tool_use.stream.works") @pytest.mark.covers("llm.messages.azure_foundry.tool_use.stream.works")
def test_tool_use_stream( def test_tool_use_stream(
self, endpoints_client: EndpointsClient, resources: ResourceManager self, proxy: ProxyClient, resources: ResourceManager, sdk: SdkClients
) -> None: ) -> None:
model, key = self._register(endpoints_client, resources) model = self._register(proxy, resources)
result = endpoints_client.proxy.messages_stream( client = sdk.anthropic(resources.key(models=[model]))
key,
AnthropicMessagesBody( stream = client.messages.create(
model=model, model=model,
max_tokens=256, max_tokens=256,
stream=True, stream=True,
tools=[WEATHER_TOOL], tools=[WEATHER_TOOL],
messages=[ messages=[{"role": "user", "content": "What is the weather in Paris? Use the tool."}],
ChatMessage(role="user", content="What is the weather in Paris? Use the tool.")
],
),
)
require_successful_call(result)
assert result.is_streaming, f"response was not streamed: {result.headers}"
assert not result.stream_error, f"stream errored: {result.stream_error}"
assert result.stream_events, "stream produced no SSE events"
assert any("tool_use" in event for event in result.stream_events), (
"stream carried no tool_use block"
)
assert any("message_stop" in event for event in result.stream_events), (
"stream never reached message_stop"
) )
events: list[RawMessageStreamEvent] = list(stream)
event_types = [event.type for event in events]
assert event_types, "stream produced no SSE events"
assert any(
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"

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

@ -1,41 +1,35 @@
"""Live e2e: POST /v1/messages (Anthropic Messages API) returns a real completion. """Live e2e: POST /v1/messages (Anthropic Messages API) returns a real completion.
Registers an Anthropic deployment at runtime, drives the Messages endpoint through Registers an Anthropic deployment at runtime and drives the Messages endpoint
the gateway, and asserts an assistant message with text came back, both through the gateway with the real Anthropic SDK, the client customers actually
non-streaming and streamed. Migrated from use (LIT-4577), asserting an assistant message with text came back, both
litellm-regression-tests/tests/test_inference_endpoints.py. non-streaming and streamed.
""" """
from __future__ import annotations from __future__ import annotations
import pytest import pytest
from anthropic.types import Message, ToolParam
from e2e_config import require_env, unique_marker from e2e_config import require_env, unique_marker
from e2e_http import require_successful_call, unwrap
from endpoints_client import EndpointsClient, MessagesResult
from lifecycle import ResourceManager from lifecycle import ResourceManager
from models import ( from models import LiteLLMParamsBody, SpendLogRow
AnthropicCustomTool, from proxy_client import ProxyClient
AnthropicMessagesBody, from sdk_clients import SdkClients, response_header
ChatMessage,
JsonSchemaProperty,
LiteLLMParamsBody,
SpendLogRow,
ToolInputSchema,
)
pytestmark = pytest.mark.e2e pytestmark = pytest.mark.e2e
ANTHROPIC_BACKEND = "anthropic/claude-haiku-4-5" ANTHROPIC_BACKEND = "anthropic/claude-haiku-4-5"
WEATHER_TOOL = AnthropicCustomTool( WEATHER_TOOL: ToolParam = {
name="get_weather", "name": "get_weather",
description="Get the current weather for a city.", "description": "Get the current weather for a city.",
input_schema=ToolInputSchema( "input_schema": {
properties={"city": JsonSchemaProperty(type="string")}, "type": "object",
required=["city"], "properties": {"city": {"type": "string"}},
), "required": ["city"],
) },
}
def _approx_equal(actual: float, expected: float) -> bool: def _approx_equal(actual: float, expected: float) -> bool:
@ -43,60 +37,66 @@ def _approx_equal(actual: float, expected: float) -> bool:
return abs(actual - expected) <= max(1e-9, abs(expected) * 1e-2) return abs(actual - expected) <= max(1e-9, abs(expected) * 1e-2)
def _text(message: Message) -> str:
return "".join(block.text for block in message.content if block.type == "text")
class TestAnthropicMessages: class TestAnthropicMessages:
def _register( def _register(
self, endpoints_client: EndpointsClient, resources: ResourceManager self, proxy: ProxyClient, resources: ResourceManager, prefix: str = "e2e-messages"
) -> tuple[str, str]: ) -> str:
model = f"e2e-messages-{unique_marker()}" model = f"{prefix}-{unique_marker()}"
model_id = endpoints_client.create_model( model_id = proxy.create_model(
model, model,
LiteLLMParamsBody( LiteLLMParamsBody(model=ANTHROPIC_BACKEND, api_key="os.environ/ANTHROPIC_API_KEY"),
model=ANTHROPIC_BACKEND, api_key="os.environ/ANTHROPIC_API_KEY"
),
) )
resources.defer(lambda: endpoints_client.delete_model(model_id)) resources.defer(lambda: proxy.delete_model(model_id))
return model, resources.key() return model
@pytest.mark.covers("llm.messages.anthropic.basic.nonstream.works") @pytest.mark.covers("llm.messages.anthropic.basic.nonstream.works")
def test_messages_returns_completion( def test_messages_returns_completion(
self, endpoints_client: EndpointsClient, resources: ResourceManager self, proxy: ProxyClient, resources: ResourceManager, sdk: SdkClients
) -> None: ) -> None:
model, key = self._register(endpoints_client, resources) model = self._register(proxy, resources)
client = sdk.anthropic(resources.key())
result = endpoints_client.messages(key, model, "reply with one word") message = client.messages.create(
require_successful_call(result) model=model,
parsed = MessagesResult.model_validate_json(result.body) max_tokens=64,
assert parsed.role == "assistant", f"unexpected role: {result.body[:300]}" messages=[{"role": "user", "content": "reply with one word"}],
assert parsed.text.strip(), f"/v1/messages returned no text: {result.body[:300]}" )
assert message.role == "assistant", f"unexpected role: {message.role!r}"
assert _text(message).strip(), f"/v1/messages returned no text: {message.content!r}"
@pytest.mark.covers("llm.messages.anthropic.basic.nonstream.cost_logged") @pytest.mark.covers("llm.messages.anthropic.basic.nonstream.cost_logged")
def test_messages_logs_cost_matching_the_response_header( def test_messages_logs_cost_matching_the_response_header(
self, endpoints_client: EndpointsClient, resources: ResourceManager self, proxy: ProxyClient, resources: ResourceManager, sdk: SdkClients
) -> None: ) -> None:
require_env("ANTHROPIC_API_KEY") require_env("ANTHROPIC_API_KEY")
model = f"e2e-messages-cost-{unique_marker()}" model = self._register(proxy, resources, prefix="e2e-messages-cost")
model_id = endpoints_client.create_model(
model,
LiteLLMParamsBody(
model=ANTHROPIC_BACKEND, api_key="os.environ/ANTHROPIC_API_KEY"
),
)
resources.defer(lambda: endpoints_client.delete_model(model_id))
key = resources.key() key = resources.key()
client = sdk.anthropic(key)
result = endpoints_client.messages(key, model, f"reply with one word {unique_marker()}") raw = client.messages.with_raw_response.create(
require_successful_call(result) model=model,
parsed = MessagesResult.model_validate_json(result.body) max_tokens=64,
assert parsed.role == "assistant" and parsed.text.strip(), ( messages=[{"role": "user", "content": f"reply with one word {unique_marker()}"}],
f"/v1/messages returned no assistant text: {result.body[:300]}" )
message = raw.parse()
assert message.role == "assistant" and _text(message).strip(), (
f"/v1/messages returned no assistant text: {message.content!r}"
) )
# The customer reads per-request cost off the response header (LIT-4076), so # The customer reads per-request cost off the response header (LIT-4076), so
# it must be present and positive on /v1/messages, not only /chat/completions. # it must be present and positive on /v1/messages, not only /chat/completions.
header_cost = result.response_cost raw_header_cost = response_header(raw.headers, "x-litellm-response-cost")
assert header_cost is not None and header_cost > 0, ( assert raw_header_cost is not None, (
"x-litellm-response-cost header missing or non-positive on /v1/messages; " "x-litellm-response-cost header missing on /v1/messages; "
f"headers={result.headers}" f"headers={dict(raw.headers)}"
)
header_cost = float(raw_header_cost)
assert header_cost > 0, (
f"x-litellm-response-cost header non-positive on /v1/messages: {header_cost}"
) )
# Correlate the spend row by the unique scoped key, not the Anthropic response # Correlate the spend row by the unique scoped key, not the Anthropic response
@ -107,7 +107,7 @@ class TestAnthropicMessages:
def _priced(rows: list[SpendLogRow]) -> bool: def _priced(rows: list[SpendLogRow]) -> bool:
return any(r.spend is not None and r.spend > 0 for r in rows) return any(r.spend is not None and r.spend > 0 for r in rows)
rows = endpoints_client.proxy.poll_logs_for_key(key, predicate=_priced) rows = proxy.poll_logs_for_key(key, predicate=_priced)
priced = [r for r in rows if r.spend is not None and r.spend > 0] priced = [r for r in rows if r.spend is not None and r.spend > 0]
assert priced, ( assert priced, (
f"no priced /spend/logs row landed for key {key} within the poll window; got {rows}" f"no priced /spend/logs row landed for key {key} within the poll window; got {rows}"
@ -123,50 +123,36 @@ class TestAnthropicMessages:
@pytest.mark.covers("llm.messages.anthropic.basic.stream.works") @pytest.mark.covers("llm.messages.anthropic.basic.stream.works")
def test_messages_streams_completion( def test_messages_streams_completion(
self, endpoints_client: EndpointsClient, resources: ResourceManager self, proxy: ProxyClient, resources: ResourceManager, sdk: SdkClients
) -> None: ) -> None:
model, key = self._register(endpoints_client, resources) model = self._register(proxy, resources)
client = sdk.anthropic(resources.key())
result = endpoints_client.proxy.messages_stream( stream = client.messages.create(
key, model=model,
AnthropicMessagesBody( max_tokens=64,
model=model, stream=True,
max_tokens=64, messages=[{"role": "user", "content": "Count from one to three."}],
stream=True,
messages=[ChatMessage(role="user", content="Count from one to three.")],
),
)
require_successful_call(result)
assert result.is_streaming, f"response was not streamed: {result.headers}"
assert not result.stream_error, f"stream errored: {result.stream_error}"
assert result.stream_events, "stream produced no SSE events"
assert any("content_block_delta" in event for event in result.stream_events), (
"stream carried no content deltas"
)
assert any("message_stop" in event for event in result.stream_events), (
"stream never reached message_stop"
) )
event_types = [event.type for event in stream]
assert event_types, "stream produced no SSE events"
assert "content_block_delta" in event_types, "stream carried no content deltas"
assert "message_stop" in event_types, "stream never reached message_stop"
@pytest.mark.covers("llm.messages.anthropic.tool_use.nonstream.works") @pytest.mark.covers("llm.messages.anthropic.tool_use.nonstream.works")
def test_messages_tool_use( def test_messages_tool_use(
self, endpoints_client: EndpointsClient, resources: ResourceManager self, proxy: ProxyClient, resources: ResourceManager, sdk: SdkClients
) -> None: ) -> None:
model, key = self._register(endpoints_client, resources) model = self._register(proxy, resources)
client = sdk.anthropic(resources.key())
response = unwrap( message = client.messages.create(
endpoints_client.proxy.messages( model=model,
key, max_tokens=256,
AnthropicMessagesBody( tools=[WEATHER_TOOL],
model=model, messages=[{"role": "user", "content": "What is the weather in Paris? Use the tool."}],
max_tokens=256,
tools=[WEATHER_TOOL],
messages=[
ChatMessage(role="user", content="What is the weather in Paris? Use the tool.")
],
),
)
) )
assert response.content, f"no content blocks in response: {response}" assert message.content, f"no content blocks in response: {message!r}"
assert any(block.type == "tool_use" for block in response.content), ( assert any(block.type == "tool_use" for block in message.content), (
f"model did not call the tool: {response}" f"model did not call the tool: {message.content!r}"
) )

124
tests/e2e/llm_translation/test_messages_mid_conversation_system_e2e.py
View file

@ -17,27 +17,28 @@ entry whose prefix spans ``system`` plus message turns is invalidated when the
reminder is hoisted (the ``system`` field mutates and a turn disappears from reminder is hoisted (the ``system`` field mutates and a turn disappears from
``messages``), while an entry ending at the system block itself would survive ``messages``), while an entry ending at the system block itself would survive
the hoist and mask the regression. the hoist and mask the regression.
Calls go through the real Anthropic SDK (LIT-4577). The SDK's ``MessageParam``
type only admits user/assistant roles, so the system reminder turn is cast to
it; the SDK serializes the dict verbatim, which is exactly the wire shape under
test.
""" """
from __future__ import annotations from __future__ import annotations
import time import time
from typing import cast
import pytest import pytest
from anthropic import Anthropic
from anthropic.types import Message, MessageParam, TextBlockParam
from pydantic import BaseModel from pydantic import BaseModel
from e2e_config import unique_marker from e2e_config import unique_marker
from e2e_http import Result, unwrap
from endpoints_client import (
CacheControl,
EndpointsClient,
MessagesResult,
RichMessage,
RichMessagesRequest,
TextBlock,
)
from lifecycle import ResourceManager from lifecycle import ResourceManager
from models import LiteLLMParamsBody from models import LiteLLMParamsBody
from proxy_client import ProxyClient
from sdk_clients import SdkClients
pytestmark = pytest.mark.e2e pytestmark = pytest.mark.e2e
@ -48,50 +49,51 @@ CACHE_PRIMING_DEADLINE_SECONDS = 60.0
CACHE_PRIMING_INTERVAL_SECONDS = 3.0 CACHE_PRIMING_INTERVAL_SECONDS = 3.0
def _cacheable_system_block(marker: str) -> TextBlock: def _cacheable_system_block(marker: str) -> TextBlockParam:
"""A system prompt comfortably above Sonnet's 1024-token minimum cacheable """A system prompt comfortably above Sonnet's 1024-token minimum cacheable
size, unique per run so no other run's cache entry can satisfy the read.""" size, unique per run so no other run's cache entry can satisfy the read."""
text = " ".join( text = " ".join(
f"Reference paragraph {index} for run {marker}." for index in range(300) f"Reference paragraph {index} for run {marker}." for index in range(300)
) )
return TextBlock(text=text, cache_control=CacheControl()) return {"type": "text", "text": text, "cache_control": {"type": "ephemeral"}}
def _user_turn(text: str, *, cached: bool = False) -> RichMessage: def _user_turn(text: str, *, cached: bool = False) -> MessageParam:
block = TextBlock(text=text, cache_control=CacheControl() if cached else None) block: TextBlockParam = (
return RichMessage(role="user", content=[block]) {"type": "text", "text": text, "cache_control": {"type": "ephemeral"}}
if cached
else {"type": "text", "text": text}
)
return {"role": "user", "content": [block]}
def _system_reminder_turn() -> RichMessage: def _system_reminder_turn() -> MessageParam:
return RichMessage( return cast(
role="system", "MessageParam",
content=[ {
TextBlock( "role": "system",
text="<system-reminder>Answer with exactly one word.</system-reminder>" "content": [
) {
], "type": "text",
"text": "<system-reminder>Answer with exactly one word.</system-reminder>",
}
],
},
) )
def _post_messages( def _text(message: Message) -> str:
client: EndpointsClient, key: str, body: RichMessagesRequest return "".join(block.text for block in message.content if block.type == "text")
) -> Result[MessagesResult]:
return client.proxy.transport.post(
"/v1/messages",
headers=client.proxy.transport.bearer(key),
json=body,
response_type=MessagesResult,
)
def _register_invoke_deployment( def _register_invoke_deployment(
client: EndpointsClient, resources: ResourceManager, bedrock_model: str proxy: ProxyClient, resources: ResourceManager, bedrock_model: str
) -> str: ) -> str:
model = f"e2e-midsys-{unique_marker()}" model = f"e2e-midsys-{unique_marker()}"
model_id = client.create_model( model_id = proxy.create_model(
model, LiteLLMParamsBody(model=bedrock_model, aws_region_name=AWS_REGION) model, LiteLLMParamsBody(model=bedrock_model, aws_region_name=AWS_REGION)
) )
resources.defer(lambda: client.delete_model(model_id)) resources.defer(lambda: proxy.delete_model(model_id))
return model return model
@ -114,7 +116,7 @@ class PrimedCache(BaseModel):
def _prime_prompt_cache( def _prime_prompt_cache(
client: EndpointsClient, key: str, model: str, system_block: TextBlock client: Anthropic, model: str, system_block: TextBlockParam
) -> PrimedCache: ) -> PrimedCache:
"""Send first-turn calls (fresh cache-marked user turn each attempt, """Send first-turn calls (fresh cache-marked user turn each attempt,
identical system prefix) until one both reads the system prefix back from identical system prefix) until one both reads the system prefix back from
@ -125,17 +127,19 @@ def _prime_prompt_cache(
deadline = time.monotonic() + CACHE_PRIMING_DEADLINE_SECONDS deadline = time.monotonic() + CACHE_PRIMING_DEADLINE_SECONDS
while True: while True:
user_text = _first_turn_user_text(unique_marker()) user_text = _first_turn_user_text(unique_marker())
body = RichMessagesRequest( usage = client.messages.create(
model=model, model=model,
max_tokens=64,
system=[system_block], system=[system_block],
messages=[_user_turn(user_text, cached=True)], messages=[_user_turn(user_text, cached=True)],
) ).usage
usage = unwrap(_post_messages(client, key, body)).usage read_tokens = usage.cache_read_input_tokens or 0
if usage.cache_read_input_tokens > 0 and usage.cache_creation_input_tokens > 0: creation_tokens = usage.cache_creation_input_tokens or 0
if read_tokens > 0 and creation_tokens > 0:
return PrimedCache( return PrimedCache(
first_user_text=user_text, first_user_text=user_text,
prefix_read_tokens=usage.cache_read_input_tokens, prefix_read_tokens=read_tokens,
first_turn_creation_tokens=usage.cache_creation_input_tokens, first_turn_creation_tokens=creation_tokens,
) )
if time.monotonic() >= deadline: if time.monotonic() >= deadline:
pytest.fail( pytest.fail(
@ -151,32 +155,30 @@ class TestBedrockInvokeMidConversationSystem:
exercised_on=[], exercised_on=[],
) )
def test_flagged_model_keeps_prompt_cache_across_system_reminder( def test_flagged_model_keeps_prompt_cache_across_system_reminder(
self, endpoints_client: EndpointsClient, resources: ResourceManager self, proxy: ProxyClient, resources: ResourceManager, sdk: SdkClients
) -> None: ) -> None:
model = _register_invoke_deployment( model = _register_invoke_deployment(proxy, resources, FLAGGED_INVOKE_MODEL)
endpoints_client, resources, FLAGGED_INVOKE_MODEL client = sdk.anthropic(resources.key(models=[model]))
)
key = resources.key(models=[model])
system_block = _cacheable_system_block(unique_marker()) system_block = _cacheable_system_block(unique_marker())
primed = _prime_prompt_cache(endpoints_client, key, model, system_block) primed = _prime_prompt_cache(client, model, system_block)
reminder_turn_body = RichMessagesRequest( second = client.messages.create(
model=model, model=model,
max_tokens=64,
system=[system_block], system=[system_block],
messages=[ messages=[
_user_turn(primed.first_user_text, cached=True), _user_turn(primed.first_user_text, cached=True),
_system_reminder_turn(), _system_reminder_turn(),
RichMessage(role="assistant", content=[TextBlock(text="OK.")]), {"role": "assistant", "content": [{"type": "text", "text": "OK."}]},
_user_turn("Reply with one word again.", cached=True), _user_turn("Reply with one word again.", cached=True),
], ],
) )
second = unwrap(_post_messages(endpoints_client, key, reminder_turn_body))
assert second.text.strip(), ( assert _text(second).strip(), (
f"{model}: reminder turn returned no completion text" f"{model}: reminder turn returned no completion text"
) )
assert second.usage.cache_read_input_tokens >= primed.full_prefix_tokens, ( assert (second.usage.cache_read_input_tokens or 0) >= primed.full_prefix_tokens, (
f"{model}: turn with a mid-conversation system reminder read " f"{model}: turn with a mid-conversation system reminder read "
f"{second.usage.cache_read_input_tokens} cached tokens, expected at " f"{second.usage.cache_read_input_tokens} cached tokens, expected at "
f"least the {primed.full_prefix_tokens} cached on turn one " f"least the {primed.full_prefix_tokens} cached on turn one "
@ -191,29 +193,27 @@ class TestBedrockInvokeMidConversationSystem:
exercised_on=[], exercised_on=[],
) )
def test_unflagged_model_hoists_system_reminder_and_succeeds( def test_unflagged_model_hoists_system_reminder_and_succeeds(
self, endpoints_client: EndpointsClient, resources: ResourceManager self, proxy: ProxyClient, resources: ResourceManager, sdk: SdkClients
) -> None: ) -> None:
model = _register_invoke_deployment( model = _register_invoke_deployment(proxy, resources, UNFLAGGED_INVOKE_MODEL)
endpoints_client, resources, UNFLAGGED_INVOKE_MODEL client = sdk.anthropic(resources.key(models=[model]))
)
key = resources.key(models=[model])
body = RichMessagesRequest( completion = client.messages.create(
model=model, model=model,
system=[TextBlock(text="You are terse.")], max_tokens=64,
system=[{"type": "text", "text": "You are terse."}],
messages=[ messages=[
_user_turn(f"Say hi. Run {unique_marker()}."), _user_turn(f"Say hi. Run {unique_marker()}."),
_system_reminder_turn(), _system_reminder_turn(),
RichMessage(role="assistant", content=[TextBlock(text="Hi.")]), {"role": "assistant", "content": [{"type": "text", "text": "Hi."}]},
_user_turn("Say bye."), _user_turn("Say bye."),
], ],
) )
completion = unwrap(_post_messages(endpoints_client, key, body))
assert completion.role == "assistant", ( assert completion.role == "assistant", (
f"{model}: unexpected role {completion.role!r}" f"{model}: unexpected role {completion.role!r}"
) )
assert completion.text.strip(), ( assert _text(completion).strip(), (
f"{model}: conversation with a mid-conversation system reminder " f"{model}: conversation with a mid-conversation system reminder "
f"returned no text; the reminder was forwarded in place to a model " f"returned no text; the reminder was forwarded in place to a model "
f"that rejects role 'system' inside messages instead of being hoisted" f"that rejects role 'system' inside messages instead of being hoisted"

136
tests/e2e/llm_translation/test_messages_mid_conversation_system_native_providers_e2e.py
View file

@ -24,27 +24,28 @@ entry whose prefix spans ``system`` plus message turns is invalidated when the
reminder is hoisted (the ``system`` field mutates and a turn disappears from reminder is hoisted (the ``system`` field mutates and a turn disappears from
``messages``), while an entry ending at the system block itself would survive ``messages``), while an entry ending at the system block itself would survive
the hoist and mask the regression. the hoist and mask the regression.
Calls go through the real Anthropic SDK (LIT-4577). The SDK's ``MessageParam``
type only admits user/assistant roles, so the system reminder turn is cast to
it; the SDK serializes the dict verbatim, which is exactly the wire shape under
test.
""" """
from __future__ import annotations from __future__ import annotations
import time import time
from typing import cast
import pytest import pytest
from anthropic import Anthropic
from anthropic.types import Message, MessageParam, TextBlockParam
from pydantic import BaseModel from pydantic import BaseModel
from e2e_config import unique_marker from e2e_config import unique_marker
from e2e_http import Result, unwrap
from endpoints_client import (
CacheControl,
EndpointsClient,
MessagesResult,
RichMessage,
RichMessagesRequest,
TextBlock,
)
from lifecycle import ResourceManager from lifecycle import ResourceManager
from models import LiteLLMParamsBody from models import LiteLLMParamsBody
from proxy_client import ProxyClient
from sdk_clients import SdkClients
pytestmark = pytest.mark.e2e pytestmark = pytest.mark.e2e
@ -68,40 +69,47 @@ def _vertex_params(model: str) -> LiteLLMParamsBody:
) )
def _cacheable_system_block(marker: str) -> TextBlock: def _cacheable_system_block(marker: str) -> TextBlockParam:
"""A system prompt comfortably above the 1024-token minimum cacheable size, """A system prompt comfortably above the 1024-token minimum cacheable size,
unique per run so no other run's cache entry can satisfy the read.""" unique per run so no other run's cache entry can satisfy the read."""
text = " ".join(f"Reference paragraph {index} for run {marker}." for index in range(300)) text = " ".join(f"Reference paragraph {index} for run {marker}." for index in range(300))
return TextBlock(text=text, cache_control=CacheControl()) return {"type": "text", "text": text, "cache_control": {"type": "ephemeral"}}
def _user_turn(text: str, *, cached: bool = False) -> RichMessage: def _user_turn(text: str, *, cached: bool = False) -> MessageParam:
block = TextBlock(text=text, cache_control=CacheControl() if cached else None) block: TextBlockParam = (
return RichMessage(role="user", content=[block]) {"type": "text", "text": text, "cache_control": {"type": "ephemeral"}}
if cached
else {"type": "text", "text": text}
)
return {"role": "user", "content": [block]}
def _system_reminder_turn() -> RichMessage: def _system_reminder_turn() -> MessageParam:
return RichMessage( return cast(
role="system", "MessageParam",
content=[TextBlock(text="<system-reminder>Answer with exactly one word.</system-reminder>")], {
"role": "system",
"content": [
{
"type": "text",
"text": "<system-reminder>Answer with exactly one word.</system-reminder>",
}
],
},
) )
def _post_messages(client: EndpointsClient, key: str, body: RichMessagesRequest) -> Result[MessagesResult]: def _text(message: Message) -> str:
return client.proxy.transport.post( return "".join(block.text for block in message.content if block.type == "text")
"/v1/messages",
headers=client.proxy.transport.bearer(key),
json=body,
response_type=MessagesResult,
)
def _register_deployment( def _register_deployment(
client: EndpointsClient, resources: ResourceManager, params: LiteLLMParamsBody proxy: ProxyClient, resources: ResourceManager, params: LiteLLMParamsBody
) -> str: ) -> str:
model = f"e2e-midsys-{unique_marker()}" model = f"e2e-midsys-{unique_marker()}"
model_id = client.create_model(model, params) model_id = proxy.create_model(model, params)
resources.defer(lambda: client.delete_model(model_id)) resources.defer(lambda: proxy.delete_model(model_id))
return model return model
@ -124,7 +132,7 @@ class PrimedCache(BaseModel):
def _prime_prompt_cache( def _prime_prompt_cache(
client: EndpointsClient, key: str, model: str, system_block: TextBlock client: Anthropic, model: str, system_block: TextBlockParam
) -> PrimedCache: ) -> PrimedCache:
"""Send first-turn calls (fresh cache-marked user turn each attempt, """Send first-turn calls (fresh cache-marked user turn each attempt,
identical system prefix) until one both reads the system prefix back from identical system prefix) until one both reads the system prefix back from
@ -135,17 +143,19 @@ def _prime_prompt_cache(
deadline = time.monotonic() + CACHE_PRIMING_DEADLINE_SECONDS deadline = time.monotonic() + CACHE_PRIMING_DEADLINE_SECONDS
while True: while True:
user_text = _first_turn_user_text(unique_marker()) user_text = _first_turn_user_text(unique_marker())
body = RichMessagesRequest( usage = client.messages.create(
model=model, model=model,
max_tokens=64,
system=[system_block], system=[system_block],
messages=[_user_turn(user_text, cached=True)], messages=[_user_turn(user_text, cached=True)],
) ).usage
usage = unwrap(_post_messages(client, key, body)).usage read_tokens = usage.cache_read_input_tokens or 0
if usage.cache_read_input_tokens > 0 and usage.cache_creation_input_tokens > 0: creation_tokens = usage.cache_creation_input_tokens or 0
if read_tokens > 0 and creation_tokens > 0:
return PrimedCache( return PrimedCache(
first_user_text=user_text, first_user_text=user_text,
prefix_read_tokens=usage.cache_read_input_tokens, prefix_read_tokens=read_tokens,
first_turn_creation_tokens=usage.cache_creation_input_tokens, first_turn_creation_tokens=creation_tokens,
) )
if time.monotonic() >= deadline: if time.monotonic() >= deadline:
pytest.fail( pytest.fail(
@ -156,28 +166,31 @@ def _prime_prompt_cache(
def _assert_flagged_model_keeps_cache( def _assert_flagged_model_keeps_cache(
client: EndpointsClient, resources: ResourceManager, params: LiteLLMParamsBody proxy: ProxyClient,
resources: ResourceManager,
sdk: SdkClients,
params: LiteLLMParamsBody,
) -> None: ) -> None:
model = _register_deployment(client, resources, params) model = _register_deployment(proxy, resources, params)
key = resources.key(models=[model]) client = sdk.anthropic(resources.key(models=[model]))
system_block = _cacheable_system_block(unique_marker()) system_block = _cacheable_system_block(unique_marker())
primed = _prime_prompt_cache(client, key, model, system_block) primed = _prime_prompt_cache(client, model, system_block)
reminder_turn_body = RichMessagesRequest( second = client.messages.create(
model=model, model=model,
max_tokens=64,
system=[system_block], system=[system_block],
messages=[ messages=[
_user_turn(primed.first_user_text, cached=True), _user_turn(primed.first_user_text, cached=True),
_system_reminder_turn(), _system_reminder_turn(),
RichMessage(role="assistant", content=[TextBlock(text="OK.")]), {"role": "assistant", "content": [{"type": "text", "text": "OK."}]},
_user_turn("Reply with one word again.", cached=True), _user_turn("Reply with one word again.", cached=True),
], ],
) )
second = unwrap(_post_messages(client, key, reminder_turn_body))
assert second.text.strip(), f"{model}: reminder turn returned no completion text" assert _text(second).strip(), f"{model}: reminder turn returned no completion text"
assert second.usage.cache_read_input_tokens >= primed.full_prefix_tokens, ( assert (second.usage.cache_read_input_tokens or 0) >= primed.full_prefix_tokens, (
f"{model}: turn with a mid-conversation system reminder read " f"{model}: turn with a mid-conversation system reminder read "
f"{second.usage.cache_read_input_tokens} cached tokens, expected at " f"{second.usage.cache_read_input_tokens} cached tokens, expected at "
f"least the {primed.full_prefix_tokens} cached on turn one " f"least the {primed.full_prefix_tokens} cached on turn one "
@ -189,25 +202,28 @@ def _assert_flagged_model_keeps_cache(
def _assert_unflagged_model_hoists_and_succeeds( def _assert_unflagged_model_hoists_and_succeeds(
client: EndpointsClient, resources: ResourceManager, params: LiteLLMParamsBody proxy: ProxyClient,
resources: ResourceManager,
sdk: SdkClients,
params: LiteLLMParamsBody,
) -> None: ) -> None:
model = _register_deployment(client, resources, params) model = _register_deployment(proxy, resources, params)
key = resources.key(models=[model]) client = sdk.anthropic(resources.key(models=[model]))
body = RichMessagesRequest( completion = client.messages.create(
model=model, model=model,
system=[TextBlock(text="You are terse.")], max_tokens=64,
system=[{"type": "text", "text": "You are terse."}],
messages=[ messages=[
_user_turn(f"Say hi. Run {unique_marker()}."), _user_turn(f"Say hi. Run {unique_marker()}."),
_system_reminder_turn(), _system_reminder_turn(),
RichMessage(role="assistant", content=[TextBlock(text="Hi.")]), {"role": "assistant", "content": [{"type": "text", "text": "Hi."}]},
_user_turn("Say bye."), _user_turn("Say bye."),
], ],
) )
completion = unwrap(_post_messages(client, key, body))
assert completion.role == "assistant", f"{model}: unexpected role {completion.role!r}" assert completion.role == "assistant", f"{model}: unexpected role {completion.role!r}"
assert completion.text.strip(), ( assert _text(completion).strip(), (
f"{model}: conversation with a mid-conversation system reminder returned " f"{model}: conversation with a mid-conversation system reminder returned "
f"no text; the reminder was forwarded in place to a model that rejects " f"no text; the reminder was forwarded in place to a model that rejects "
f"role 'system' inside messages instead of being hoisted" f"role 'system' inside messages instead of being hoisted"
@ -223,19 +239,19 @@ class TestAzureFoundryMidConversationSystem:
exercised_on=[], exercised_on=[],
) )
def test_flagged_model_keeps_prompt_cache_across_system_reminder( def test_flagged_model_keeps_prompt_cache_across_system_reminder(
self, endpoints_client: EndpointsClient, resources: ResourceManager self, proxy: ProxyClient, resources: ResourceManager, sdk: SdkClients
) -> None: ) -> None:
_assert_flagged_model_keeps_cache(endpoints_client, resources, _azure_params(self.FLAGGED_MODEL)) _assert_flagged_model_keeps_cache(proxy, resources, sdk, _azure_params(self.FLAGGED_MODEL))
@pytest.mark.covers( @pytest.mark.covers(
"llm.messages.azure_foundry.mid_conversation_system.nonstream.works", "llm.messages.azure_foundry.mid_conversation_system.nonstream.works",
exercised_on=[], exercised_on=[],
) )
def test_unflagged_model_hoists_system_reminder_and_succeeds( def test_unflagged_model_hoists_system_reminder_and_succeeds(
self, endpoints_client: EndpointsClient, resources: ResourceManager self, proxy: ProxyClient, resources: ResourceManager, sdk: SdkClients
) -> None: ) -> None:
_assert_unflagged_model_hoists_and_succeeds( _assert_unflagged_model_hoists_and_succeeds(
endpoints_client, resources, _azure_params(self.UNFLAGGED_MODEL) proxy, resources, sdk, _azure_params(self.UNFLAGGED_MODEL)
) )
@ -248,17 +264,17 @@ class TestVertexMidConversationSystem:
exercised_on=[], exercised_on=[],
) )
def test_flagged_model_keeps_prompt_cache_across_system_reminder( def test_flagged_model_keeps_prompt_cache_across_system_reminder(
self, endpoints_client: EndpointsClient, resources: ResourceManager self, proxy: ProxyClient, resources: ResourceManager, sdk: SdkClients
) -> None: ) -> None:
_assert_flagged_model_keeps_cache(endpoints_client, resources, _vertex_params(self.FLAGGED_MODEL)) _assert_flagged_model_keeps_cache(proxy, resources, sdk, _vertex_params(self.FLAGGED_MODEL))
@pytest.mark.covers( @pytest.mark.covers(
"llm.messages.vertex.mid_conversation_system.nonstream.works", "llm.messages.vertex.mid_conversation_system.nonstream.works",
exercised_on=[], exercised_on=[],
) )
def test_unflagged_model_hoists_system_reminder_and_succeeds( def test_unflagged_model_hoists_system_reminder_and_succeeds(
self, endpoints_client: EndpointsClient, resources: ResourceManager self, proxy: ProxyClient, resources: ResourceManager, sdk: SdkClients
) -> None: ) -> None:
_assert_unflagged_model_hoists_and_succeeds( _assert_unflagged_model_hoists_and_succeeds(
endpoints_client, resources, _vertex_params(self.UNFLAGGED_MODEL) proxy, resources, sdk, _vertex_params(self.UNFLAGGED_MODEL)
) )

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

@ -1,19 +1,22 @@
"""Live e2e: POST /v1/moderations classifies content against the provider policy. """Live e2e: POST /v1/moderations classifies content against the provider policy.
Registers OpenAI's omni moderation model at runtime and asserts the product Registers OpenAI's omni moderation model at runtime, drives it through the real
promise on both sides of the decision: clearly violent text comes back flagged OpenAI SDK (LIT-4577), and asserts the product promise on both sides of the
with at least one policy category tripped, and benign text comes back not flagged. decision: clearly violent text comes back flagged with at least one policy
category tripped, and benign text comes back not flagged.
""" """
from __future__ import annotations from __future__ import annotations
import pytest import pytest
from openai.types import Moderation
from pydantic import TypeAdapter
from e2e_config import unique_marker from e2e_config import unique_marker
from e2e_http import unwrap
from endpoints_client import EndpointsClient
from lifecycle import ResourceManager from lifecycle import ResourceManager
from models import LiteLLMParamsBody from models import LiteLLMParamsBody
from proxy_client import ProxyClient
from sdk_clients import SdkClients
pytestmark = pytest.mark.e2e pytestmark = pytest.mark.e2e
@ -21,45 +24,49 @@ VIOLENT_TEXT = "I am going to find you and kill you, and I will hurt everyone yo
BENIGN_TEXT = "I enjoyed the sunny afternoon and a relaxing walk in the park today." BENIGN_TEXT = "I enjoyed the sunny afternoon and a relaxing walk in the park today."
def _register_moderation_model( def _register_moderation_model(proxy: ProxyClient, resources: ResourceManager) -> str:
endpoints_client: EndpointsClient, resources: ResourceManager
) -> str:
model = f"e2e-moderation-{unique_marker()}" model = f"e2e-moderation-{unique_marker()}"
model_id = endpoints_client.create_model( model_id = proxy.create_model(
model, model,
LiteLLMParamsBody( LiteLLMParamsBody(
model="openai/omni-moderation-latest", api_key="os.environ/OPENAI_API_KEY" model="openai/omni-moderation-latest", api_key="os.environ/OPENAI_API_KEY"
), ),
) )
resources.defer(lambda: endpoints_client.delete_model(model_id)) resources.defer(lambda: proxy.delete_model(model_id))
return model return model
_CATEGORY_FLAGS = TypeAdapter(dict[str, bool | None])
def _flagged_categories(item: Moderation) -> tuple[str, ...]:
flags = _CATEGORY_FLAGS.validate_python(item.categories.model_dump())
return tuple(name for name, hit in flags.items() if hit)
class TestModerations: class TestModerations:
@pytest.mark.covers("llm.moderations.openai.basic.nonstream.works") @pytest.mark.covers("llm.moderations.openai.basic.nonstream.works")
def test_moderations_flags_violent_content( def test_moderations_flags_violent_content(
self, endpoints_client: EndpointsClient, resources: ResourceManager self, proxy: ProxyClient, resources: ResourceManager, sdk: SdkClients
) -> None: ) -> None:
model = _register_moderation_model(endpoints_client, resources) model = _register_moderation_model(proxy, resources)
key = resources.key() client = sdk.openai(resources.key())
result = unwrap(endpoints_client.moderations(key, model, VIOLENT_TEXT)) moderation = client.moderations.create(model=model, input=VIOLENT_TEXT)
item = result.first assert moderation.results, f"/moderations returned no results: {moderation!r}"
assert item is not None, f"/moderations returned no results: {result}" item = moderation.results[0]
assert item.flagged, f"violent text was not flagged: {item}" assert item.flagged, f"violent text was not flagged: {item!r}"
assert item.flagged_categories, ( assert _flagged_categories(item), f"flagged result reported no true category: {item!r}"
f"flagged result reported no true category: {item}"
)
def test_moderations_passes_benign_content( def test_moderations_passes_benign_content(
self, endpoints_client: EndpointsClient, resources: ResourceManager self, proxy: ProxyClient, resources: ResourceManager, sdk: SdkClients
) -> None: ) -> None:
model = _register_moderation_model(endpoints_client, resources) model = _register_moderation_model(proxy, resources)
key = resources.key() client = sdk.openai(resources.key())
result = unwrap(endpoints_client.moderations(key, model, BENIGN_TEXT)) moderation = client.moderations.create(model=model, input=BENIGN_TEXT)
item = result.first assert moderation.results, f"/moderations returned no results: {moderation!r}"
assert item is not None, f"/moderations returned no results: {result}" item = moderation.results[0]
assert not item.flagged, ( assert not item.flagged, (
f"benign text was flagged as {item.flagged_categories}: {item}" f"benign text was flagged as {_flagged_categories(item)}: {item!r}"
) )

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

@ -22,9 +22,9 @@ import pytest
from e2e_config import unique_marker from e2e_config import unique_marker
from e2e_http import unwrap from e2e_http import unwrap
from endpoints_client import EndpointsClient
from lifecycle import ResourceManager from lifecycle import ResourceManager
from models import LiteLLMParamsBody, OcrBody, OcrDocument, OcrResponse from models import LiteLLMParamsBody, OcrBody, OcrDocument, OcrResponse
from proxy_client import ProxyClient
pytestmark = pytest.mark.e2e pytestmark = pytest.mark.e2e
@ -143,14 +143,14 @@ def _assert_ocr_document(response: OcrResponse) -> None:
class TestRustOcrGateway: class TestRustOcrGateway:
@pytest.mark.parametrize("case", RUST_OCR_CASES, ids=_CASE_IDS) @pytest.mark.parametrize("case", RUST_OCR_CASES, ids=_CASE_IDS)
def test_rust_ocr_response( def test_rust_ocr_response(
self, endpoints_client: EndpointsClient, resources: ResourceManager, case: _OcrCase self, proxy: ProxyClient, resources: ResourceManager, case: _OcrCase
) -> None: ) -> None:
model = f"rust-ocr-{case.suffix}-{unique_marker()}" model = f"rust-ocr-{case.suffix}-{unique_marker()}"
model_id = endpoints_client.create_model(model, case.provider.litellm_params()) model_id = proxy.create_model(model, case.provider.litellm_params())
resources.defer(lambda: endpoints_client.delete_model(model_id)) resources.defer(lambda: proxy.delete_model(model_id))
key = resources.key() key = resources.key()
response = unwrap(endpoints_client.proxy.ocr(key, OcrBody(model=model, document=case.document))) response = unwrap(proxy.ocr(key, OcrBody(model=model, document=case.document)))
_assert_ocr_document(response) _assert_ocr_document(response)

8
tests/e2e/llm_translation/test_passthrough_headers_e2e.py
View file

@ -18,9 +18,8 @@ from pydantic import BaseModel, Field
from e2e_config import unique_marker from e2e_config import unique_marker
from e2e_http import AuthHeaders, NoBody, require_successful_call, unwrap from e2e_http import AuthHeaders, NoBody, require_successful_call, unwrap
from endpoints_client import MessagesResult
from lifecycle import ResourceManager from lifecycle import ResourceManager
from models import ChatMessage, KeyGenerateBody from models import AnthropicMessagesResponse, ChatMessage, KeyGenerateBody
from passthrough_client import PassthroughClient from passthrough_client import PassthroughClient
pytestmark = pytest.mark.e2e pytestmark = pytest.mark.e2e
@ -128,8 +127,9 @@ class TestPassthroughHeaders:
json=_messages_body(), json=_messages_body(),
) )
require_successful_call(result) require_successful_call(result)
completion = MessagesResult.model_validate_json(result.body) completion = AnthropicMessagesResponse.model_validate_json(result.body)
assert completion.text.strip(), ( text = "".join(block.text or "" for block in (completion.content or []))
assert text.strip(), (
f"static x-api-key must reach Anthropic for the call to succeed at all; got {result.body[:300]}" f"static x-api-key must reach Anthropic for the call to succeed at all; got {result.body[:300]}"
) )

48
tests/e2e/llm_translation/test_rerank_e2e.py
View file

@ -1,8 +1,9 @@
"""Live e2e: POST /v1/rerank ranks documents by relevance. """Live e2e: POST /v1/rerank ranks documents by relevance.
Registers a Cohere rerank deployment at runtime and asserts the endpoint returns Registers a Cohere rerank deployment at runtime and asserts the endpoint returns
scored results within the requested top_n. Migrated from scored results within the requested top_n. No official OpenAI/Anthropic SDK
litellm-regression-tests/tests/test_inference_endpoints.py. covers /v1/rerank, so the call rides the shared typed transport via
ProxyClient.rerank.
""" """
from __future__ import annotations from __future__ import annotations
@ -10,10 +11,10 @@ from __future__ import annotations
import pytest import pytest
from e2e_config import require_env, unique_marker from e2e_config import require_env, unique_marker
from e2e_http import require_successful_call from e2e_http import unwrap
from endpoints_client import EndpointsClient, RerankResult
from lifecycle import ResourceManager from lifecycle import ResourceManager
from models import LiteLLMParamsBody from models import LiteLLMParamsBody, RerankBody, RerankResponse
from proxy_client import ProxyClient
pytestmark = pytest.mark.e2e pytestmark = pytest.mark.e2e
@ -26,39 +27,39 @@ DOCUMENTS = [
QUERY = "What is the capital of the United States?" QUERY = "What is the capital of the United States?"
def _assert_top_n_scored(body: str) -> None: def _assert_top_n_scored(response: RerankResponse) -> None:
parsed = RerankResult.model_validate_json(body) assert response.results, f"/rerank returned no results: {response!r}"
assert parsed.results, f"/rerank returned no results: {body[:300]}" assert len(response.results) <= 3, f"top_n=3 not honored: {response!r}"
assert len(parsed.results) <= 3, f"top_n=3 not honored: {body[:300]}" assert response.results[0].relevance_score is not None, (
assert parsed.results[0].relevance_score is not None, ( f"top rerank result has no relevance_score: {response!r}"
f"top rerank result has no relevance_score: {body[:300]}"
) )
class TestRerank: class TestRerank:
@pytest.mark.covers("llm.rerank.cohere.basic.nonstream.works") @pytest.mark.covers("llm.rerank.cohere.basic.nonstream.works")
def test_rerank_scores_top_n( def test_rerank_scores_top_n(
self, endpoints_client: EndpointsClient, resources: ResourceManager self, proxy: ProxyClient, resources: ResourceManager
) -> None: ) -> None:
model = f"e2e-rerank-{unique_marker()}" model = f"e2e-rerank-{unique_marker()}"
model_id = endpoints_client.create_model( model_id = proxy.create_model(
model, model,
LiteLLMParamsBody(model="cohere/rerank-v3.5", api_key="os.environ/COHERE_API_KEY"), LiteLLMParamsBody(model="cohere/rerank-v3.5", api_key="os.environ/COHERE_API_KEY"),
) )
resources.defer(lambda: endpoints_client.delete_model(model_id)) resources.defer(lambda: proxy.delete_model(model_id))
key = resources.key() key = resources.key()
result = endpoints_client.rerank(key, model, QUERY, DOCUMENTS, top_n=3) response = unwrap(
require_successful_call(result) proxy.rerank(key, RerankBody(model=model, query=QUERY, documents=DOCUMENTS, top_n=3))
_assert_top_n_scored(result.body) )
_assert_top_n_scored(response)
@pytest.mark.covers("llm.rerank.bedrock.basic.nonstream.works", exercised_on=["rerank"]) @pytest.mark.covers("llm.rerank.bedrock.basic.nonstream.works", exercised_on=["rerank"])
def test_bedrock_rerank_scores_top_n( def test_bedrock_rerank_scores_top_n(
self, endpoints_client: EndpointsClient, resources: ResourceManager self, proxy: ProxyClient, resources: ResourceManager
) -> None: ) -> None:
require_env("AWS_ACCESS_KEY_ID", "AWS_SECRET_ACCESS_KEY", "AWS_REGION") require_env("AWS_ACCESS_KEY_ID", "AWS_SECRET_ACCESS_KEY", "AWS_REGION")
model = f"e2e-bedrock-rerank-{unique_marker()}" model = f"e2e-bedrock-rerank-{unique_marker()}"
model_id = endpoints_client.create_model( model_id = proxy.create_model(
model, model,
LiteLLMParamsBody( LiteLLMParamsBody(
model="bedrock/amazon.rerank-v1:0", model="bedrock/amazon.rerank-v1:0",
@ -67,9 +68,10 @@ class TestRerank:
aws_region_name="os.environ/AWS_REGION", aws_region_name="os.environ/AWS_REGION",
), ),
) )
resources.defer(lambda: endpoints_client.delete_model(model_id)) resources.defer(lambda: proxy.delete_model(model_id))
key = resources.key() key = resources.key()
result = endpoints_client.rerank(key, model, QUERY, DOCUMENTS, top_n=3) response = unwrap(
require_successful_call(result) proxy.rerank(key, RerankBody(model=model, query=QUERY, documents=DOCUMENTS, top_n=3))
_assert_top_n_scored(result.body) )
_assert_top_n_scored(response)

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

@ -1,8 +1,8 @@
"""Live e2e: POST /v1/responses returns a real completion. """Live e2e: POST /v1/responses returns a real completion.
Registers an OpenAI deployment at runtime, drives the Responses API through the Registers an OpenAI deployment at runtime and drives the Responses API through
gateway, and asserts output text came back. Migrated from the gateway with the real OpenAI SDK, the client customers actually use
litellm-regression-tests/tests/test_inference_endpoints.py. (LIT-4577), asserting output text came back.
""" """
from __future__ import annotations from __future__ import annotations
@ -11,34 +11,45 @@ import json
from typing import cast from typing import cast
import pytest import pytest
from pydantic import BaseModel, ValidationError from openai.types.responses import (
FunctionToolParam,
Response,
ResponseFunctionToolCall,
ResponseInputParam,
)
from pydantic import BaseModel
from e2e_config import require_env, unique_marker from e2e_config import require_env, unique_marker
from e2e_http import require_successful_call
from endpoints_client import (
EndpointsClient,
FunctionParameterProperty,
FunctionParameters,
ResponsesFunctionTool,
ResponsesOutputTextDeltaEvent,
ResponsesResult,
ResponsesStreamEventType,
)
from lifecycle import ResourceManager from lifecycle import ResourceManager
from models import LiteLLMParamsBody from models import LiteLLMParamsBody
from proxy_client import ProxyClient
from sdk_clients import SdkClients
pytestmark = pytest.mark.e2e pytestmark = pytest.mark.e2e
BEDROCK_CONVERSE_BACKEND = "bedrock/us.anthropic.claude-haiku-4-5-20251001-v1:0" BEDROCK_CONVERSE_BACKEND = "bedrock/us.anthropic.claude-haiku-4-5-20251001-v1:0"
INSTRUCTIONS = "You are a helpful assistant"
CAT_IMAGE_URL = "https://upload.wikimedia.org/wikipedia/commons/3/3a/Cat03.jpg"
WEATHER_TOOL = ResponsesFunctionTool( WEATHER_TOOL: FunctionToolParam = {
name="get_weather", "type": "function",
description="Get the weather for a location", "name": "get_weather",
parameters=FunctionParameters( "description": "Get the weather for a location",
properties={"location": FunctionParameterProperty(type="string")}, "parameters": {
required=["location"], "type": "object",
), "properties": {"location": {"type": "string"}},
) "required": ["location"],
},
"strict": False,
}
def _openai_params() -> LiteLLMParamsBody:
return LiteLLMParamsBody(model="openai/gpt-4o-mini", api_key="os.environ/OPENAI_API_KEY")
def _anthropic_params() -> LiteLLMParamsBody:
return LiteLLMParamsBody(model="anthropic/claude-haiku-4-5", api_key="os.environ/ANTHROPIC_API_KEY")
def _bedrock_params() -> LiteLLMParamsBody: def _bedrock_params() -> LiteLLMParamsBody:
@ -50,6 +61,25 @@ def _bedrock_params() -> LiteLLMParamsBody:
) )
def _register(proxy: ProxyClient, resources: ResourceManager, params: LiteLLMParamsBody) -> str:
model = f"e2e-responses-{unique_marker()}"
model_id = proxy.create_model(model, params)
resources.defer(lambda: proxy.delete_model(model_id))
return model
def _function_calls(response: Response) -> tuple[ResponseFunctionToolCall, ...]:
return tuple(item for item in response.output if isinstance(item, ResponseFunctionToolCall))
def _assert_weather_call(response: Response) -> None:
function_call = next((call for call in _function_calls(response) if call.name == "get_weather"), None)
assert function_call is not None, f"no get_weather function call: {response.output!r}"
raw_arguments = cast(object, json.loads(function_call.arguments))
arguments = WeatherArguments.model_validate(raw_arguments)
assert arguments.location, f"function call arguments missing location: {function_call.arguments}"
class WeatherArguments(BaseModel): class WeatherArguments(BaseModel):
location: str location: str
@ -57,242 +87,157 @@ class WeatherArguments(BaseModel):
class TestResponses: class TestResponses:
@pytest.mark.covers("llm.responses.openai.basic.nonstream.works") @pytest.mark.covers("llm.responses.openai.basic.nonstream.works")
def test_responses_returns_completion( def test_responses_returns_completion(
self, endpoints_client: EndpointsClient, resources: ResourceManager self, proxy: ProxyClient, resources: ResourceManager, sdk: SdkClients
) -> None: ) -> None:
model = f"e2e-responses-{unique_marker()}" model = _register(proxy, resources, _openai_params())
model_id = endpoints_client.create_model( client = sdk.openai(resources.key())
model,
LiteLLMParamsBody(model="openai/gpt-4o-mini", api_key="os.environ/OPENAI_API_KEY"),
)
resources.defer(lambda: endpoints_client.delete_model(model_id))
key = resources.key()
result = endpoints_client.responses(key, model, "reply with one word") response = client.responses.create(
require_successful_call(result) model=model, input="reply with one word", instructions=INSTRUCTIONS
parsed = ResponsesResult.model_validate_json(result.body) )
assert parsed.text.strip(), f"/responses returned no output text: {result.body[:300]}" assert response.output_text.strip(), f"/responses returned no output text: {response.output!r}"
@pytest.mark.covers("llm.responses.openai.basic.stream.works") @pytest.mark.covers("llm.responses.openai.basic.stream.works")
def test_responses_streaming_returns_completion( def test_responses_streaming_returns_completion(
self, endpoints_client: EndpointsClient, resources: ResourceManager self, proxy: ProxyClient, resources: ResourceManager, sdk: SdkClients
) -> None: ) -> None:
model = f"e2e-responses-{unique_marker()}" model = _register(proxy, resources, _openai_params())
model_id = endpoints_client.create_model( client = sdk.openai(resources.key())
model,
LiteLLMParamsBody(model="openai/gpt-4o-mini", api_key="os.environ/OPENAI_API_KEY"),
)
resources.defer(lambda: endpoints_client.delete_model(model_id))
key = resources.key()
result = endpoints_client.responses(key, model, "reply with one word", stream=True) stream = client.responses.create(
require_successful_call(result) model=model, input="reply with one word", instructions=INSTRUCTIONS, stream=True
delta_events = tuple( )
parsed events = list(stream)
for event in result.stream_events assert events, "responses stream returned no events"
if (parsed := _parse_stream_event(event)) is not None deltas = [event.delta for event in events if event.type == "response.output_text.delta"]
assert any(delta for delta in deltas), "responses stream returned no text deltas"
assert events[-1].type == "response.completed", (
f"responses stream did not terminate with response.completed: {events[-1].type}"
) )
assert any(event.delta for event in delta_events), "responses stream returned no text deltas"
assert result.stream_events, "responses stream returned no events"
assert (
ResponsesStreamEventType.model_validate_json(result.stream_events[-1]).type
== "response.completed"
), "responses stream did not terminate with response.completed"
@pytest.mark.covers("llm.responses.openai.basic.nonstream.cost_logged") @pytest.mark.covers("llm.responses.openai.basic.nonstream.cost_logged")
def test_responses_logs_cost( def test_responses_logs_cost(
self, endpoints_client: EndpointsClient, resources: ResourceManager self, proxy: ProxyClient, resources: ResourceManager, sdk: SdkClients
) -> None: ) -> None:
model = f"e2e-responses-{unique_marker()}" model = _register(proxy, resources, _openai_params())
model_id = endpoints_client.create_model( client = sdk.openai(resources.key())
model,
LiteLLMParamsBody(model="openai/gpt-4o-mini", api_key="os.environ/OPENAI_API_KEY"), raw = client.responses.with_raw_response.create(
model=model, input=f"reply with one word {unique_marker()}", instructions=INSTRUCTIONS
)
response = raw.parse()
assert response.output_text.strip(), f"/responses returned no output text: {response.output!r}"
assert raw.headers.get("x-litellm-call-id") and response.id, (
f"missing response identifiers: id={response.id!r}, headers={dict(raw.headers)}"
) )
resources.defer(lambda: endpoints_client.delete_model(model_id))
key = resources.key()
result = endpoints_client.responses(key, model, f"reply with one word {unique_marker()}") rows = proxy.poll_logs_for_request_id(
require_successful_call(result) response.id,
parsed = ResponsesResult.model_validate_json(result.body)
assert parsed.text.strip(), f"/responses returned no output text: {result.body[:300]}"
assert result.call_id and parsed.id, f"missing response identifiers: {result.body[:300]}"
rows = endpoints_client.proxy.poll_logs_for_request_id(
parsed.id,
predicate=lambda logged_rows: any((row.spend or 0) > 0 for row in logged_rows), predicate=lambda logged_rows: any((row.spend or 0) > 0 for row in logged_rows),
) )
row = next((logged_row for logged_row in rows if (logged_row.spend or 0) > 0), None) row = next((logged_row for logged_row in rows if (logged_row.spend or 0) > 0), None)
assert row is not None, f"no costed spend row for response id {parsed.id}" assert row is not None, f"no costed spend row for response id {response.id}"
assert "gpt-4o-mini" in (row.model or ""), f"unexpected spend row model: {row.model}" assert "gpt-4o-mini" in (row.model or ""), f"unexpected spend row model: {row.model}"
@pytest.mark.covers("llm.responses.openai.tool_use.nonstream.works") @pytest.mark.covers("llm.responses.openai.tool_use.nonstream.works")
def test_responses_returns_function_call( def test_responses_returns_function_call(
self, endpoints_client: EndpointsClient, resources: ResourceManager self, proxy: ProxyClient, resources: ResourceManager, sdk: SdkClients
) -> None: ) -> None:
model = f"e2e-responses-{unique_marker()}" model = _register(proxy, resources, _openai_params())
model_id = endpoints_client.create_model( client = sdk.openai(resources.key())
model,
LiteLLMParamsBody(model="openai/gpt-4o-mini", api_key="os.environ/OPENAI_API_KEY"),
)
resources.defer(lambda: endpoints_client.delete_model(model_id))
key = resources.key()
result = endpoints_client.responses_with_tools( response = client.responses.create(
key, model=model,
model, input="What is the weather in San Francisco? Use the get_weather tool.",
"What is the weather in San Francisco? Use the get_weather tool.", instructions=INSTRUCTIONS,
[ tools=[WEATHER_TOOL],
ResponsesFunctionTool(
name="get_weather",
description="Get the weather for a location",
parameters=FunctionParameters(
properties={"location": FunctionParameterProperty(type="string")},
required=["location"],
),
)
],
) )
require_successful_call(result) _assert_weather_call(response)
parsed = ResponsesResult.model_validate_json(result.body)
function_call = next(
(call for call in parsed.function_calls if call.name == "get_weather"),
None,
)
assert function_call is not None, f"no get_weather function call: {result.body[:500]}"
assert function_call.arguments is not None
raw_arguments = cast(object, json.loads(function_call.arguments))
arguments = WeatherArguments.model_validate(raw_arguments)
assert arguments.location, f"function call arguments missing location: {function_call.arguments}"
@pytest.mark.covers("llm.responses.openai.vision.nonstream.works") @pytest.mark.covers("llm.responses.openai.vision.nonstream.works")
def test_responses_vision_describes_image( def test_responses_vision_describes_image(
self, endpoints_client: EndpointsClient, resources: ResourceManager self, proxy: ProxyClient, resources: ResourceManager, sdk: SdkClients
) -> None: ) -> None:
model = f"e2e-responses-{unique_marker()}" model = _register(
model_id = endpoints_client.create_model( proxy,
model, resources,
LiteLLMParamsBody(model="openai/gpt-4o", api_key="os.environ/OPENAI_API_KEY"), LiteLLMParamsBody(model="openai/gpt-4o", api_key="os.environ/OPENAI_API_KEY"),
) )
resources.defer(lambda: endpoints_client.delete_model(model_id)) client = sdk.openai(resources.key())
key = resources.key()
result = endpoints_client.responses_vision( vision_input: ResponseInputParam = [
key, {
model, "role": "user",
"What animal is shown in this image? Answer in one word", "content": [
"https://upload.wikimedia.org/wikipedia/commons/3/3a/Cat03.jpg", {
"type": "input_text",
"text": "What animal is shown in this image? Answer in one word",
},
{"type": "input_image", "image_url": CAT_IMAGE_URL, "detail": "auto"},
],
}
]
response = client.responses.create(model=model, input=vision_input, instructions=INSTRUCTIONS)
text = response.output_text.strip().lower()
assert text, f"/responses vision returned no output text: {response.output!r}"
assert any(keyword in text for keyword in ("cat", "feline")), (
f"vision response did not describe the image: {text[:300]}"
) )
require_successful_call(result)
parsed = ResponsesResult.model_validate_json(result.body)
text = parsed.text.strip().lower()
assert text, f"/responses vision returned no output text: {result.body[:300]}"
assert any(
keyword in text
for keyword in ("cat", "feline")
), f"vision response did not describe the image: {parsed.text[:300]}"
@pytest.mark.covers("llm.responses.anthropic.basic.nonstream.works") @pytest.mark.covers("llm.responses.anthropic.basic.nonstream.works")
def test_responses_anthropic_returns_completion( def test_responses_anthropic_returns_completion(
self, endpoints_client: EndpointsClient, resources: ResourceManager self, proxy: ProxyClient, resources: ResourceManager, sdk: SdkClients
) -> None: ) -> None:
model = f"e2e-responses-{unique_marker()}" model = _register(proxy, resources, _anthropic_params())
model_id = endpoints_client.create_model( client = sdk.openai(resources.key())
model,
LiteLLMParamsBody(
model="anthropic/claude-haiku-4-5", api_key="os.environ/ANTHROPIC_API_KEY"
),
)
resources.defer(lambda: endpoints_client.delete_model(model_id))
key = resources.key()
result = endpoints_client.responses(key, model, "reply with one word") response = client.responses.create(
require_successful_call(result) model=model, input="reply with one word", instructions=INSTRUCTIONS
parsed = ResponsesResult.model_validate_json(result.body) )
assert parsed.text.strip(), f"/responses returned no output text: {result.body[:300]}" assert response.output_text.strip(), f"/responses returned no output text: {response.output!r}"
@pytest.mark.covers("llm.responses.anthropic.tool_use.nonstream.works") @pytest.mark.covers("llm.responses.anthropic.tool_use.nonstream.works")
def test_responses_anthropic_returns_function_call( def test_responses_anthropic_returns_function_call(
self, endpoints_client: EndpointsClient, resources: ResourceManager self, proxy: ProxyClient, resources: ResourceManager, sdk: SdkClients
) -> None: ) -> None:
model = f"e2e-responses-{unique_marker()}" model = _register(proxy, resources, _anthropic_params())
model_id = endpoints_client.create_model( client = sdk.openai(resources.key())
model,
LiteLLMParamsBody(
model="anthropic/claude-haiku-4-5", api_key="os.environ/ANTHROPIC_API_KEY"
),
)
resources.defer(lambda: endpoints_client.delete_model(model_id))
key = resources.key()
result = endpoints_client.responses_with_tools( response = client.responses.create(
key, model=model,
model, input="What is the weather in San Francisco? Use the get_weather tool.",
"What is the weather in San Francisco? Use the get_weather tool.", instructions=INSTRUCTIONS,
[ tools=[WEATHER_TOOL],
ResponsesFunctionTool(
name="get_weather",
description="Get the weather for a location",
parameters=FunctionParameters(
properties={"location": FunctionParameterProperty(type="string")},
required=["location"],
),
)
],
) )
require_successful_call(result) _assert_weather_call(response)
parsed = ResponsesResult.model_validate_json(result.body)
function_call = next(
(call for call in parsed.function_calls if call.name == "get_weather"),
None,
)
assert function_call is not None, f"no get_weather function call: {result.body[:500]}"
assert function_call.arguments is not None
raw_arguments = cast(object, json.loads(function_call.arguments))
arguments = WeatherArguments.model_validate(raw_arguments)
assert arguments.location, f"function call arguments missing location: {function_call.arguments}"
@pytest.mark.covers("llm.responses.bedrock_converse.basic.nonstream.works") @pytest.mark.covers("llm.responses.bedrock_converse.basic.nonstream.works")
def test_responses_bedrock_returns_completion( def test_responses_bedrock_returns_completion(
self, endpoints_client: EndpointsClient, resources: ResourceManager self, proxy: ProxyClient, resources: ResourceManager, sdk: SdkClients
) -> None: ) -> None:
require_env("AWS_ACCESS_KEY_ID", "AWS_SECRET_ACCESS_KEY", "AWS_REGION") require_env("AWS_ACCESS_KEY_ID", "AWS_SECRET_ACCESS_KEY", "AWS_REGION")
model = f"e2e-responses-{unique_marker()}" model = _register(proxy, resources, _bedrock_params())
model_id = endpoints_client.create_model(model, _bedrock_params()) client = sdk.openai(resources.key())
resources.defer(lambda: endpoints_client.delete_model(model_id))
key = resources.key()
result = endpoints_client.responses(key, model, "reply with one word") response = client.responses.create(
require_successful_call(result) model=model, input="reply with one word", instructions=INSTRUCTIONS
parsed = ResponsesResult.model_validate_json(result.body) )
assert parsed.text.strip(), f"/responses over bedrock returned no output text: {result.body[:300]}" assert response.output_text.strip(), (
f"/responses over bedrock returned no output text: {response.output!r}"
)
@pytest.mark.covers("llm.responses.bedrock_converse.tool_use.nonstream.works") @pytest.mark.covers("llm.responses.bedrock_converse.tool_use.nonstream.works")
def test_responses_bedrock_returns_function_call( def test_responses_bedrock_returns_function_call(
self, endpoints_client: EndpointsClient, resources: ResourceManager self, proxy: ProxyClient, resources: ResourceManager, sdk: SdkClients
) -> None: ) -> None:
require_env("AWS_ACCESS_KEY_ID", "AWS_SECRET_ACCESS_KEY", "AWS_REGION") require_env("AWS_ACCESS_KEY_ID", "AWS_SECRET_ACCESS_KEY", "AWS_REGION")
model = f"e2e-responses-{unique_marker()}" model = _register(proxy, resources, _bedrock_params())
model_id = endpoints_client.create_model(model, _bedrock_params()) client = sdk.openai(resources.key())
resources.defer(lambda: endpoints_client.delete_model(model_id))
key = resources.key()
result = endpoints_client.responses_with_tools( response = client.responses.create(
key, model, "What is the weather in San Francisco? Use the get_weather tool.", [WEATHER_TOOL] model=model,
input="What is the weather in San Francisco? Use the get_weather tool.",
instructions=INSTRUCTIONS,
tools=[WEATHER_TOOL],
) )
require_successful_call(result) _assert_weather_call(response)
parsed = ResponsesResult.model_validate_json(result.body)
function_call = next((call for call in parsed.function_calls if call.name == "get_weather"), None)
assert function_call is not None, f"no get_weather function call over bedrock: {result.body[:500]}"
assert function_call.arguments is not None
raw_arguments = cast(object, json.loads(function_call.arguments))
arguments = WeatherArguments.model_validate(raw_arguments)
assert arguments.location, f"function call arguments missing location: {function_call.arguments}"
def _parse_stream_event(
event: str,
) -> ResponsesOutputTextDeltaEvent | None:
try:
return ResponsesOutputTextDeltaEvent.model_validate_json(event)
except ValidationError:
return None

69
tests/e2e/llm_translation/test_responses_metadata_e2e.py
View file

@ -1,8 +1,8 @@
"""Live e2e: /v1/responses with store + metadata (LIT-1201 customer path). """Live e2e: /v1/responses with store + metadata (LIT-1201 customer path).
Customers attach metadata and store=true, then continue with previous_response_id. Customers attach metadata and store=true through the OpenAI SDK, then continue
Both turns must succeed, and any Redis keys written for the session must carry a with previous_response_id. Both turns must succeed, and any Redis keys written
positive TTL (not unbounded). for the session must carry a positive TTL (not unbounded).
""" """
from __future__ import annotations from __future__ import annotations
@ -15,21 +15,14 @@ import pytest
from pydantic import BaseModel, ConfigDict from pydantic import BaseModel, ConfigDict
from e2e_config import require_env, unique_marker from e2e_config import require_env, unique_marker
from e2e_http import require_successful_call
from endpoints_client import EndpointsClient, ResponsesResult
from lifecycle import ResourceManager from lifecycle import ResourceManager
from models import LiteLLMParamsBody from models import LiteLLMParamsBody
from proxy_client import ProxyClient
from sdk_clients import SdkClients
pytestmark = pytest.mark.e2e pytestmark = pytest.mark.e2e
INSTRUCTIONS = "You are a helpful assistant."
class ResponsesMetadataBody(BaseModel):
model: str
input: str
store: bool = True
metadata: dict[str, str]
previous_response_id: str | None = None
instructions: str | None = "You are a helpful assistant."
class RedisKeyInfo(BaseModel): class RedisKeyInfo(BaseModel):
@ -67,50 +60,42 @@ class TestResponsesMetadata:
exercised_on=["responses"], exercised_on=["responses"],
) )
def test_store_metadata_continues_and_redis_keys_have_ttl( def test_store_metadata_continues_and_redis_keys_have_ttl(
self, endpoints_client: EndpointsClient, resources: ResourceManager self, proxy: ProxyClient, resources: ResourceManager, sdk: SdkClients
) -> None: ) -> None:
# Anthropic avoids OpenAI/Gemini quota flakes; Responses translation still # Anthropic avoids OpenAI/Gemini quota flakes; Responses translation still
# exercises store + metadata + previous_response_id on the proxy. # exercises store + metadata + previous_response_id on the proxy.
marker = unique_marker() marker = unique_marker()
model = f"e2e-resp-meta-{marker}" model = f"e2e-resp-meta-{marker}"
model_id = endpoints_client.create_model( model_id = proxy.create_model(
model, model,
LiteLLMParamsBody( LiteLLMParamsBody(
model="anthropic/claude-haiku-4-5-20251001", model="anthropic/claude-haiku-4-5-20251001",
api_key="os.environ/ANTHROPIC_API_KEY", api_key="os.environ/ANTHROPIC_API_KEY",
), ),
) )
resources.defer(lambda: endpoints_client.delete_model(model_id)) resources.defer(lambda: proxy.delete_model(model_id))
key = resources.key() client = sdk.openai(resources.key())
first = endpoints_client.proxy.transport.send( first = client.responses.create(
"/v1/responses", model=model,
headers=endpoints_client.proxy.transport.bearer(key), input=f"Remember marker {marker}. Reply with one word.",
json=ResponsesMetadataBody( store=True,
model=model, metadata={"session_id": marker, "customer": "e2e"},
input=f"Remember marker {marker}. Reply with one word.", instructions=INSTRUCTIONS,
metadata={"session_id": marker, "customer": "e2e"},
),
) )
require_successful_call(first) assert first.id, f"responses must return an id: {first!r}"
parsed = ResponsesResult.model_validate_json(first.body) assert first.output_text.strip(), f"responses returned empty text: {first.output!r}"
assert parsed.id, f"responses must return an id: {first.body[:300]}"
assert parsed.text.strip(), f"responses returned empty text: {first.body[:300]}"
second = endpoints_client.proxy.transport.send( second = client.responses.create(
"/v1/responses", model=model,
headers=endpoints_client.proxy.transport.bearer(key), input="Reply with the single word ok.",
json=ResponsesMetadataBody( store=True,
model=model, previous_response_id=first.id,
input="Reply with the single word ok.", metadata={"session_id": marker, "turn": "2"},
previous_response_id=parsed.id, instructions=INSTRUCTIONS,
metadata={"session_id": marker, "turn": "2"},
),
) )
require_successful_call(second) assert second.output_text.strip(), (
second_parsed = ResponsesResult.model_validate_json(second.body) f"previous_response_id follow-up returned empty text: {second.output!r}"
assert second_parsed.text.strip(), (
f"previous_response_id follow-up returned empty text: {second.body[:300]}"
) )
time.sleep(1.0) time.sleep(1.0)

19
tests/e2e/models.py
View file

@ -364,6 +364,25 @@ class EmbedResponse(BaseModel):
model: str | None = None model: str | None = None
# ---------- rerank ----------
class RerankBody(BaseModel):
model: str
query: str
documents: list[str]
top_n: int
class RerankItem(BaseModel):
index: int | None = None
relevance_score: float | None = None
class RerankResponse(BaseModel):
results: list[RerankItem] = []
# ---------- ocr ---------- # ---------- ocr ----------

12
tests/e2e/proxy_client.py
View file

@ -57,6 +57,8 @@ from models import (
ModelUpdateBody, ModelUpdateBody,
OcrBody, OcrBody,
OcrResponse, OcrResponse,
RerankBody,
RerankResponse,
SpendLogRow, SpendLogRow,
SpendLogs, SpendLogs,
SpendLogsPage, SpendLogsPage,
@ -291,6 +293,16 @@ class ProxyClient:
response_type=OcrResponse, response_type=OcrResponse,
) )
def rerank(self, key: str, body: RerankBody) -> Result[RerankResponse]:
"""POST /v1/rerank (Cohere-format). No official OpenAI/Anthropic SDK
covers this route, so it stays on the shared typed transport."""
return self.transport.post(
"/v1/rerank",
headers=self.transport.bearer(key),
json=body,
response_type=RerankResponse,
)
def count_tokens(self, key: str, body: CountTokensBody) -> Result[CountTokensResponse]: def count_tokens(self, key: str, body: CountTokensBody) -> Result[CountTokensResponse]:
"""POST /v1/messages/count_tokens (Anthropic-native). Sends the """POST /v1/messages/count_tokens (Anthropic-native). Sends the
anthropic-version header so the native path accepts it; harmless on the anthropic-version header so the native path accepts it; harmless on the

4
uv.lock generated
View file

@ -10,7 +10,7 @@ resolution-markers = [
] ]
[options] [options]
exclude-newer = "2026-07-19T00:00:06.091071Z" exclude-newer = "2026-07-20T03:20:37.107777782Z"
exclude-newer-span = "P3D" exclude-newer-span = "P3D"
[manifest] [manifest]
@ -4300,6 +4300,7 @@ dev = [
{ name = "vcrpy" }, { name = "vcrpy" },
] ]
e2e-dev = [ e2e-dev = [
{ name = "anthropic" },
{ name = "locust" }, { name = "locust" },
{ name = "playwright" }, { name = "playwright" },
{ name = "websockets" }, { name = "websockets" },
@ -4477,6 +4478,7 @@ dev = [
{ name = "vcrpy", specifier = "==8.2.1" }, { name = "vcrpy", specifier = "==8.2.1" },
] ]
e2e-dev = [ e2e-dev = [
{ name = "anthropic", specifier = "==0.84.0" },
{ name = "locust", specifier = "==2.45.0" }, { name = "locust", specifier = "==2.45.0" },
{ name = "playwright", specifier = "==1.61.0" }, { name = "playwright", specifier = "==1.61.0" },
{ name = "websockets", specifier = ">=15.0.1,<16.0" }, { name = "websockets", specifier = ">=15.0.1,<16.0" },