mirror of
https://github.com/BerriAI/litellm.git
synced 2026-09-09 22:31:41 +00:00
ec8088f064
* test(e2e): cover vendor strategy gaps for chat contract, image edits, auth, team activity Resolves the first slice of LIT-4778 (vendor API testing strategy): image edits happy path, chat multi-turn + validation + sanitization, LLM-route auth header matrix, and /team/daily/activity structure * test(e2e): expand vendor API strategy coverage across endpoints 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 * test(e2e): finish vendor strategy open items Audio transcription negatives, vector-store file attach/poll/search, OpenAI moderation category matrix across chat/messages/responses, and smoke model matrix for chat (LIT-4778) * test(e2e): harden vendor strategy suite against live env edges Fix stream [DONE] tracking, XSS no-crash contract, realtime model routing, vector store list/search models, responses validation, and provider-denied Bedrock paths so the suite is stable against a live proxy * test(e2e): rename suites, drop vendor_contract, fix greptile gaps Move shared status helpers into e2e_http, rename chat auth headers and chat security suites, remove vendor_contract and dev_config files_settings, and tighten transcription validation plus vector-store search assertions * test(e2e): route bedrock stream disconnects through e2e_http Catch mid-stream RequestException in the shared harness so bedrock native tests do not import requests directly
101 lines
3.5 KiB
Python
101 lines
3.5 KiB
Python
"""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 e2e_config import unique_marker
|
|
from e2e_http import NoBody, assert_auth_denied, unwrap
|
|
from lifecycle import ResourceManager
|
|
from models import LiteLLMParamsBody
|
|
from proxy_client import ProxyClient
|
|
from pydantic import BaseModel
|
|
|
|
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 RealtimeClientSecretSession(BaseModel):
|
|
type: str | None = None
|
|
|
|
|
|
class RealtimeClientSecretResponse(BaseModel):
|
|
value: str | None = None
|
|
expires_at: int | None = None
|
|
session: RealtimeClientSecretSession | 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=REALTIME_BACKEND,
|
|
instructions="You are a helpful assistant.",
|
|
output_modalities=["text"],
|
|
),
|
|
),
|
|
response_type=RealtimeClientSecretResponse,
|
|
)
|
|
)
|
|
assert secret.value, f"client secret value missing: {secret}"
|
|
if secret.session is not None:
|
|
assert secret.session.type in (None, "realtime"), f"unexpected session type: {secret.session.type}"
|
|
|
|
@pytest.mark.covers("other.auth.realtime.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("other.auth.realtime.missing_header_denied")
|
|
def test_calls_without_auth_is_denied(self, proxy: ProxyClient) -> None:
|
|
result = proxy.transport.send(
|
|
"/v1/realtime/calls",
|
|
headers=NoBody(),
|
|
json=NoBody(),
|
|
)
|
|
assert_auth_denied(result, "realtime calls missing auth")
|