scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-10-07 02:59:05 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 103

test(e2e): expand vendor API strategy coverage across endpoints

Browse source 
Adds validation cases on existing endpoint suites, plus vector stores, search,
bedrock native, realtime HTTP secrets/calls, responses retrieve, files/batches
contract, and chat stream SSE. Registers coverage cells for LIT-4778
This commit is contained in:
mubashir1osmani 2026-07-24 14:47:33 -07:00
parent 5aa40595a2
commit b1fb112002
20 changed files with 1543 additions and 30 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

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

@ -40,6 +40,7 @@
- {id: llm.chat_completions.azure_openai.tool_use.nonstream.works, module: llm, tier: P0, subject_endpoint: chat_completions, route: azure_openai, capability: tool_use, streaming: nonstream, assertions: [works], source: "model_prices json", rationale: "Azure OpenAI function_calling"}
- {id: llm.chat_completions.azure_foundry.basic.nonstream.works, module: llm, tier: P1, subject_endpoint: chat_completions, route: azure_foundry, capability: basic, streaming: nonstream, assertions: [works], source: "proxy_server.py:8455", rationale: "Azure Foundry (azure_ai); newer, smoke"}
- {id: llm.messages.anthropic.basic.nonstream.works, module: llm, tier: P0, subject_endpoint: messages, route: anthropic, capability: basic, streaming: nonstream, assertions: [works], source: "anthropic_endpoints/endpoints.py:64", rationale: "Core endpoint; Anthropic Messages native"}
- {id: llm.messages.anthropic.input_validation.nonstream.works, module: llm, tier: P1, subject_endpoint: messages, route: anthropic, capability: input_validation, streaming: nonstream, assertions: [works], source: "vendor strategy §9.10 / LIT-4778", rationale: "Messages missing messages/max_tokens/model rejected"}
- {id: llm.messages.anthropic.basic.stream.works, module: llm, tier: P0, subject_endpoint: messages, route: anthropic, capability: basic, streaming: stream, assertions: [works], source: "anthropic_endpoints/endpoints.py:64", rationale: "Streaming Messages API"}
- {id: llm.messages.anthropic.basic.nonstream.cost_logged, module: llm, tier: P0, subject_endpoint: messages, route: anthropic, capability: basic, streaming: nonstream, assertions: [works, cost_logged], source: "anthropic_endpoints/endpoints.py:64", rationale: "Cost logged on passthrough"}
- {id: llm.messages.anthropic.tool_use.nonstream.works, module: llm, tier: P0, subject_endpoint: messages, route: anthropic, capability: tool_use, streaming: nonstream, assertions: [works], source: "model_prices json", rationale: "Tool calls via Messages API"}
@ -54,6 +55,7 @@
- {id: llm.messages.vertex.mid_conversation_system.nonstream.cache_hit, module: llm, tier: P0, subject_endpoint: messages, route: vertex, capability: mid_conversation_system, streaming: nonstream, assertions: [works, cache_hit], source: "llms/vertex_ai/vertex_ai_partner_models/anthropic/experimental_pass_through/transformation.py", rationale: "Vertex serves Claude on the native Anthropic contract, so flagged 4.8+/5 must keep mid-conversation system reminders in messages; hoisting mutates the system prefix and collapses the prompt cache (customer RCA gap)", fail_before_fix: proven}
- {id: llm.messages.vertex.mid_conversation_system.nonstream.works, module: llm, tier: P0, subject_endpoint: messages, route: vertex, capability: mid_conversation_system, streaming: nonstream, assertions: [works], source: "llms/vertex_ai/vertex_ai_partner_models/anthropic/experimental_pass_through/transformation.py", rationale: "Vertex Claude <= 4.7 rejects role system inside messages; unflagged models must hoist reminders into top-level system or every Claude Code session 400s (customer RCA gap)", fail_before_fix: proven}
- {id: llm.responses.openai.basic.nonstream.works, module: llm, tier: P0, subject_endpoint: responses, route: openai, capability: basic, streaming: nonstream, assertions: [works], source: "response_api_endpoints/endpoints.py:26", rationale: "Core endpoint; OpenAI Responses native"}
- {id: llm.responses.openai.input_validation.nonstream.works, module: llm, tier: P1, subject_endpoint: responses, route: openai, capability: input_validation, streaming: nonstream, assertions: [works], source: "vendor strategy §9.9 / LIT-4778", rationale: "Responses missing/empty input, missing model, invalid max_output_tokens"}
- {id: llm.responses.openai.basic.stream.works, module: llm, tier: P0, subject_endpoint: responses, route: openai, capability: basic, streaming: stream, assertions: [works], source: "response_api_endpoints/endpoints.py:26", rationale: "Streaming via /v1/responses"}
- {id: llm.responses.openai.basic.nonstream.cost_logged, module: llm, tier: P0, subject_endpoint: responses, route: openai, capability: basic, streaming: nonstream, assertions: [works, cost_logged], source: "response_api_endpoints/endpoints.py:26", rationale: "Cost logged on responses"}
- {id: llm.responses.openai.tool_use.nonstream.works, module: llm, tier: P0, subject_endpoint: responses, route: openai, capability: tool_use, streaming: nonstream, assertions: [works], source: "model_prices json", rationale: "Tool calls via Responses API"}

19
tests/e2e/coverage_registry/llm_nonconversational.yaml
View file

@ -1,5 +1,6 @@
# LLM non-conversational endpoints. Grounded in litellm/proxy endpoints + llms/ handlers.
- {id: llm.embeddings.openai.basic.nonstream.works, module: llm, tier: P0, subject_endpoint: embeddings, route: openai, capability: basic, streaming: nonstream, assertions: [works], source: "test_embeddings_endpoint_e2e.py:23", rationale: "Core endpoint, live vector response"}
- {id: llm.embeddings.openai.input_validation.nonstream.works, module: llm, tier: P1, subject_endpoint: embeddings, route: openai, capability: input_validation, streaming: nonstream, assertions: [works], source: "vendor strategy §9.3 / LIT-4778", rationale: "Missing model/input on /embeddings return client or known server errors"}
- {id: llm.embeddings.openai.basic.nonstream.cost_logged, module: llm, tier: P0, subject_endpoint: embeddings, route: openai, capability: basic, streaming: nonstream, assertions: [cost_logged], source: "SPEND_TRACKING_COVERAGE_MATRIX.md:34", rationale: "Cost tracking on embeddings"}
- {id: llm.embeddings.azure_openai.basic.nonstream.works, module: llm, tier: P0, subject_endpoint: embeddings, route: azure_openai, capability: basic, streaming: nonstream, assertions: [works], source: "llms/azure/azure.py", rationale: "Azure embeddings via translation"}
- {id: llm.embeddings.bedrock.basic.nonstream.works, module: llm, tier: P1, subject_endpoint: embeddings, route: bedrock_converse, capability: basic, streaming: nonstream, assertions: [works], source: "llms/bedrock/embed/embedding.py", rationale: "Bedrock Titan embeddings"}
@ -21,7 +22,9 @@
- {id: llm.batches.bedrock.assume_role.nonstream.works, module: llm, tier: P0, subject_endpoint: batches, route: bedrock_converse, capability: assume_role, streaming: nonstream, assertions: [works], source: "test_batches_e2e.py", rationale: "Bedrock batch create under STS assume-role credentials"}
- {id: llm.batches.hosted_vllm.basic.nonstream.works, module: llm, tier: P1, subject_endpoint: batches, route: hosted_vllm, capability: basic, streaming: nonstream, assertions: [works], source: "test_batches_e2e.py", rationale: "hosted_vllm OpenAI-compatible batch create"}
- {id: llm.batches.openai.key_model_access_denied.nonstream.works, module: llm, tier: P0, subject_endpoint: batches, route: openai, capability: basic, streaming: nonstream, assertions: [works], source: "test_batches_e2e.py", rationale: "Key model restriction 403 on upload/create"}
- {id: llm.batches.openai.input_validation.nonstream.works, module: llm, tier: P1, subject_endpoint: batches, route: openai, capability: input_validation, streaming: nonstream, assertions: [works], source: "vendor strategy §9.18 / LIT-4778", rationale: "Missing input_file_id and invalid batch id rejected"}
- {id: llm.files.openai.upload.nonstream.works, module: llm, tier: P0, subject_endpoint: files, route: openai, capability: basic, streaming: nonstream, assertions: [works], source: "openai_files_endpoints/files_endpoints.py:46", rationale: "File upload returns OpenAIFileObject"}
- {id: llm.files.openai.input_validation.nonstream.works, module: llm, tier: P1, subject_endpoint: files, route: openai, capability: input_validation, streaming: nonstream, assertions: [works], source: "vendor strategy §9.16 / LIT-4778", rationale: "File upload without purpose rejected"}
- {id: llm.files.openai.retrieve.nonstream.works, module: llm, tier: P0, subject_endpoint: files, route: openai, capability: basic, streaming: nonstream, assertions: [works], source: "files_endpoints.py", rationale: "File retrieve by id"}
- {id: llm.files.openai.delete.nonstream.works, module: llm, tier: P0, subject_endpoint: files, route: openai, capability: basic, streaming: nonstream, assertions: [works], source: "files_endpoints.py", rationale: "File delete returns deleted=true"}
- {id: llm.files.openai.list.nonstream.works, module: llm, tier: P0, subject_endpoint: files, route: openai, capability: basic, streaming: nonstream, assertions: [works], source: "files_endpoints.py", rationale: "File list paginated"}
@ -33,16 +36,31 @@
- {id: llm.rerank.cohere.basic.nonstream.works, module: llm, tier: P1, subject_endpoint: rerank, route: cohere, capability: basic, streaming: nonstream, assertions: [works], source: "test_rerank_e2e.py:29", rationale: "Cohere rerank, top_n + relevance_score"}
- {id: llm.files.openai.content.nonstream.works, module: llm, tier: P0, subject_endpoint: files, route: openai, capability: basic, streaming: nonstream, assertions: [works], source: "test_batches_e2e.py", rationale: "GET /v1/files/{id}/content returns uploaded batch JSONL bytes"}
- {id: llm.realtime.bedrock_converse.basic.stream.works, module: llm, tier: P0, subject_endpoint: realtime, route: bedrock_converse, capability: basic, streaming: stream, assertions: [works], source: "test_realtime_bedrock_e2e.py", rationale: "Nova Sonic realtime session emits response.done (LIT-2239)"}
- {id: llm.realtime.openai.basic.nonstream.works, module: llm, tier: P1, subject_endpoint: realtime, route: openai, capability: basic, streaming: nonstream, assertions: [works], source: "vendor strategy §9.19 / LIT-4778", rationale: "HTTP /v1/realtime/client_secrets and /calls reachable with auth"}
- {id: llm.vector_stores.openai.basic.nonstream.works, module: llm, tier: P1, subject_endpoint: vector_stores, route: openai, capability: basic, streaming: nonstream, assertions: [works], source: "vendor strategy §9.17 / LIT-4778", rationale: "Vector store create/list/retrieve/delete lifecycle"}
- {id: llm.vector_stores.openai.input_validation.nonstream.works, module: llm, tier: P1, subject_endpoint: vector_stores, route: openai, capability: input_validation, streaming: nonstream, assertions: [works], source: "vendor strategy §9.17 / LIT-4778", rationale: "Vector store search and invalid id errors"}
- {id: llm.search.openai.basic.nonstream.works, module: llm, tier: P1, subject_endpoint: search, route: openai, capability: basic, streaming: nonstream, assertions: [works], source: "vendor strategy §9.14 / LIT-4778", rationale: "POST /v1/search returns results for a registered tool"}
- {id: llm.search.openai.input_validation.nonstream.works, module: llm, tier: P1, subject_endpoint: search, route: openai, capability: input_validation, streaming: nonstream, assertions: [works], source: "vendor strategy §9.14 / LIT-4778", rationale: "Search missing/empty/invalid query rejected"}
- {id: llm.bedrock_native.bedrock_converse.basic.nonstream.works, module: llm, tier: P1, subject_endpoint: bedrock_native, route: bedrock_converse, capability: basic, streaming: nonstream, assertions: [works], source: "vendor strategy §9.12 / LIT-4778", rationale: "Bedrock native converse happy path"}
- {id: llm.bedrock_native.bedrock_converse.basic.stream.works, module: llm, tier: P1, subject_endpoint: bedrock_native, route: bedrock_converse, capability: basic, streaming: stream, assertions: [works], source: "vendor strategy §9.12 / LIT-4778", rationale: "Bedrock native converse-stream"}
- {id: llm.bedrock_native.bedrock_converse.input_validation.nonstream.works, module: llm, tier: P1, subject_endpoint: bedrock_native, route: bedrock_converse, capability: input_validation, streaming: nonstream, assertions: [works], source: "vendor strategy §9.12 / LIT-4778", rationale: "Bedrock converse missing/empty messages and invalid model"}
- {id: llm.bedrock_native.bedrock_invoke.basic.nonstream.works, module: llm, tier: P1, subject_endpoint: bedrock_native, route: bedrock_invoke, capability: basic, streaming: nonstream, assertions: [works], source: "vendor strategy §9.12 / LIT-4778", rationale: "Bedrock native invoke happy path"}
- {id: llm.bedrock_native.bedrock_invoke.basic.stream.works, module: llm, tier: P1, subject_endpoint: bedrock_native, route: bedrock_invoke, capability: basic, streaming: stream, assertions: [works], source: "vendor strategy §9.12 / LIT-4778", rationale: "Bedrock native invoke stream"}
- {id: llm.bedrock_native.bedrock_invoke.input_validation.nonstream.works, module: llm, tier: P1, subject_endpoint: bedrock_native, route: bedrock_invoke, capability: input_validation, streaming: nonstream, assertions: [works], source: "vendor strategy §9.12 / LIT-4778", rationale: "Bedrock invoke missing fields and invalid temperature"}
- {id: llm.ocr.openai.input_validation.nonstream.works, module: llm, tier: P1, subject_endpoint: ocr, route: openai, capability: input_validation, streaming: nonstream, assertions: [works], source: "vendor strategy §9.13 / LIT-4778", rationale: "OCR missing document rejected"}
- {id: llm.rerank.bedrock.basic.nonstream.works, module: llm, tier: P1, subject_endpoint: rerank, route: bedrock_converse, capability: basic, streaming: nonstream, assertions: [works], source: "llms/bedrock/rerank/handler.py", rationale: "Bedrock rerank"}
- {id: llm.rerank.together_ai.basic.nonstream.works, module: llm, tier: P1, subject_endpoint: rerank, route: together_ai, capability: basic, streaming: nonstream, assertions: [works], source: "llms/together_ai/rerank/handler.py", rationale: "Together rerank"}
- {id: llm.images_generations.openai.basic.nonstream.works, module: llm, tier: P1, subject_endpoint: images_generations, route: openai, capability: basic, streaming: nonstream, assertions: [works], source: "test_image_generation_e2e.py:22", rationale: "OpenAI image gen, b64/url"}
- {id: llm.images_edits.openai.basic.nonstream.works, module: llm, tier: P1, subject_endpoint: images_edits, route: openai, capability: basic, streaming: nonstream, assertions: [works], source: "test_image_edits_e2e.py", rationale: "OpenAI /v1/images/edits multipart image+prompt (vendor strategy / LIT-4778)"}
- {id: llm.images_edits.openai.input_validation.nonstream.works, module: llm, tier: P1, subject_endpoint: images_edits, route: openai, capability: input_validation, streaming: nonstream, assertions: [works], source: "vendor strategy §9.5 / LIT-4778", rationale: "Image edit empty prompt and empty image rejected"}
- {id: llm.images_generations.openai.input_validation.nonstream.works, module: llm, tier: P1, subject_endpoint: images_generations, route: openai, capability: input_validation, streaming: nonstream, assertions: [works], source: "vendor strategy §9.4 / LIT-4778", rationale: "Image gen missing/empty prompt and invalid size/n rejected"}
- {id: llm.images_generations.azure_openai.basic.nonstream.works, module: llm, tier: P1, subject_endpoint: images_generations, route: azure_openai, capability: basic, streaming: nonstream, assertions: [works], source: "llms/azure/azure.py", rationale: "Azure DALL-E"}
- {id: llm.images_generations.vertex.basic.nonstream.works, module: llm, tier: P1, subject_endpoint: images_generations, route: vertex, capability: basic, streaming: nonstream, assertions: [works], source: "vertex_ai/image_generation/image_generation_handler.py", rationale: "Vertex Imagen"}
- {id: llm.images_generations.bedrock.basic.nonstream.works, module: llm, tier: P1, subject_endpoint: images_generations, route: bedrock_converse, capability: basic, streaming: nonstream, assertions: [works], source: "bedrock/image_generation/image_handler.py", rationale: "Bedrock Titan Image"}
- {id: llm.images_generations.black_forest_labs.basic.nonstream.works, module: llm, tier: P1, subject_endpoint: images_generations, route: openai, capability: basic, streaming: nonstream, assertions: [works], source: "black_forest_labs/image_generation/handler.py", rationale: "BFL Flux via OpenAI-compat"}
- {id: llm.audio_speech.openai.basic.nonstream.works, module: llm, tier: P1, subject_endpoint: audio_speech, route: openai, capability: basic, streaming: nonstream, assertions: [works], source: "test_audio_speech_e2e.py:22", rationale: "OpenAI TTS binary audio"}
- {id: llm.audio_speech.openai.basic.stream.works, module: llm, tier: P1, subject_endpoint: audio_speech, route: openai, capability: basic, streaming: stream, assertions: [works], source: "proxy_server.py:9043", rationale: "TTS streaming chunk generator"}
- {id: llm.audio_speech.openai.input_validation.nonstream.works, module: llm, tier: P1, subject_endpoint: audio_speech, route: openai, capability: input_validation, streaming: nonstream, assertions: [works], source: "vendor strategy §9.6 / LIT-4778", rationale: "TTS missing input/model, invalid voice, empty input rejected"}
- {id: llm.audio_speech.azure_openai.basic.nonstream.works, module: llm, tier: P1, subject_endpoint: audio_speech, route: azure_openai, capability: basic, streaming: nonstream, assertions: [works], source: "llms/azure/azure.py", rationale: "Azure TTS"}
- {id: llm.audio_speech.vertex.basic.nonstream.works, module: llm, tier: P1, subject_endpoint: audio_speech, route: vertex, capability: basic, streaming: nonstream, assertions: [works], source: "vertex_ai/text_to_speech/text_to_speech_handler.py", rationale: "Vertex TTS"}
- {id: llm.audio_transcriptions.openai.basic.nonstream.works, module: llm, tier: P1, subject_endpoint: audio_transcriptions, route: openai, capability: basic, streaming: nonstream, assertions: [works], source: "openai/transcriptions/handler.py", rationale: "OpenAI Whisper"}
@ -50,3 +68,4 @@
- {id: llm.audio_transcriptions.soniox.basic.nonstream.works, module: llm, tier: P2, subject_endpoint: audio_transcriptions, route: openai, capability: basic, streaming: nonstream, assertions: [works], source: "soniox/audio_transcription/handler.py", rationale: "Soniox via OpenAI-compat (smoke)"}
- {id: llm.audio_transcriptions.nvidia_riva.basic.nonstream.works, module: llm, tier: P2, subject_endpoint: audio_transcriptions, route: openai, capability: basic, streaming: nonstream, assertions: [works], source: "nvidia_riva/audio_transcription/handler.py", rationale: "NVIDIA Riva (smoke)"}
- {id: llm.moderations.openai.basic.nonstream.works, module: llm, tier: P1, subject_endpoint: moderations, route: openai, capability: basic, streaming: nonstream, assertions: [works], source: "proxy_server.py", rationale: "OpenAI moderations (only provider)"}
- {id: llm.moderations.openai.input_validation.nonstream.works, module: llm, tier: P1, subject_endpoint: moderations, route: openai, capability: input_validation, streaming: nonstream, assertions: [works], source: "vendor strategy §9.8 / LIT-4778", rationale: "Moderations missing input rejected"}

4
tests/e2e/coverage_registry/schema.py
View file

@ -39,6 +39,10 @@ LlmEndpoint = Literal[
"audio_transcriptions",
"moderations",
"realtime",
"vector_stores",
"search",
"ocr",
"bedrock_native",
]
LlmRoute = Literal[

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

@ -22,6 +22,8 @@ __all__ = [
"CacheControl",
"RichMessage",
"TextBlock",
"ImageEditForm",
"ImagesResult",
]

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

@ -9,31 +9,42 @@ non-zero audio bytes.
from __future__ import annotations
import pytest
from pydantic import BaseModel
from e2e_config import unique_marker
from e2e_http import require_successful_call
from endpoints_client import EndpointsClient
from lifecycle import ResourceManager
from models import LiteLLMParamsBody
from vendor_contract import assert_error_or_server_known
pytestmark = pytest.mark.e2e
class _OptionalSpeechBody(BaseModel):
model: str | None = None
input: str | None = None
voice: str | None = None
def _register_tts(
endpoints_client: EndpointsClient, resources: ResourceManager
) -> tuple[str, str]:
model = f"e2e-speech-{unique_marker()}"
model_id = endpoints_client.create_model(
model,
LiteLLMParamsBody(model="openai/gpt-4o-mini-tts", api_key="os.environ/OPENAI_API_KEY"),
)
resources.defer(lambda: endpoints_client.delete_model(model_id))
return model, resources.key()
class TestAudioSpeech:
@pytest.mark.covers("llm.audio_speech.openai.basic.nonstream.works")
def test_audio_speech_returns_audio(
self, endpoints_client: EndpointsClient, resources: ResourceManager
) -> None:
model = f"e2e-speech-{unique_marker()}"
model_id = endpoints_client.create_model(
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()
model, key = _register_tts(endpoints_client, resources)
result = endpoints_client.audio_speech(key, model, "Hello!")
require_successful_call(result)
assert "audio" in (result.content_type or ""), (
@ -45,16 +56,7 @@ class TestAudioSpeech:
def test_audio_speech_streams_audio_chunks(
self, endpoints_client: EndpointsClient, resources: ResourceManager
) -> None:
model = f"e2e-speech-stream-{unique_marker()}"
model_id = endpoints_client.create_model(
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()
model, key = _register_tts(endpoints_client, resources)
result = endpoints_client.audio_speech_stream(
key,
model,
@ -76,3 +78,52 @@ class TestAudioSpeech:
f"streamed response (a buffered body is not a stream)"
)
assert result.total_bytes > 0, "/audio/speech stream returned no audio bytes"
@pytest.mark.covers("llm.audio_speech.openai.input_validation.nonstream.works")
def test_missing_input_returns_error(
self, endpoints_client: EndpointsClient, resources: ResourceManager
) -> None:
model, key = _register_tts(endpoints_client, resources)
result = endpoints_client.proxy.transport.send(
"/v1/audio/speech",
headers=endpoints_client.proxy.transport.bearer(key),
json=_OptionalSpeechBody(model=model, voice="alloy"),
)
assert_error_or_server_known(result, "speech missing input")
@pytest.mark.covers("llm.audio_speech.openai.input_validation.nonstream.works")
def test_missing_model_returns_error(
self, endpoints_client: EndpointsClient, resources: ResourceManager
) -> None:
_, key = _register_tts(endpoints_client, resources)
result = endpoints_client.proxy.transport.send(
"/v1/audio/speech",
headers=endpoints_client.proxy.transport.bearer(key),
json=_OptionalSpeechBody(input="hello", voice="alloy"),
)
assert_error_or_server_known(result, "speech missing model")
@pytest.mark.covers("llm.audio_speech.openai.input_validation.nonstream.works")
def test_invalid_voice_returns_error(
self, endpoints_client: EndpointsClient, resources: ResourceManager
) -> None:
model, key = _register_tts(endpoints_client, resources)
result = endpoints_client.proxy.transport.send(
"/v1/audio/speech",
headers=endpoints_client.proxy.transport.bearer(key),
json=_OptionalSpeechBody(model=model, input="hello", voice="invalid_voice_xyz"),
)
assert_error_or_server_known(result, "speech invalid voice")
@pytest.mark.covers("llm.audio_speech.openai.input_validation.nonstream.works")
def test_empty_input_returns_error(
self, endpoints_client: EndpointsClient, resources: ResourceManager
) -> None:
model, key = _register_tts(endpoints_client, resources)
result = endpoints_client.proxy.transport.send(
"/v1/audio/speech",
headers=endpoints_client.proxy.transport.bearer(key),
json=_OptionalSpeechBody(model=model, input="", voice="alloy"),
)
assert_error_or_server_known(result, "speech empty input")

225
tests/e2e/llm_translation/test_bedrock_native_e2e.py Normal file
View file

@ -0,0 +1,225 @@
"""Vendor §9.12: Bedrock native converse/invoke passthrough (LIT-4778).
Model is path-scoped. Happy paths assert assistant-shaped bodies; negatives pin
missing messages and invalid model handling without crashing the proxy.
"""
from __future__ import annotations
import pytest
from pydantic import BaseModel
from e2e_config import unique_marker
from e2e_http import require_successful_call
from lifecycle import ResourceManager
from models import LiteLLMParamsBody
from proxy_client import ProxyClient
from vendor_contract import assert_client_error, assert_error_or_server_known
pytestmark = pytest.mark.e2e
BEDROCK_BACKEND = "bedrock/us.anthropic.claude-haiku-4-5-20251001-v1:0"
class ConverseContent(BaseModel):
text: str
class ConverseMessage(BaseModel):
role: str
content: list[ConverseContent]
class ConverseInferenceConfig(BaseModel):
maxTokens: int = 50
temperature: float = 0.5
class ConverseBody(BaseModel):
messages: list[ConverseMessage] | None = None
system: list[ConverseContent] | None = None
inferenceConfig: ConverseInferenceConfig | None = None
class InvokeBody(BaseModel):
anthropic_version: str | None = None
messages: list[dict[str, str]] | None = None
max_tokens: int | None = None
temperature: float | None = None
system: str | None = None
def _register(proxy: ProxyClient, resources: ResourceManager) -> tuple[str, str]:
model = f"e2e-bedrock-native-{unique_marker()}"
model_id = proxy.create_model(
model,
LiteLLMParamsBody(
model=BEDROCK_BACKEND,
aws_access_key_id="os.environ/AWS_ACCESS_KEY_ID",
aws_secret_access_key="os.environ/AWS_SECRET_ACCESS_KEY",
aws_region_name="os.environ/AWS_REGION",
),
)
resources.defer(lambda: proxy.delete_model(model_id))
return model, resources.key()
def _default_converse() -> ConverseBody:
return ConverseBody(
messages=[ConverseMessage(role="user", content=[ConverseContent(text="Hello")])],
inferenceConfig=ConverseInferenceConfig(),
)
def _default_invoke() -> InvokeBody:
return InvokeBody(
anthropic_version="bedrock-2023-05-31",
messages=[{"role": "user", "content": "Hello"}],
max_tokens=50,
temperature=0.7,
)
class TestBedrockNative:
@pytest.mark.covers("llm.bedrock_native.bedrock_converse.basic.nonstream.works")
def test_converse_returns_assistant(
self, proxy: ProxyClient, resources: ResourceManager
) -> None:
model, key = _register(proxy, resources)
result = proxy.transport.send(
f"/bedrock/model/{model}/converse",
headers=proxy.transport.bearer(key),
json=_default_converse(),
)
require_successful_call(result)
assert result.body.strip(), f"converse returned empty body: {result.body[:300]}"
assert "assistant" in result.body or "output" in result.body or "message" in result.body, (
f"unexpected converse body: {result.body[:300]}"
)
@pytest.mark.covers("llm.bedrock_native.bedrock_converse.basic.stream.works")
def test_converse_stream_returns_chunks(
self, proxy: ProxyClient, resources: ResourceManager
) -> None:
model, key = _register(proxy, resources)
result = proxy.transport.send(
f"/bedrock/model/{model}/converse-stream",
headers=proxy.transport.bearer(key),
json=_default_converse(),
stream=True,
)
require_successful_call(result)
assert result.body or result.chunks > 0 or result.stream_events, (
"converse-stream returned no content"
)
@pytest.mark.covers("llm.bedrock_native.bedrock_invoke.basic.nonstream.works")
def test_invoke_returns_message(
self, proxy: ProxyClient, resources: ResourceManager
) -> None:
model, key = _register(proxy, resources)
result = proxy.transport.send(
f"/bedrock/model/{model}/invoke",
headers=proxy.transport.bearer(key),
json=_default_invoke(),
)
require_successful_call(result)
assert result.body.strip(), f"invoke returned empty body: {result.body[:300]}"
@pytest.mark.covers("llm.bedrock_native.bedrock_invoke.basic.stream.works")
def test_invoke_stream_returns_chunks(
self, proxy: ProxyClient, resources: ResourceManager
) -> None:
model, key = _register(proxy, resources)
result = proxy.transport.send(
f"/bedrock/model/{model}/invoke-with-response-stream",
headers=proxy.transport.bearer(key),
json=_default_invoke(),
stream=True,
)
require_successful_call(result)
assert result.body or result.chunks > 0 or result.stream_events, (
"invoke stream returned no content"
)
@pytest.mark.covers("llm.bedrock_native.bedrock_converse.input_validation.nonstream.works")
def test_converse_missing_messages_returns_error(
self, proxy: ProxyClient, resources: ResourceManager
) -> None:
model, key = _register(proxy, resources)
result = proxy.transport.send(
f"/bedrock/model/{model}/converse",
headers=proxy.transport.bearer(key),
json=ConverseBody(inferenceConfig=ConverseInferenceConfig()),
)
assert_error_or_server_known(result, "converse missing messages")
@pytest.mark.covers("llm.bedrock_native.bedrock_converse.input_validation.nonstream.works")
def test_converse_empty_messages_returns_client_error(
self, proxy: ProxyClient, resources: ResourceManager
) -> None:
model, key = _register(proxy, resources)
result = proxy.transport.send(
f"/bedrock/model/{model}/converse",
headers=proxy.transport.bearer(key),
json=ConverseBody(messages=[]),
)
assert_client_error(result, "converse empty messages")
@pytest.mark.covers("llm.bedrock_native.bedrock_converse.input_validation.nonstream.works")
def test_converse_invalid_model_returns_error(
self, proxy: ProxyClient, resources: ResourceManager
) -> None:
_, key = _register(proxy, resources)
result = proxy.transport.send(
"/bedrock/model/does-not-exist/converse",
headers=proxy.transport.bearer(key),
json=_default_converse(),
)
assert result.status_code in (400, 404), (
f"invalid model expected 400/404, got {result.status_code}: {result.body[:300]}"
)
@pytest.mark.covers("llm.bedrock_native.bedrock_invoke.input_validation.nonstream.works")
def test_invoke_missing_messages_returns_error(
self, proxy: ProxyClient, resources: ResourceManager
) -> None:
model, key = _register(proxy, resources)
result = proxy.transport.send(
f"/bedrock/model/{model}/invoke",
headers=proxy.transport.bearer(key),
json=InvokeBody(anthropic_version="bedrock-2023-05-31", max_tokens=50),
)
assert_error_or_server_known(result, "invoke missing messages")
@pytest.mark.covers("llm.bedrock_native.bedrock_invoke.input_validation.nonstream.works")
def test_invoke_missing_max_tokens_returns_error(
self, proxy: ProxyClient, resources: ResourceManager
) -> None:
model, key = _register(proxy, resources)
result = proxy.transport.send(
f"/bedrock/model/{model}/invoke",
headers=proxy.transport.bearer(key),
json=InvokeBody(
anthropic_version="bedrock-2023-05-31",
messages=[{"role": "user", "content": "Hello"}],
),
)
assert_error_or_server_known(result, "invoke missing max_tokens")
@pytest.mark.covers("llm.bedrock_native.bedrock_invoke.input_validation.nonstream.works")
def test_invoke_invalid_temperature_returns_client_error(
self, proxy: ProxyClient, resources: ResourceManager
) -> None:
model, key = _register(proxy, resources)
result = proxy.transport.send(
f"/bedrock/model/{model}/invoke",
headers=proxy.transport.bearer(key),
json=InvokeBody(
anthropic_version="bedrock-2023-05-31",
messages=[{"role": "user", "content": "Hello"}],
max_tokens=50,
temperature=5.0,
),
)
assert_client_error(result, "invoke invalid temperature")

60
tests/e2e/llm_translation/test_chat_stream_contract_e2e.py Normal file
View file

@ -0,0 +1,60 @@
"""Vendor §12.3: chat completions streaming SSE contract (LIT-4778).
Asserts a streamed /chat/completions response is SSE, carries content chunks,
and terminates with the OpenAI [DONE] sentinel.
"""
from __future__ import annotations
import pytest
from e2e_config import unique_marker
from e2e_http import require_successful_call
from lifecycle import ResourceManager
from models import ChatBody, ChatMessage, LiteLLMParamsBody
from proxy_client import ProxyClient
pytestmark = pytest.mark.e2e
class TestChatStreamContract:
@pytest.mark.covers("llm.chat_completions.openai.basic.stream.works")
def test_chat_stream_is_sse_and_ends_with_done(
self, proxy: ProxyClient, resources: ResourceManager
) -> None:
model = f"e2e-chat-stream-{unique_marker()}"
model_id = proxy.create_model(
model,
LiteLLMParamsBody(model="openai/gpt-4o-mini", api_key="os.environ/OPENAI_API_KEY"),
)
resources.defer(lambda: proxy.delete_model(model_id))
key = resources.key()
result = proxy.chat_stream(
key,
ChatBody(
model=model,
messages=[
ChatMessage(
role="user",
content=f"Reply with the single word ok. {unique_marker()}",
)
],
stream=True,
max_completion_tokens=32,
temperature=0.0,
),
)
require_successful_call(result)
assert result.is_streaming or "text/event-stream" in (result.content_type or ""), (
f"expected SSE content-type, got {result.content_type!r}"
)
assert result.stream_events or result.chunks > 0, "stream returned no events"
body = result.body
assert "data:" in body or result.stream_events, (
f"stream body missing data: lines: {body[:300]}"
)
joined = "\n".join(result.stream_events) if result.stream_events else body
assert "[DONE]" in joined or "data: [DONE]" in body, (
f"stream must terminate with [DONE], body={joined[:400]!r}"
)

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

@ -9,16 +9,23 @@ covered by tests/e2e/quota_management/spend_tracking/.
from __future__ import annotations
import pytest
from pydantic import BaseModel
from e2e_config import unique_marker
from e2e_http import require_successful_call
from endpoints_client import EmbeddingsResult, EndpointsClient
from lifecycle import ResourceManager
from models import LiteLLMParamsBody
from vendor_contract import assert_client_error, assert_error_or_server_known
pytestmark = pytest.mark.e2e
class _OptionalEmbeddingsBody(BaseModel):
model: str | None = None
input: str | list[str] | None = None
class TestEmbeddingsEndpoint:
@pytest.mark.covers("llm.embeddings.openai.basic.nonstream.works")
def test_embeddings_returns_vector(
@ -87,3 +94,57 @@ class TestEmbeddingsEndpoint:
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.openai.basic.nonstream.works")
def test_array_input_returns_vectors(
self, endpoints_client: EndpointsClient, resources: ResourceManager
) -> None:
model = f"e2e-embeddings-array-{unique_marker()}"
model_id = endpoints_client.create_model(
model,
LiteLLMParamsBody(
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.proxy.transport.send(
"/embeddings",
headers=endpoints_client.proxy.transport.bearer(key),
json=_OptionalEmbeddingsBody(model=model, input=["Hello", "World", "Test"]),
)
require_successful_call(result)
parsed = EmbeddingsResult.model_validate_json(result.body)
assert len(parsed.data) == 3, f"expected 3 vectors: {result.body[:300]}"
@pytest.mark.covers("llm.embeddings.openai.input_validation.nonstream.works")
def test_missing_model_returns_client_error(
self, endpoints_client: EndpointsClient, resources: ResourceManager
) -> None:
key = resources.key()
result = endpoints_client.proxy.transport.send(
"/embeddings",
headers=endpoints_client.proxy.transport.bearer(key),
json=_OptionalEmbeddingsBody(input="hello"),
)
assert_client_error(result, "embeddings missing model")
@pytest.mark.covers("llm.embeddings.openai.input_validation.nonstream.works")
def test_missing_input_returns_error(
self, endpoints_client: EndpointsClient, resources: ResourceManager
) -> None:
model = f"e2e-embeddings-missin-{unique_marker()}"
model_id = endpoints_client.create_model(
model,
LiteLLMParamsBody(
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.proxy.transport.send(
"/embeddings",
headers=endpoints_client.proxy.transport.bearer(key),
json=_OptionalEmbeddingsBody(model=model),
)
assert_error_or_server_known(result, "embeddings missing input")

106
tests/e2e/llm_translation/test_files_batches_contract_e2e.py Normal file
View file

@ -0,0 +1,106 @@
"""Vendor §9.16/9.18 contract negatives for files + batches (LIT-4778).
Happy-path file/batch lifecycle is covered under batches/; this pins upload
without purpose/file and invalid batch id retrieve.
"""
from __future__ import annotations
import pytest
from pydantic import BaseModel
from e2e_config import unique_marker
from e2e_http import NoBody, Success, UnknownApiError
from lifecycle import ResourceManager
from models import LiteLLMParamsBody
from proxy_client import ProxyClient
from vendor_contract import assert_error_or_server_known
pytestmark = pytest.mark.e2e
class BatchCreateBody(BaseModel):
input_file_id: str | None = None
endpoint: str = "/v1/chat/completions"
completion_window: str = "24h"
class BatchObject(BaseModel):
id: str
status: str | None = None
class TestFilesBatchesContract:
@pytest.mark.covers("llm.files.openai.input_validation.nonstream.works")
def test_upload_without_purpose_returns_error(
self, proxy: ProxyClient, resources: ResourceManager
) -> None:
model = f"e2e-files-contract-{unique_marker()}"
model_id = proxy.create_model(
model,
LiteLLMParamsBody(model="openai/gpt-4o-mini", api_key="os.environ/OPENAI_API_KEY"),
)
resources.defer(lambda: proxy.delete_model(model_id))
key = resources.key()
class EmptyForm(BaseModel):
pass
result = proxy.transport.upload(
"/v1/files",
headers=proxy.transport.bearer(key),
form=EmptyForm(),
filename="batch_input.jsonl",
content=b'{"custom_id":"1","method":"POST","url":"/v1/chat/completions","body":{}}\n',
response_type=NoBody,
)
match result:
case Success():
pytest.fail("upload without purpose must not succeed")
case UnknownApiError(status_code=status):
assert status in range(400, 600), f"unexpected {status}"
case _:
return
@pytest.mark.covers("llm.batches.openai.input_validation.nonstream.works")
def test_create_batch_missing_input_file_id_returns_error(
self, proxy: ProxyClient, resources: ResourceManager
) -> None:
model = f"e2e-batch-contract-{unique_marker()}"
model_id = proxy.create_model(
model,
LiteLLMParamsBody(model="openai/gpt-4o-mini", api_key="os.environ/OPENAI_API_KEY"),
)
resources.defer(lambda: proxy.delete_model(model_id))
key = resources.key()
result = proxy.transport.send(
"/v1/batches",
headers=proxy.transport.bearer(key),
json=BatchCreateBody(),
)
assert_error_or_server_known(result, "batch missing input_file_id")
@pytest.mark.covers("llm.batches.openai.input_validation.nonstream.works")
def test_retrieve_invalid_batch_id_returns_error(
self, proxy: ProxyClient, resources: ResourceManager
) -> None:
model = f"e2e-batch-contract-{unique_marker()}"
model_id = proxy.create_model(
model,
LiteLLMParamsBody(model="openai/gpt-4o-mini", api_key="os.environ/OPENAI_API_KEY"),
)
resources.defer(lambda: proxy.delete_model(model_id))
key = resources.key()
result = proxy.transport.get(
"/v1/batches/invalid-batch-id",
headers=proxy.transport.bearer(key),
params=NoBody(),
response_type=BatchObject,
)
match result:
case Success():
pytest.fail("invalid batch id must not succeed")
case UnknownApiError(status_code=status):
assert status in (400, 404, 500), f"unexpected {status}"
case _:
return

54
tests/e2e/llm_translation/test_image_edits_e2e.py
View file

@ -52,3 +52,57 @@ class TestImageEdit:
assert first.b64_json or first.url, (
f"edited image has neither b64_json nor url: {first}"
)
@pytest.mark.covers("llm.images_edits.openai.input_validation.nonstream.works")
def test_empty_prompt_returns_error(
self, endpoints_client: EndpointsClient, resources: ResourceManager
) -> None:
from e2e_http import Success, UnknownApiError
model = f"e2e-image-edit-empty-{unique_marker()}"
model_id = endpoints_client.create_model(
model,
LiteLLMParamsBody(model="openai/gpt-image-1", api_key="os.environ/OPENAI_API_KEY"),
)
resources.defer(lambda: endpoints_client.delete_model(model_id))
key = resources.key()
result = endpoints_client.image_edit(key, model, "", _TEST_PNG)
match result:
case Success():
pytest.fail("empty prompt on image edit must not succeed")
case UnknownApiError(status_code=status):
assert status in range(400, 600), f"unexpected {status}"
case _:
return
@pytest.mark.covers("llm.images_edits.openai.input_validation.nonstream.works")
def test_missing_image_returns_error(
self, endpoints_client: EndpointsClient, resources: ResourceManager
) -> None:
from e2e_http import Success, UnknownApiError
from endpoints_client import ImageEditForm, ImagesResult
model = f"e2e-image-edit-noimg-{unique_marker()}"
model_id = endpoints_client.create_model(
model,
LiteLLMParamsBody(model="openai/gpt-image-1", api_key="os.environ/OPENAI_API_KEY"),
)
resources.defer(lambda: endpoints_client.delete_model(model_id))
key = resources.key()
result = endpoints_client.proxy.transport.upload(
"/v1/images/edits",
headers=endpoints_client.proxy.transport.bearer(key),
form=ImageEditForm(model=model, prompt="add a red circle"),
filename="image.png",
content=b"",
file_content_type="image/png",
file_field="image",
response_type=ImagesResult,
)
match result:
case Success():
pytest.fail("empty image bytes must not succeed")
case UnknownApiError(status_code=status):
assert status in range(400, 600), f"unexpected {status}"
case _:
return

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

@ -8,16 +8,25 @@ litellm-regression-tests/tests/test_inference_endpoints.py.
from __future__ import annotations
import pytest
from pydantic import BaseModel
from e2e_config import unique_marker
from e2e_http import require_successful_call
from endpoints_client import EndpointsClient, ImagesResult
from lifecycle import ResourceManager
from models import LiteLLMParamsBody
from vendor_contract import assert_client_error, assert_error_or_server_known
pytestmark = pytest.mark.e2e
class _OptionalImageBody(BaseModel):
model: str | None = None
prompt: str | None = None
n: int | None = None
size: str | None = None
def _assert_image_returned(body: str) -> None:
parsed = ImagesResult.model_validate_json(body)
assert parsed.data, f"/images/generations returned no data: {body[:300]}"
@ -27,21 +36,24 @@ def _assert_image_returned(body: str) -> None:
)
def _register_openai_image(
endpoints_client: EndpointsClient, resources: ResourceManager
) -> tuple[str, str]:
model = f"e2e-image-{unique_marker()}"
model_id = endpoints_client.create_model(
model,
LiteLLMParamsBody(model="openai/gpt-image-1-mini", api_key="os.environ/OPENAI_API_KEY"),
)
resources.defer(lambda: endpoints_client.delete_model(model_id))
return model, resources.key()
class TestImageGeneration:
@pytest.mark.covers("llm.images_generations.openai.basic.nonstream.works")
def test_image_generation_returns_image(
self, endpoints_client: EndpointsClient, resources: ResourceManager
) -> None:
model = f"e2e-image-{unique_marker()}"
model_id = endpoints_client.create_model(
model,
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()
model, key = _register_openai_image(endpoints_client, resources)
result = endpoints_client.images(key, model, "Draw a cute cat")
require_successful_call(result)
_assert_image_returned(result.body)
@ -66,3 +78,52 @@ class TestImageGeneration:
result = endpoints_client.images(key, model, "Draw a cute cat")
require_successful_call(result)
_assert_image_returned(result.body)
@pytest.mark.covers("llm.images_generations.openai.input_validation.nonstream.works")
def test_missing_prompt_returns_error(
self, endpoints_client: EndpointsClient, resources: ResourceManager
) -> None:
model, key = _register_openai_image(endpoints_client, resources)
result = endpoints_client.proxy.transport.send(
"/v1/images/generations",
headers=endpoints_client.proxy.transport.bearer(key),
json=_OptionalImageBody(model=model),
)
assert_error_or_server_known(result, "images missing prompt")
@pytest.mark.covers("llm.images_generations.openai.input_validation.nonstream.works")
def test_empty_prompt_returns_client_error(
self, endpoints_client: EndpointsClient, resources: ResourceManager
) -> None:
model, key = _register_openai_image(endpoints_client, resources)
result = endpoints_client.proxy.transport.send(
"/v1/images/generations",
headers=endpoints_client.proxy.transport.bearer(key),
json=_OptionalImageBody(model=model, prompt=""),
)
assert_client_error(result, "images empty prompt")
@pytest.mark.covers("llm.images_generations.openai.input_validation.nonstream.works")
def test_invalid_size_returns_client_error(
self, endpoints_client: EndpointsClient, resources: ResourceManager
) -> None:
model, key = _register_openai_image(endpoints_client, resources)
result = endpoints_client.proxy.transport.send(
"/v1/images/generations",
headers=endpoints_client.proxy.transport.bearer(key),
json=_OptionalImageBody(model=model, prompt="a blue square", size="999x999"),
)
assert_client_error(result, "images invalid size")
@pytest.mark.covers("llm.images_generations.openai.input_validation.nonstream.works")
def test_invalid_n_returns_client_error(
self, endpoints_client: EndpointsClient, resources: ResourceManager
) -> None:
model, key = _register_openai_image(endpoints_client, resources)
result = endpoints_client.proxy.transport.send(
"/v1/images/generations",
headers=endpoints_client.proxy.transport.bearer(key),
json=_OptionalImageBody(model=model, prompt="a blue square", n=0),
)
assert_client_error(result, "images invalid n")

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

@ -9,6 +9,7 @@ litellm-regression-tests/tests/test_inference_endpoints.py.
from __future__ import annotations
import pytest
from pydantic import BaseModel
from e2e_config import unique_marker
from e2e_http import require_successful_call, unwrap
@ -23,9 +24,17 @@ from models import (
SpendLogRow,
ToolInputSchema,
)
from vendor_contract import assert_error_or_server_known
pytestmark = pytest.mark.e2e
class _OptionalMessagesBody(BaseModel):
model: str | None = None
messages: list[ChatMessage] | None = None
max_tokens: int | None = None
ANTHROPIC_BACKEND = "anthropic/claude-haiku-4-5"
WEATHER_TOOL = AnthropicCustomTool(
@ -169,3 +178,43 @@ class TestAnthropicMessages:
assert any(block.type == "tool_use" for block in response.content), (
f"model did not call the tool: {response}"
)
@pytest.mark.covers("llm.messages.anthropic.input_validation.nonstream.works")
def test_missing_messages_returns_error(
self, endpoints_client: EndpointsClient, resources: ResourceManager
) -> None:
model, key = self._register(endpoints_client, resources)
result = endpoints_client.proxy.transport.send(
"/v1/messages",
headers=endpoints_client.proxy.transport.bearer(key),
json=_OptionalMessagesBody(model=model, max_tokens=50),
)
assert_error_or_server_known(result, "messages missing messages")
@pytest.mark.covers("llm.messages.anthropic.input_validation.nonstream.works")
def test_missing_max_tokens_returns_error(
self, endpoints_client: EndpointsClient, resources: ResourceManager
) -> None:
model, key = self._register(endpoints_client, resources)
result = endpoints_client.proxy.transport.send(
"/v1/messages",
headers=endpoints_client.proxy.transport.bearer(key),
json=_OptionalMessagesBody(
model=model, messages=[ChatMessage(role="user", content="hi")]
),
)
assert_error_or_server_known(result, "messages missing max_tokens")
@pytest.mark.covers("llm.messages.anthropic.input_validation.nonstream.works")
def test_missing_model_returns_error(
self, endpoints_client: EndpointsClient, resources: ResourceManager
) -> None:
_, key = self._register(endpoints_client, resources)
result = endpoints_client.proxy.transport.send(
"/v1/messages",
headers=endpoints_client.proxy.transport.bearer(key),
json=_OptionalMessagesBody(
messages=[ChatMessage(role="user", content="hi")], max_tokens=50
),
)
assert_error_or_server_known(result, "messages missing model")

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

@ -8,12 +8,14 @@ with at least one policy category tripped, and benign text comes back not flagge
from __future__ import annotations
import pytest
from pydantic import BaseModel
from e2e_config import unique_marker
from e2e_http import unwrap
from endpoints_client import EndpointsClient
from lifecycle import ResourceManager
from models import LiteLLMParamsBody
from vendor_contract import assert_error_or_server_known
pytestmark = pytest.mark.e2e
@ -21,6 +23,11 @@ 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."
class _OptionalModerationBody(BaseModel):
model: str | None = None
input: str | None = None
def _register_moderation_model(
endpoints_client: EndpointsClient, resources: ResourceManager
) -> str:
@ -63,3 +70,16 @@ class TestModerations:
assert not item.flagged, (
f"benign text was flagged as {item.flagged_categories}: {item}"
)
@pytest.mark.covers("llm.moderations.openai.input_validation.nonstream.works")
def test_missing_input_returns_error(
self, endpoints_client: EndpointsClient, resources: ResourceManager
) -> None:
model = _register_moderation_model(endpoints_client, resources)
key = resources.key()
result = endpoints_client.proxy.transport.send(
"/v1/moderations",
headers=endpoints_client.proxy.transport.bearer(key),
json=_OptionalModerationBody(model=model),
)
assert_error_or_server_known(result, "moderations missing input")

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

@ -20,14 +20,23 @@ from typing import Protocol
import pytest
from pydantic import BaseModel
from e2e_config import unique_marker
from e2e_http import unwrap
from endpoints_client import EndpointsClient
from lifecycle import ResourceManager
from models import LiteLLMParamsBody, OcrBody, OcrDocument, OcrResponse
from vendor_contract import assert_error_or_server_known
pytestmark = pytest.mark.e2e
class _OptionalOcrBody(BaseModel):
model: str | None = None
document: dict[str, object] | None = None
# Tiny in-repo fixtures served via jsdelivr (sha-pinned, immutable) so the request
# bodies stay stable across runs.
TEST_PDF_URL = (
@ -153,4 +162,19 @@ class TestRustOcrGateway:
response = unwrap(endpoints_client.proxy.ocr(key, OcrBody(model=model, document=case.document)))
_assert_ocr_document(response)
@pytest.mark.covers("llm.ocr.openai.input_validation.nonstream.works")
def test_missing_document_returns_error(
self, endpoints_client: EndpointsClient, resources: ResourceManager
) -> None:
model = f"rust-ocr-val-{unique_marker()}"
model_id = endpoints_client.create_model(model, MistralOcr().litellm_params())
resources.defer(lambda: endpoints_client.delete_model(model_id))
key = resources.key()
result = endpoints_client.proxy.transport.send(
"/v1/ocr",
headers=endpoints_client.proxy.transport.bearer(key),
json=_OptionalOcrBody(model=model),
)
assert_error_or_server_known(result, "ocr missing document")

138
tests/e2e/llm_translation/test_realtime_http_e2e.py Normal file
View file

@ -0,0 +1,138 @@
"""Vendor §9.19: realtime client_secrets + calls HTTP surface (LIT-4778).
Websocket coverage already lives under realtime/; this file pins the HTTP
client-secret mint and the missing-auth contract.
"""
from __future__ import annotations
import pytest
from pydantic import BaseModel
from e2e_config import unique_marker
from e2e_http import NoBody, unwrap
from lifecycle import ResourceManager
from models import LiteLLMParamsBody
from proxy_client import ProxyClient
from vendor_contract import assert_auth_denied
pytestmark = pytest.mark.e2e
REALTIME_BACKEND = "openai/gpt-realtime"
class RealtimeSession(BaseModel):
type: str = "realtime"
model: str | None = None
instructions: str | None = None
output_modalities: list[str] | None = None
class RealtimeExpiresAfter(BaseModel):
anchor: str = "created_at"
seconds: int = 600
class RealtimeClientSecretRequest(BaseModel):
model: str
expires_after: RealtimeExpiresAfter | None = None
session: RealtimeSession | None = None
class RealtimeClientSecretResponse(BaseModel):
value: str | None = None
expires_at: int | None = None
session: dict[str, object] | None = None
def _register(proxy: ProxyClient, resources: ResourceManager) -> tuple[str, str]:
model = f"e2e-realtime-http-{unique_marker()}"
model_id = proxy.create_model(
model,
LiteLLMParamsBody(model=REALTIME_BACKEND, api_key="os.environ/OPENAI_API_KEY"),
)
resources.defer(lambda: proxy.delete_model(model_id))
return model, resources.key()
class TestRealtimeHttp:
@pytest.mark.covers("llm.realtime.openai.basic.nonstream.works")
def test_create_client_secret(
self, proxy: ProxyClient, resources: ResourceManager
) -> None:
model, key = _register(proxy, resources)
secret = unwrap(
proxy.transport.post(
"/v1/realtime/client_secrets",
headers=proxy.transport.bearer(key),
json=RealtimeClientSecretRequest(
model=model,
expires_after=RealtimeExpiresAfter(),
session=RealtimeSession(
model=model,
instructions="You are a helpful assistant.",
output_modalities=["text"],
),
),
response_type=RealtimeClientSecretResponse,
)
)
assert secret.value or secret.session, f"client secret empty: {secret}"
if secret.session is not None:
session_type = secret.session.get("type")
assert session_type in (None, "realtime"), f"unexpected session type: {session_type}"
@pytest.mark.covers("other.auth.llm_chat.missing_header_denied")
def test_client_secret_missing_auth_is_denied(
self, proxy: ProxyClient, resources: ResourceManager
) -> None:
model, _ = _register(proxy, resources)
result = proxy.transport.send(
"/v1/realtime/client_secrets",
headers=NoBody(),
json=RealtimeClientSecretRequest(model=model),
)
assert_auth_denied(result, "realtime client_secrets missing auth")
@pytest.mark.covers("llm.realtime.openai.basic.nonstream.works")
def test_calls_without_auth_is_denied(
self, proxy: ProxyClient, resources: ResourceManager
) -> None:
result = proxy.transport.send(
"/v1/realtime/calls",
headers=NoBody(),
json=NoBody(),
)
assert result.status_code in (401, 403, 405, 415, 422), (
f"realtime calls missing auth unexpected {result.status_code}: {result.body[:300]}"
)
@pytest.mark.covers("llm.realtime.openai.basic.nonstream.works")
def test_calls_authenticated_route_is_reachable(
self, proxy: ProxyClient, resources: ResourceManager
) -> None:
model, key = _register(proxy, resources)
secret = unwrap(
proxy.transport.post(
"/v1/realtime/client_secrets",
headers=proxy.transport.bearer(key),
json=RealtimeClientSecretRequest(
model=model,
session=RealtimeSession(model=model, output_modalities=["text"]),
),
response_type=RealtimeClientSecretResponse,
)
)
assert secret.value, f"need client secret value for calls: {secret}"
result = proxy.transport.send(
"/v1/realtime/calls",
headers=proxy.transport.bearer(secret.value),
json=NoBody(),
)
assert result.status_code not in (401, 403, 404), (
f"authenticated calls route must not be auth/not-found, "
f"got {result.status_code}: {result.body[:300]}"
)
assert result.status_code < 500, (
f"authenticated calls must not 5xx: {result.status_code} {result.body[:300]}"
)

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

@ -26,9 +26,17 @@ from endpoints_client import (
)
from lifecycle import ResourceManager
from models import LiteLLMParamsBody
from vendor_contract import assert_client_error, assert_error_or_server_known
pytestmark = pytest.mark.e2e
class _OptionalResponsesBody(BaseModel):
model: str | None = None
input: str | None = None
max_output_tokens: int | None = None
BEDROCK_CONVERSE_BACKEND = "bedrock/us.anthropic.claude-haiku-4-5-20251001-v1:0"
WEATHER_TOOL = ResponsesFunctionTool(
@ -286,6 +294,78 @@ class TestResponses:
arguments = WeatherArguments.model_validate(raw_arguments)
assert arguments.location, f"function call arguments missing location: {function_call.arguments}"
@pytest.mark.covers("llm.responses.openai.input_validation.nonstream.works")
def test_missing_input_returns_error(
self, endpoints_client: EndpointsClient, resources: ResourceManager
) -> None:
model = f"e2e-responses-val-{unique_marker()}"
model_id = endpoints_client.create_model(
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.proxy.transport.send(
"/v1/responses",
headers=endpoints_client.proxy.transport.bearer(key),
json=_OptionalResponsesBody(model=model),
)
assert_error_or_server_known(result, "responses missing input")
@pytest.mark.covers("llm.responses.openai.input_validation.nonstream.works")
def test_missing_model_returns_client_error(
self, endpoints_client: EndpointsClient, resources: ResourceManager
) -> None:
key = resources.key()
result = endpoints_client.proxy.transport.send(
"/v1/responses",
headers=endpoints_client.proxy.transport.bearer(key),
json=_OptionalResponsesBody(input="ping"),
)
assert_client_error(result, "responses missing model")
@pytest.mark.covers("llm.responses.openai.input_validation.nonstream.works")
def test_empty_input_returns_client_error(
self, endpoints_client: EndpointsClient, resources: ResourceManager
) -> None:
model = f"e2e-responses-val-{unique_marker()}"
model_id = endpoints_client.create_model(
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.proxy.transport.send(
"/v1/responses",
headers=endpoints_client.proxy.transport.bearer(key),
json=_OptionalResponsesBody(model=model, input=""),
)
assert_client_error(result, "responses empty input")
@pytest.mark.covers("llm.responses.openai.input_validation.nonstream.works")
@pytest.mark.parametrize("max_output_tokens", [-1, 0, -100])
def test_invalid_max_output_tokens_returns_client_error(
self,
endpoints_client: EndpointsClient,
resources: ResourceManager,
max_output_tokens: int,
) -> None:
model = f"e2e-responses-val-{unique_marker()}"
model_id = endpoints_client.create_model(
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.proxy.transport.send(
"/v1/responses",
headers=endpoints_client.proxy.transport.bearer(key),
json=_OptionalResponsesBody(
model=model, input="ping", max_output_tokens=max_output_tokens
),
)
assert_client_error(result, f"responses max_output_tokens={max_output_tokens}")
def _parse_stream_event(
event: str,
@ -294,3 +374,4 @@ def _parse_stream_event(
return ResponsesOutputTextDeltaEvent.model_validate_json(event)
except ValidationError:
return None

98
tests/e2e/llm_translation/test_responses_retrieve_e2e.py Normal file
View file

@ -0,0 +1,98 @@
"""Vendor §9.9: GET /v1/responses/{id} retrieve after store (LIT-4778).
Creates a stored response, retrieves it by id, and pins invalid-id error handling.
"""
from __future__ import annotations
import pytest
from pydantic import BaseModel
from e2e_config import unique_marker
from e2e_http import NoBody, Success, UnknownApiError, unwrap
from lifecycle import ResourceManager
from models import LiteLLMParamsBody
from proxy_client import ProxyClient
pytestmark = pytest.mark.e2e
class ResponsesCreateBody(BaseModel):
model: str
input: str
store: bool = True
stream: bool = False
max_output_tokens: int = 64
class ResponsesObject(BaseModel):
id: str
object: str | None = None
status: str | None = None
class TestResponsesRetrieve:
@pytest.mark.covers("llm.responses.openai.basic.nonstream.works")
def test_store_and_retrieve_by_id(
self, proxy: ProxyClient, resources: ResourceManager
) -> None:
model = f"e2e-resp-store-{unique_marker()}"
model_id = proxy.create_model(
model,
LiteLLMParamsBody(model="openai/gpt-4o-mini", api_key="os.environ/OPENAI_API_KEY"),
)
resources.defer(lambda: proxy.delete_model(model_id))
key = resources.key()
created = unwrap(
proxy.transport.post(
"/v1/responses",
headers=proxy.transport.bearer(key),
json=ResponsesCreateBody(
model=model,
input=f"Say pong. {unique_marker()}",
store=True,
),
response_type=ResponsesObject,
)
)
assert created.id, f"create returned no id: {created}"
assert created.object in (None, "response")
assert created.status in (None, "completed", "in_progress", "queued")
retrieved = unwrap(
proxy.transport.get(
f"/v1/responses/{created.id}",
headers=proxy.transport.bearer(key),
params=NoBody(),
response_type=ResponsesObject,
)
)
assert retrieved.id == created.id
@pytest.mark.covers("llm.responses.openai.input_validation.nonstream.works")
def test_invalid_response_id_returns_error(
self, proxy: ProxyClient, resources: ResourceManager
) -> None:
model = f"e2e-resp-badid-{unique_marker()}"
model_id = proxy.create_model(
model,
LiteLLMParamsBody(model="openai/gpt-4o-mini", api_key="os.environ/OPENAI_API_KEY"),
)
resources.defer(lambda: proxy.delete_model(model_id))
key = resources.key()
get_result = proxy.transport.get(
"/v1/responses/invalid-id",
headers=proxy.transport.bearer(key),
params=NoBody(),
response_type=ResponsesObject,
)
match get_result:
case Success():
pytest.fail("invalid response id must not succeed")
case UnknownApiError(status_code=status):
assert status in (400, 404, 500), (
f"invalid id expected 404/500-ish, got {status}"
)
case _:
return

177
tests/e2e/llm_translation/test_search_e2e.py Normal file
View file

@ -0,0 +1,177 @@
"""Vendor §9.14: POST /v1/search through a registered search tool (LIT-4778).
Registers a Perplexity-backed search tool at runtime, runs a basic search, and
pins missing/empty/invalid query handling.
"""
from __future__ import annotations
import os
import pytest
from pydantic import BaseModel
from e2e_config import unique_marker
from e2e_http import NoBody, unwrap
from lifecycle import ResourceManager
from proxy_client import ProxyClient
from vendor_contract import assert_client_error, assert_error_or_server_known
pytestmark = pytest.mark.e2e
class SearchToolLiteLLMParams(BaseModel):
search_provider: str
api_key: str
class SearchToolBody(BaseModel):
search_tool_name: str
litellm_params: SearchToolLiteLLMParams
search_tool_info: dict[str, str] | None = None
class CreateSearchToolRequest(BaseModel):
search_tool: SearchToolBody
class SearchToolResponse(BaseModel):
search_tool_id: str | None = None
search_tool_name: str | None = None
class SearchRequest(BaseModel):
search_tool_name: str | None = None
query: str | None = None
max_results: int | None = None
country: str | None = None
class SearchResultItem(BaseModel):
title: str | None = None
url: str | None = None
class SearchResponse(BaseModel):
object: str | None = None
results: list[SearchResultItem] = []
def _register_search_tool(proxy: ProxyClient, resources: ResourceManager) -> str:
api_key = os.environ.get("PERPLEXITY_API_KEY") or os.environ.get("TAVILY_API_KEY")
provider = "perplexity" if os.environ.get("PERPLEXITY_API_KEY") else "tavily"
if not api_key:
pytest.fail(
"set PERPLEXITY_API_KEY or TAVILY_API_KEY for /v1/search e2e coverage"
)
name = f"e2e-search-{unique_marker()}"
created = unwrap(
proxy.transport.post(
"/search_tools",
headers=proxy.transport.master,
json=CreateSearchToolRequest(
search_tool=SearchToolBody(
search_tool_name=name,
litellm_params=SearchToolLiteLLMParams(
search_provider=provider, api_key=api_key
),
search_tool_info={"description": "e2e search tool"},
)
),
response_type=SearchToolResponse,
)
)
tool_id = created.search_tool_id
if tool_id is not None:
def _delete_tool() -> None:
_ = proxy.transport.delete(
f"/search_tools/{tool_id}",
headers=proxy.transport.master,
json=NoBody(),
response_type=NoBody,
)
resources.defer(_delete_tool)
return name
class TestSearch:
@pytest.mark.covers("llm.search.openai.basic.nonstream.works")
def test_basic_search_returns_results(
self, proxy: ProxyClient, resources: ResourceManager
) -> None:
tool = _register_search_tool(proxy, resources)
key = resources.key()
result = unwrap(
proxy.transport.post(
"/v1/search",
headers=proxy.transport.bearer(key),
json=SearchRequest(
search_tool_name=tool,
query="latest AI news",
max_results=3,
country="US",
),
response_type=SearchResponse,
)
)
assert result.object in (None, "search")
assert isinstance(result.results, list), f"expected results array: {result}"
@pytest.mark.covers("llm.search.openai.basic.nonstream.works")
@pytest.mark.parametrize("max_results", [1, 5, 10])
def test_max_results_boundaries(
self, proxy: ProxyClient, resources: ResourceManager, max_results: int
) -> None:
tool = _register_search_tool(proxy, resources)
key = resources.key()
result = unwrap(
proxy.transport.post(
"/v1/search",
headers=proxy.transport.bearer(key),
json=SearchRequest(
search_tool_name=tool, query="weather forecast", max_results=max_results
),
response_type=SearchResponse,
)
)
assert isinstance(result.results, list)
@pytest.mark.covers("llm.search.openai.input_validation.nonstream.works")
def test_missing_query_returns_error(
self, proxy: ProxyClient, resources: ResourceManager
) -> None:
tool = _register_search_tool(proxy, resources)
key = resources.key()
result = proxy.transport.send(
"/v1/search",
headers=proxy.transport.bearer(key),
json=SearchRequest(search_tool_name=tool, max_results=3),
)
assert_error_or_server_known(result, "search missing query")
@pytest.mark.covers("llm.search.openai.input_validation.nonstream.works")
def test_empty_query_returns_client_error(
self, proxy: ProxyClient, resources: ResourceManager
) -> None:
tool = _register_search_tool(proxy, resources)
key = resources.key()
result = proxy.transport.send(
"/v1/search",
headers=proxy.transport.bearer(key),
json=SearchRequest(search_tool_name=tool, query=""),
)
assert_client_error(result, "search empty query")
@pytest.mark.covers("llm.search.openai.input_validation.nonstream.works")
def test_invalid_max_results_returns_client_error(
self, proxy: ProxyClient, resources: ResourceManager
) -> None:
tool = _register_search_tool(proxy, resources)
key = resources.key()
result = proxy.transport.send(
"/v1/search",
headers=proxy.transport.bearer(key),
json=SearchRequest(search_tool_name=tool, query="tech trends", max_results=-1),
)
assert_client_error(result, "search invalid max_results")

238
tests/e2e/llm_translation/test_vector_stores_e2e.py Normal file
View file

@ -0,0 +1,238 @@
"""Vendor §9.17: OpenAI vector store CRUD through the gateway (LIT-4778).
Create -> list -> retrieve -> delete against a live OpenAI-backed deployment.
Negatives pin missing search query and invalid store id handling.
"""
from __future__ import annotations
import pytest
from pydantic import BaseModel
from e2e_config import unique_marker
from e2e_http import NoBody, unwrap
from lifecycle import ResourceManager
from models import LiteLLMParamsBody
from proxy_client import ProxyClient
from vendor_contract import assert_error_or_server_known
pytestmark = pytest.mark.e2e
class VectorStoreCreateBody(BaseModel):
name: str
metadata: dict[str, str] | None = None
class VectorStoreObject(BaseModel):
id: str
object: str | None = None
name: str | None = None
metadata: dict[str, str] | None = None
class VectorStoreList(BaseModel):
object: str | None = None
data: list[VectorStoreObject] = []
class VectorStoreDeleteResponse(BaseModel):
id: str | None = None
object: str | None = None
deleted: bool | None = None
class VectorStoreSearchBody(BaseModel):
query: str | None = None
max_num_results: int | None = None
class VectorStoreUpdateBody(BaseModel):
name: str | None = None
def _register_openai_model(proxy: ProxyClient, resources: ResourceManager) -> str:
model = f"e2e-vs-{unique_marker()}"
model_id = proxy.create_model(
model,
LiteLLMParamsBody(model="openai/gpt-4o-mini", api_key="os.environ/OPENAI_API_KEY"),
)
resources.defer(lambda: proxy.delete_model(model_id))
return resources.key()
class TestVectorStores:
@pytest.mark.covers("llm.vector_stores.openai.basic.nonstream.works")
def test_create_list_retrieve_delete_lifecycle(
self, proxy: ProxyClient, resources: ResourceManager
) -> None:
key = _register_openai_model(proxy, resources)
name = f"e2e-vector-store-{unique_marker()}"
created = unwrap(
proxy.transport.post(
"/v1/vector_stores",
headers=proxy.transport.bearer(key),
json=VectorStoreCreateBody(
name=name, metadata={"project": "e2e", "env": "test"}
),
response_type=VectorStoreObject,
)
)
assert created.id, f"create returned no id: {created}"
store_id = created.id
def _delete_store() -> None:
_ = proxy.transport.delete(
f"/v1/vector_stores/{store_id}",
headers=proxy.transport.bearer(key),
json=NoBody(),
response_type=VectorStoreDeleteResponse,
)
resources.defer(_delete_store)
listed = unwrap(
proxy.transport.get(
"/v1/vector_stores",
headers=proxy.transport.bearer(key),
params=NoBody(),
response_type=VectorStoreList,
)
)
assert any(item.id == created.id for item in listed.data), (
f"created store {created.id} missing from list: {listed}"
)
retrieved = unwrap(
proxy.transport.get(
f"/v1/vector_stores/{created.id}",
headers=proxy.transport.bearer(key),
params=NoBody(),
response_type=VectorStoreObject,
)
)
assert retrieved.id == created.id
assert retrieved.object in (None, "vector_store")
deleted = unwrap(
proxy.transport.delete(
f"/v1/vector_stores/{created.id}",
headers=proxy.transport.bearer(key),
json=NoBody(),
response_type=VectorStoreDeleteResponse,
)
)
assert deleted.deleted is True or deleted.id == created.id
@pytest.mark.covers("llm.vector_stores.openai.input_validation.nonstream.works")
def test_search_missing_query_returns_error(
self, proxy: ProxyClient, resources: ResourceManager
) -> None:
key = _register_openai_model(proxy, resources)
created = unwrap(
proxy.transport.post(
"/v1/vector_stores",
headers=proxy.transport.bearer(key),
json=VectorStoreCreateBody(name=f"e2e-vs-search-{unique_marker()}"),
response_type=VectorStoreObject,
)
)
store_id = created.id
def _delete_search_store() -> None:
_ = proxy.transport.delete(
f"/v1/vector_stores/{store_id}",
headers=proxy.transport.bearer(key),
json=NoBody(),
response_type=VectorStoreDeleteResponse,
)
resources.defer(_delete_search_store)
result = proxy.transport.send(
f"/v1/vector_stores/{created.id}/search",
headers=proxy.transport.bearer(key),
json=VectorStoreSearchBody(max_num_results=10),
)
assert_error_or_server_known(result, "vector store search missing query")
@pytest.mark.covers("llm.vector_stores.openai.input_validation.nonstream.works")
def test_search_empty_query_returns_error_or_empty(
self, proxy: ProxyClient, resources: ResourceManager
) -> None:
key = _register_openai_model(proxy, resources)
created = unwrap(
proxy.transport.post(
"/v1/vector_stores",
headers=proxy.transport.bearer(key),
json=VectorStoreCreateBody(name=f"e2e-vs-empty-{unique_marker()}"),
response_type=VectorStoreObject,
)
)
store_id = created.id
def _delete_empty_store() -> None:
_ = proxy.transport.delete(
f"/v1/vector_stores/{store_id}",
headers=proxy.transport.bearer(key),
json=NoBody(),
response_type=VectorStoreDeleteResponse,
)
resources.defer(_delete_empty_store)
result = proxy.transport.send(
f"/v1/vector_stores/{created.id}/search",
headers=proxy.transport.bearer(key),
json=VectorStoreSearchBody(query="", max_num_results=10),
)
assert result.status_code in (200, 400, 500), (
f"empty search query unexpected status {result.status_code}: {result.body[:300]}"
)
@pytest.mark.covers("llm.vector_stores.openai.input_validation.nonstream.works")
def test_retrieve_invalid_id_returns_error(
self, proxy: ProxyClient, resources: ResourceManager
) -> None:
from e2e_http import Success, UnauthorizedError, UnknownApiError
key = _register_openai_model(proxy, resources)
result = proxy.transport.get(
"/v1/vector_stores/vs_does_not_exist_xyz",
headers=proxy.transport.bearer(key),
params=NoBody(),
response_type=VectorStoreObject,
)
match result:
case Success():
pytest.fail("invalid vector store id must not succeed")
case UnknownApiError(status_code=status):
assert status in (400, 401, 404, 500), f"unexpected status {status}"
case UnauthorizedError():
return
case _:
return
@pytest.mark.covers("llm.vector_stores.openai.input_validation.nonstream.works")
def test_invalid_chunking_returns_error(
self, proxy: ProxyClient, resources: ResourceManager
) -> None:
key = _register_openai_model(proxy, resources)
class ChunkingCreate(BaseModel):
name: str
chunking_strategy: dict[str, object]
result = proxy.transport.send(
"/v1/vector_stores",
headers=proxy.transport.bearer(key),
json=ChunkingCreate(
name=f"e2e-vs-chunk-{unique_marker()}",
chunking_strategy={
"type": "static",
"static": {
"max_chunk_size_tokens": 50,
"chunk_overlap_tokens": 40,
},
},
),
)
assert_error_or_server_known(result, "invalid chunking strategy")

43
tests/e2e/llm_translation/vendor_contract.py Normal file
View file

@ -0,0 +1,43 @@
"""Shared helpers for vendor API contract e2e tests (LIT-4778).
Status-centric assertions used across endpoint negatives, sanitization, and
auth matrix cases so each test file stays thin.
"""
from __future__ import annotations
from e2e_http import StreamingResponse
def is_client_error(status: int) -> bool:
return 400 <= status < 500
def is_auth_denied(status: int) -> bool:
return status in (401, 403)
def assert_not_server_error(result: StreamingResponse, context: str) -> None:
assert result.status_code not in (500, 502, 503), (
f"{context}: proxy must not 5xx, got {result.status_code}: {result.body[:300]}"
)
def assert_client_error(result: StreamingResponse, context: str) -> None:
assert is_client_error(result.status_code), (
f"{context}: expected 4xx, got {result.status_code}: {result.body[:300]}"
)
def assert_error_or_server_known(result: StreamingResponse, context: str) -> None:
"""Missing required fields may be 4xx or 5xx per known acceptable proxy behavior."""
assert result.status_code in range(400, 600), (
f"{context}: expected error status, got {result.status_code}: {result.body[:300]}"
)
assert result.status_code != 200
def assert_auth_denied(result: StreamingResponse, context: str) -> None:
assert is_auth_denied(result.status_code), (
f"{context}: expected 401/403, got {result.status_code}: {result.body[:300]}"
)