scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-11-15 09:12:18 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 311
litellm/tests/unit/realtime_api/test_main.py
quietjules 89149b4544
refactor(typing): clear 1141 basedpyright Any errors across 182 files (#50711) 
* refactor(typing): clear 340 basedpyright Any errors (wave 1)

* refactor(typing): clear 497 basedpyright Any errors (wave 2)

* refactor(typing): clear 74 basedpyright Any errors (wave 3)

* refactor(typing): clear 143 basedpyright Any errors (wave 4)

* refactor(typing): clear 106 basedpyright Any errors (wave 5)

* fix(typing): keep router fence visibility and py310-safe Self import

Bind the Protocol-narrowed routers in pass_through_endpoints and rag
ingestion to router-named locals so router_surface_check still sees the
aadapter_completion and aocr uses, and import Self from typing_extensions
in prisma_client for check_py310_typing_imports.

* perf(core_helpers): fast-path dicts and lists in the checked JSON helpers

as_str_mapping now tries the concrete dict check before the Mapping ABC
check, and iter_items returns lists before the runtime-checkable Protocol
check. Both return the same object as before, and roughly halve the
overhead CodSpeed flagged on the A2A text-extraction benchmarks.

* refactor(realtime): keep the base TypeError for a non-mapping translation expires_after

* fix(proxy): restore the require_attributes import the main merge dropped

* test: cover the retyped realtime, vector store, agent, stream, video and provider response paths

* fix(realtime): name the public request types in non-mapping TypeErrors

keyword_arguments wraps expires_after and transcription_session so a non-mapping value raises the same TypeError text as before the typing pass. Tests narrow Greptile's pegasus case with isinstance and freeze its model.

---------

Co-authored-by: mateo-berri <277851410+mateo-berri@users.noreply.github.com>
2026-11-11 22:11:47 +00:00

2637 lines
102 KiB
Python
Raw Permalink Blame History

import asyncio
import functools
import json
import time
from collections.abc import Awaitable, Callable, Mapping
from datetime import datetime, timezone
from pathlib import Path
from types import MappingProxyType, TracebackType
from typing import ClassVar, Final, Literal, Protocol
from unittest.mock import AsyncMock, MagicMock, patch
import httpx
import openai
import pytest
import respx
from openai import AsyncOpenAI
from websockets.exceptions import ConnectionClosed, ConnectionClosedError
from websockets.frames import Close
import litellm
from litellm.constants import REALTIME_SESSION_FAILURE_LOGGED_KEY
from litellm.integrations.custom_guardrail import CustomGuardrail
from litellm.integrations.custom_logger import CustomLogger
from litellm.litellm_core_utils.litellm_logging import Logging
from litellm.litellm_core_utils.realtime_streaming import TRANSLATION_GUARDRAILS_UNSUPPORTED_MESSAGE, RealTimeStreaming
from litellm.llms.base_llm.chat.transformation import BaseLLMException
from litellm.llms.custom_httpx.http_handler import AsyncHTTPHandler
from litellm.llms.openai.workload_identity import _workload_identity_auth
from litellm.models.credentials import CredentialItem
from litellm.proxy.common_utils.callback_utils import initialize_callbacks_on_proxy
from litellm.proxy.litellm_pre_call_utils import _apply_resolved_guardrails_to_metadata
from litellm.realtime_api import main as realtime_main
from litellm.realtime_api.main import with_resolved_session_model
from litellm.types.guardrails import GuardrailEventHooks
from litellm.types.proxy.policy_engine import Policy, PolicyCondition, PolicyMatchContext
from litellm.types.proxy.policy_engine.pipeline_types import GuardrailPipeline, PipelineStep
from litellm.types.realtime import RealtimeQueryParams
@pytest.fixture
def local_model_cost_map(monkeypatch):
monkeypatch.setenv("LITELLM_LOCAL_MODEL_COST_MAP", "True")
monkeypatch.setattr(litellm, "model_cost", litellm.get_model_cost_map(url=""))
litellm.get_model_info.cache_clear()
yield
litellm.get_model_info.cache_clear()
class FakeLogging:
def update_from_kwargs(self, **kwargs):
pass
def pre_call(self, **kwargs):
pass
class _CallCapture:
def __init__(self) -> None:
self.values: Mapping[str, object] | None = None
def test_resolves_top_level_session_model():
resolved = with_resolved_session_model({"model": "alias/gpt-realtime"}, "gpt-realtime")
assert resolved == {"model": "gpt-realtime"}
def test_session_without_model_is_returned_unchanged():
session = {"type": "realtime", "audio": {"input": {}}}
assert with_resolved_session_model(session, "gpt-realtime") == session
def test_does_not_clobber_flat_transcription_model():
"""The nested transcription model is a different model than the realtime
conversation model and must not be overwritten with the routing model."""
resolved = with_resolved_session_model(
{"model": "gpt-4o-realtime-preview", "input_audio_transcription": {"model": "whisper-1"}},
"gpt-4o-realtime-preview",
)
assert resolved["input_audio_transcription"]["model"] == "whisper-1"
def test_does_not_clobber_nested_audio_transcription_model():
resolved = with_resolved_session_model(
{
"model": "gpt-4o-realtime-preview",
"audio": {"input": {"transcription": {"model": "whisper-1"}}},
},
"gpt-4o-realtime-preview",
)
assert resolved["audio"]["input"]["transcription"]["model"] == "whisper-1"
def test_original_session_is_not_mutated():
session = {"model": "alias/gpt-realtime"}
with_resolved_session_model(session, "gpt-realtime")
assert session == {"model": "alias/gpt-realtime"}
def _run_client_secret(session, model, monkeypatch):
captured = {}
async def mock_handler(**kwargs):
captured.update(kwargs)
return object()
def mock_get_llm_provider(model, api_base, api_key, custom_llm_provider: str | None = None):
return model, "openai", None, api_base
monkeypatch.setattr(realtime_main, "get_llm_provider", mock_get_llm_provider)
monkeypatch.setattr(
realtime_main.base_llm_http_handler,
"async_realtime_client_secret_handler",
mock_handler,
)
asyncio.run(
realtime_main.acreate_realtime_client_secret.__wrapped__(
model=model,
session=session,
litellm_logging_obj=FakeLogging(),
)
)
return captured
def test_client_secret_session_model_takes_priority_over_top_level(monkeypatch):
"""Backwards-compatible ordering: an explicit session.model wins over the
top-level model, matching the proxy's own resolution order."""
captured = _run_client_secret(
session={"model": "gpt-realtime-session"},
model="gpt-realtime-top-level",
monkeypatch=monkeypatch,
)
assert captured["model"] == "gpt-realtime-session"
assert captured["request_data"]["session"]["model"] == "gpt-realtime-session"
async def _hanging_resolver(credentials, project_id, custom_llm_provider) -> tuple[str, str]:
await asyncio.sleep(30)
return "", ""
async def _thread_offloaded_hanging_resolver(credentials, project_id, custom_llm_provider) -> tuple[str, str]:
from litellm.litellm_core_utils.asyncify import asyncify
await asyncify(time.sleep)(30)
return "", ""
async def _instant_resolver(credentials, project_id, custom_llm_provider) -> tuple[str, str]:
return "token-abc", "resolved-project"
@pytest.mark.asyncio
async def test_vertex_credential_resolution_returns_the_resolved_token_and_project():
assert await realtime_main._resolve_vertex_access_token_bounded(
credentials="fake-credentials",
project_id="fake-project",
resolver=_instant_resolver,
timeout_seconds=5,
) == ("token-abc", "resolved-project")
@pytest.mark.asyncio
async def test_vertex_credential_resolution_times_out_instead_of_hanging():
"""Regression for the realtime accept-then-silence hang: a stalled Google
OAuth token refresh used to block the vertex branch unbounded (minutes of
zero frames for the client). It must raise promptly and name the timeout."""
start = time.monotonic()
with pytest.raises(ValueError, match="timed out fetching Google OAuth access token"):
await realtime_main._resolve_vertex_access_token_bounded(
credentials="fake-credentials",
project_id="fake-project",
resolver=_hanging_resolver,
timeout_seconds=0.05,
)
assert time.monotonic() - start < 5
@pytest.mark.asyncio
async def test_vertex_credential_resolution_bounds_a_thread_offloaded_refresh():
"""The real stall is a blocking google-auth refresh that runs in a worker
thread via asyncify, not a plain awaitable sleep. A timeout that only bounds
cancellable awaits would leave that shape hanging, so bound the shape the
proxy actually runs."""
start = time.monotonic()
with pytest.raises(ValueError, match="timed out fetching Google OAuth access token"):
await realtime_main._resolve_vertex_access_token_bounded(
credentials="fake-credentials",
project_id="fake-project",
resolver=_thread_offloaded_hanging_resolver,
timeout_seconds=0.05,
)
assert time.monotonic() - start < 5
@pytest.mark.asyncio
async def test_meta_realtime_dispatches_to_base_handler_with_meta_config(monkeypatch: pytest.MonkeyPatch):
from litellm.llms.meta.realtime.transformation import MetaRealtimeConfig
captured: dict[str, object] = {}
def mock_get_llm_provider(model, api_base, api_key, custom_llm_provider: str | None = None):
return model.removeprefix("meta/"), "meta", None, api_base
async def mock_async_realtime(**kwargs):
captured.update(kwargs)
monkeypatch.setattr(realtime_main, "get_llm_provider", mock_get_llm_provider)
monkeypatch.setattr(realtime_main.base_llm_http_handler, "async_realtime", mock_async_realtime)
await realtime_main._arealtime.__wrapped__(
model="meta/muse-voice-transcribe-1.0",
websocket=MagicMock(),
litellm_logging_obj=FakeLogging(),
query_params={"model": "meta/muse-voice-transcribe-1.0", "intent": "transcription"},
)
assert isinstance(captured["provider_config"], MetaRealtimeConfig)
assert captured["model"] == "muse-voice-transcribe-1.0"
assert captured["query_params"] == {"model": "muse-voice-transcribe-1.0", "intent": "transcription"}
@pytest.mark.asyncio
async def test_arealtime_vertex_branch_resolves_credentials_under_a_bound(monkeypatch):
"""The wiring half of the regression: the vertex branch of _arealtime must
go through the bounded resolver, so a hung token refresh surfaces as a
prompt error there rather than as an accepted-then-silent websocket."""
async def hanging_token_refresh(**kwargs):
await asyncio.sleep(30)
def mock_get_llm_provider(model, api_base, api_key, custom_llm_provider: str | None = None):
return model, "vertex_ai", None, api_base
monkeypatch.setattr(realtime_main, "get_llm_provider", mock_get_llm_provider)
monkeypatch.setattr(realtime_main, "vertex_access_token_resolver", hanging_token_refresh)
monkeypatch.setattr(realtime_main, "REALTIME_CREDENTIAL_RESOLUTION_TIMEOUT_SECONDS", 0.05)
start = time.monotonic()
with pytest.raises(ValueError, match="timed out fetching Google OAuth access token"):
await realtime_main._arealtime.__wrapped__(
model="gemini-live-2.5-flash",
websocket=MagicMock(),
litellm_logging_obj=FakeLogging(),
vertex_credentials="fake-credentials",
vertex_project="fake-project",
vertex_location="us-central1",
)
assert time.monotonic() - start < 5
def test_client_secret_forwards_nested_transcription_model_untouched(monkeypatch):
captured = _run_client_secret(
session={
"model": "gpt-4o-realtime-preview",
"input_audio_transcription": {"model": "whisper-1"},
},
model=None,
monkeypatch=monkeypatch,
)
session = captured["request_data"]["session"]
assert session["model"] == "gpt-4o-realtime-preview"
assert session["input_audio_transcription"]["model"] == "whisper-1"
@pytest.mark.asyncio
async def test_arealtime_live_session_dispatches_to_openai_live_handler(monkeypatch: pytest.MonkeyPatch) -> None:
captured: Final = _CallCapture()
def mock_get_llm_provider(
model: str,
api_base: str | None,
api_key: str | None,
custom_llm_provider: str | None = None,
) -> tuple[str, str, str | None, str | None]:
return "gpt-live-1", "openai", "resolved-key", "https://live-gateway.example/v1"
class FakeLiveHandler:
async def async_live_session(self, **kwargs: object) -> None:
captured.values = kwargs
monkeypatch.setattr(realtime_main, "get_llm_provider", mock_get_llm_provider)
monkeypatch.setattr(realtime_main, "openai_live_sessions", FakeLiveHandler())
start_frame: Final[dict[str, object]] = {
"type": "session.start",
"session": {"model": "live-alias", "instructions": "Stay brief"},
}
websocket: Final = MagicMock()
logging_obj: Final = FakeLogging()
await realtime_main._arealtime.__wrapped__(
model="live-alias",
websocket=websocket,
litellm_logging_obj=logging_obj,
live_session_start=start_frame,
)
assert captured.values is not None
assert captured.values["model"] == "gpt-live-1"
assert captured.values["websocket"] is websocket
assert captured.values["logging_obj"] is logging_obj
assert captured.values["session_start"] == start_frame
assert captured.values["api_base"] == "https://live-gateway.example/v1"
assert captured.values["api_key"] == "resolved-key"
@pytest.mark.asyncio
async def test_arealtime_live_session_rejects_matching_guardrail_before_dispatch(
monkeypatch: pytest.MonkeyPatch,
) -> None:
class RecordingGuardrail(CustomGuardrail):
def __init__(self) -> None:
super().__init__(
guardrail_name="live-blocker",
event_hook=GuardrailEventHooks.realtime_input_transcription,
default_on=True,
)
self.request_data: dict[str, object] | None = None
def should_run_guardrail(self, data: dict[str, object], event_type: GuardrailEventHooks) -> bool:
self.request_data = data
return super().should_run_guardrail(data, event_type)
class FakeLiveHandler:
def __init__(self) -> None:
self.called = False
async def async_live_session(self, **kwargs: object) -> None:
self.called = True
guardrail: Final = RecordingGuardrail()
live_handler: Final = FakeLiveHandler()
def mock_get_llm_provider(
model: str,
api_base: str | None,
api_key: str | None,
custom_llm_provider: str | None = None,
) -> tuple[str, str, str | None, str | None]:
return "gpt-live-1", "openai", "resolved-key", "https://live-gateway.example/v1"
monkeypatch.setattr(realtime_main, "get_llm_provider", mock_get_llm_provider)
monkeypatch.setattr(realtime_main, "openai_live_sessions", live_handler)
monkeypatch.setattr(litellm, "callbacks", [guardrail])
with pytest.raises(ValueError, match="Guardrails are not supported on OpenAI Live sessions"):
await realtime_main._arealtime.__wrapped__(
model="live-alias",
websocket=MagicMock(),
litellm_logging_obj=FakeLogging(),
litellm_metadata={"session_id": "live-session"},
live_session_start={"type": "session.start", "session": {"model": "live-alias"}},
)
assert guardrail.request_data == {"litellm_metadata": {"session_id": "live-session"}, "stream": True}
assert live_handler.called is False
@pytest.mark.asyncio
async def test_arealtime_live_session_rejects_streaming_only_pre_call_guardrail(
monkeypatch: pytest.MonkeyPatch,
) -> None:
class FakeLiveHandler:
def __init__(self) -> None:
self.called = False
async def async_live_session(self, **kwargs: object) -> None:
self.called = True
guardrail: Final = CustomGuardrail(
guardrail_name="streaming-only-blocker",
event_hook=GuardrailEventHooks.pre_call,
default_on=True,
stream_scope="streaming",
)
live_handler: Final = FakeLiveHandler()
def mock_get_llm_provider(
model: str,
api_base: str | None,
api_key: str | None,
custom_llm_provider: str | None = None,
) -> tuple[str, str, str | None, str | None]:
return "gpt-live-1", "openai", "resolved-key", "https://live-gateway.example/v1"
monkeypatch.setattr(realtime_main, "get_llm_provider", mock_get_llm_provider)
monkeypatch.setattr(realtime_main, "openai_live_sessions", live_handler)
monkeypatch.setattr(litellm, "callbacks", [guardrail])
with pytest.raises(ValueError, match="Guardrails are not supported on OpenAI Live sessions"):
await realtime_main._arealtime.__wrapped__(
model="live-alias",
websocket=MagicMock(),
litellm_logging_obj=FakeLogging(),
litellm_metadata={"session_id": "live-session"},
live_session_start={"type": "session.start", "session": {"model": "live-alias"}},
)
assert live_handler.called is False
@pytest.mark.asyncio
@pytest.mark.parametrize(
"key_metadata",
(
{"user_api_key_metadata": {"disable_global_guardrails": True}},
{"user_api_key_team_metadata": {"opted_out_global_guardrails": ["default-on-blocker"]}},
),
)
async def test_arealtime_key_metadata_cannot_opt_out_of_a_default_on_realtime_guardrail(
monkeypatch: pytest.MonkeyPatch, key_metadata: dict[str, object]
) -> None:
class FakeLiveHandler:
def __init__(self) -> None:
self.called = False
async def async_live_session(self, **kwargs: object) -> None:
self.called = True
live_handler: Final = FakeLiveHandler()
monkeypatch.setattr(realtime_main, "openai_live_sessions", live_handler)
monkeypatch.setattr(
litellm,
"callbacks",
[
CustomGuardrail(
guardrail_name="default-on-blocker",
event_hook=GuardrailEventHooks.realtime_input_transcription,
default_on=True,
)
],
)
with pytest.raises(ValueError, match="Guardrails are not supported on OpenAI Live sessions"):
await realtime_main._arealtime.__wrapped__(
model="openai/gpt-live-1",
websocket=MagicMock(),
api_key="fake-key",
api_base="https://live-gateway.example/v1",
litellm_logging_obj=FakeLogging(),
metadata=key_metadata,
live_session_start={"type": "session.start", "session": {"model": "gpt-live-1"}},
)
assert live_handler.called is False
@pytest.mark.asyncio
async def test_arealtime_live_session_rejects_non_openai_provider(monkeypatch: pytest.MonkeyPatch) -> None:
def mock_get_llm_provider(
model: str,
api_base: str | None,
api_key: str | None,
custom_llm_provider: str | None = None,
) -> tuple[str, str, str | None, str | None]:
return "claude-live", "anthropic", None, None
monkeypatch.setattr(realtime_main, "get_llm_provider", mock_get_llm_provider)
with pytest.raises(ValueError, match="anthropic"):
await realtime_main._arealtime.__wrapped__(
model="anthropic/claude-live",
websocket=MagicMock(),
litellm_logging_obj=FakeLogging(),
live_session_start={"type": "session.start", "session": {"model": "claude-live"}},
)
@pytest.mark.asyncio
async def test_arealtime_without_live_start_preserves_openai_realtime_dispatch(
monkeypatch: pytest.MonkeyPatch,
) -> None:
captured: Final = _CallCapture()
def mock_get_llm_provider(
model: str,
api_base: str | None,
api_key: str | None,
custom_llm_provider: str | None = None,
) -> tuple[str, str, str | None, str | None]:
return "gpt-realtime", "openai", "resolved-key", "https://realtime-gateway.example/v1"
class FakeRealtimeHandler:
async def async_realtime(self, **kwargs: object) -> None:
captured.values = kwargs
monkeypatch.setattr(realtime_main, "get_llm_provider", mock_get_llm_provider)
monkeypatch.setattr(realtime_main, "openai_realtime", FakeRealtimeHandler())
await realtime_main._arealtime.__wrapped__(
model="gpt-realtime",
websocket=MagicMock(),
litellm_logging_obj=FakeLogging(),
)
assert captured.values is not None
assert captured.values["model"] == "gpt-realtime"
assert captured.values["api_key"] == "resolved-key"
assert captured.values["api_base"] == "https://realtime-gateway.example/v1"
class _ServerEvents(Protocol):
async def recv(self) -> str: ...
class _ScriptedConnection:
def __init__(self, *frames: str) -> None:
self._frames: Final = iter(frames)
async def recv(self) -> str:
return next(self._frames)
class _SilentConnection:
async def recv(self) -> str:
await asyncio.Event().wait()
raise AssertionError("unreachable")
class _ConnectionClosedBeforeAnyEvent:
async def recv(self) -> str:
raise ConnectionClosedError(Close(3000, "invalid_api_key"), None)
_OPENAI_INVALID_API_KEY_EVENT: Final = (
'{"type": "error", "event_id": "event_1", "error": {"type": "invalid_request_error", "code": "invalid_api_key", '
'"message": "Incorrect API key provided: sk-proj-****0000. You can find your API key at '
'https://platform.openai.com/account/api-keys.", "param": null, "event_id": null}}'
)
_OPENAI_MISSING_AUTH_EVENT: Final = (
'{"type": "error", "event_id": "event_2", "error": {"type": "invalid_request_error", "code": null, '
'"message": "Missing bearer or basic authentication in header", "param": null, "event_id": null}}'
)
_OPENAI_SERVER_ERROR_EVENT: Final = (
'{"type": "error", "event_id": "event_3", '
'"error": {"type": "server_error", "code": null, "message": "The server had an error", "param": null}}'
)
_OPENAI_RATE_LIMIT_EVENT: Final = (
'{"type": "error", "event_id": "event_5", "error": {"type": "tokens", "code": "rate_limit_exceeded", '
'"message": "Rate limit reached for gpt-realtime", "param": null, "event_id": null}}'
)
_OPENAI_INSUFFICIENT_QUOTA_EVENT: Final = (
'{"type": "error", "event_id": "event_6", "error": {"type": "insufficient_quota", "code": "insufficient_quota", '
'"message": "You exceeded your current quota", "param": null, "event_id": null}}'
)
_OPENAI_LIVE_MODEL_ON_REALTIME_EVENT: Final = (
'{"type": "error", "event_id": "event_7", "error": {"type": "invalid_request_error", "code": "invalid_model", '
'"message": "Model \\"gpt-live-1\\" is not supported in realtime mode.", "param": null, "event_id": null}}'
)
_OPENAI_CHAT_MODEL_ON_REALTIME_EVENT: Final = (
'{"type": "error", "event_id": "event_8", "error": {"type": "invalid_request_error", "code": "invalid_model", '
'"message": "Model \\"gpt-5.5\\" is not supported in realtime mode.", "param": null, "event_id": null}}'
)
_OPENAI_MODEL_NOT_FOUND_EVENT: Final = (
'{"type": "error", "event_id": "event_9", "error": {"type": "invalid_request_error", "code": "model_not_found", '
'"message": "The model gpt-realtime-2.1x does not exist or you do not have access to it.", "param": null, '
'"event_id": null}}'
)
_OPENAI_SESSION_CREATED_EVENT: Final = (
'{"type": "session.created", "event_id": "event_4", "session": {"type": "realtime", "model": "gpt-realtime"}}'
)
def _register_as_realtime_deployment(model: str, *supported_endpoints: str) -> None:
litellm.Router(
model_list=[
{
"model_name": "realtime-deployment",
"litellm_params": {"model": f"openai/{model}", "api_key": "sk-real"},
"model_info": {
"mode": "realtime",
**({"supported_endpoints": list(supported_endpoints)} if supported_endpoints else {}),
},
}
]
)
class _CapturingConnect:
def __init__(self, connection: _ServerEvents | None = None) -> None:
self.url: str | None = None
self.kwargs: dict[str, object] = {}
self._connection: Final = connection or _ScriptedConnection(_OPENAI_SESSION_CREATED_EVENT)
def __call__(self, url: str, **kwargs: object) -> "_CapturingConnect":
self.url = url
self.kwargs = kwargs
return self
async def __aenter__(self) -> _ServerEvents:
return self._connection
async def __aexit__(
self,
exc_type: type[BaseException] | None,
exc: BaseException | None,
tb: TracebackType | None,
) -> None:
return None
@pytest.mark.asyncio
async def test_azure_health_check_resolves_stored_credentials(monkeypatch):
monkeypatch.setattr(
litellm,
"credential_list",
[
CredentialItem(
credential_name="azure-rt",
credential_values={
"api_key": "sk-from-credential",
"api_base": "https://example.openai.azure.com",
"api_version": "2025-04-01-preview",
},
credential_info={},
)
],
)
connect = _CapturingConnect()
with patch("websockets.connect", connect):
assert await realtime_main._realtime_health_check(
model="gpt-realtime",
custom_llm_provider="azure",
api_key=None,
realtime_protocol="beta",
model_params={"model": "azure/gpt-realtime", "litellm_credential_name": "azure-rt"},
)
assert connect.kwargs["additional_headers"] == {"api-key": "sk-from-credential"}
assert connect.url is not None
assert connect.url.startswith("wss://example.openai.azure.com")
assert "api-version=2025-04-01-preview" in connect.url
@pytest.mark.asyncio
@pytest.mark.parametrize(
("custom_llm_provider", "model", "expected_url"),
[
("xai", "grok-voice-latest", "wss://api.x.ai/v1/realtime?model=grok-voice-latest"),
("openai", "gpt-realtime", "wss://api.openai.com/v1/realtime?model=gpt-realtime"),
],
)
async def test_bearer_health_check_sends_stored_credential_as_bearer_token(
monkeypatch, custom_llm_provider: str, model: str, expected_url: str
):
monkeypatch.setattr(
litellm,
"credential_list",
[
CredentialItem(
credential_name="voice-key",
credential_values={"api_key": "sk-from-credential"},
credential_info={},
)
],
)
connect = _CapturingConnect()
with patch("websockets.connect", connect):
assert await realtime_main._realtime_health_check(
model=model,
custom_llm_provider=custom_llm_provider,
api_key=None,
model_params={"model": f"{custom_llm_provider}/{model}", "litellm_credential_name": "voice-key"},
)
assert connect.kwargs["additional_headers"] == {"Authorization": "Bearer sk-from-credential"}
assert connect.url == expected_url
@pytest.mark.asyncio
async def test_bearer_health_check_sends_the_secret_behind_an_oidc_credential(
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
):
monkeypatch.setenv("LITELLM_OIDC_ALLOWED_CREDENTIAL_DIRS", str(tmp_path))
token: Final = tmp_path / "token"
token.write_text("sk-from-token-file")
monkeypatch.setattr(
litellm,
"credential_list",
[
CredentialItem(
credential_name="voice-sa", credential_values={"api_key": f"oidc/file/{token}"}, credential_info={}
)
],
)
connect: Final = _CapturingConnect()
with patch("websockets.connect", connect):
assert await realtime_main._realtime_health_check(
model="gpt-realtime",
custom_llm_provider="openai",
api_key=None,
model_params={"model": "openai/gpt-realtime", "litellm_credential_name": "voice-sa"},
)
assert connect.kwargs["additional_headers"] == {"Authorization": "Bearer sk-from-token-file"}
@pytest.mark.asyncio
async def test_azure_health_check_probes_ga_transcription_url_for_transcription_model(local_model_cost_map):
"""Regression for LIT-6240: transcription-only models (mode audio_transcription
in the cost map) are GA-only and 400 on the beta path, so the health probe
must hit /openai/v1/realtime?intent=transcription like real calls do."""
connect = _CapturingConnect()
with patch("websockets.connect", connect):
assert await realtime_main._realtime_health_check(
model="gpt-realtime-whisper",
custom_llm_provider="azure",
api_key="fake-key",
api_base="https://my-endpoint.openai.azure.com",
api_version="2025-04-01-preview",
)
assert connect.url == "wss://my-endpoint.openai.azure.com/openai/v1/realtime?intent=transcription"
@pytest.mark.asyncio
async def test_azure_health_check_stays_on_ga_when_deployment_registration_overwrites_mode(
local_model_cost_map, monkeypatch
):
"""In a live proxy, Router._register_deployment_in_model_cost writes the
operator's deployment model_info (mode: realtime) over the catalog entry for
azure/gpt-realtime-whisper, so mode alone misreads the model as speech-capable
and the probe regresses to the beta path. supported_endpoints survives that
registration and must keep the probe on the GA transcription path."""
polluted = {**litellm.model_cost["azure/gpt-realtime-whisper"], "mode": "realtime"}
monkeypatch.setitem(litellm.model_cost, "azure/gpt-realtime-whisper", polluted)
connect = _CapturingConnect()
with patch("websockets.connect", connect):
assert await realtime_main._realtime_health_check(
model="gpt-realtime-whisper",
custom_llm_provider="azure",
api_key="fake-key",
api_base="https://my-endpoint.openai.azure.com",
api_version="2025-04-01-preview",
)
assert connect.url == "wss://my-endpoint.openai.azure.com/openai/v1/realtime?intent=transcription"
def test_transcription_only_detection_falls_back_to_mode(local_model_cost_map):
"""azure/whisper-1 declares mode audio_transcription but no supported_endpoints,
so only the mode signal can classify it as transcription-only."""
assert realtime_main._is_transcription_only_realtime_model("whisper-1", "azure") is True
def test_transcription_only_detection_rejects_speech_model(local_model_cost_map):
assert realtime_main._is_transcription_only_realtime_model("gpt-realtime-mini", "azure") is False
@pytest.mark.asyncio
async def test_azure_health_check_probes_the_ga_upstream_for_an_unconfigured_speech_model(monkeypatch):
monkeypatch.delenv("LITELLM_AZURE_REALTIME_PROTOCOL", raising=False)
connect = _CapturingConnect()
with patch("websockets.connect", connect):
assert await realtime_main._realtime_health_check(
model="gpt-4o-realtime-preview",
custom_llm_provider="azure",
api_key="fake-key",
api_base="https://my-endpoint.openai.azure.com",
api_version="2024-10-01-preview",
)
assert connect.url == "wss://my-endpoint.openai.azure.com/openai/v1/realtime?model=gpt-4o-realtime-preview"
_AZURE_BETA_HEALTH_URL: Final = (
"wss://my-endpoint.openai.azure.com/openai/realtime"
"?api-version=2024-10-01-preview&deployment=gpt-4o-realtime-preview"
)
@pytest.mark.asyncio
async def test_azure_health_check_honors_deployment_realtime_protocol(monkeypatch):
monkeypatch.delenv("LITELLM_AZURE_REALTIME_PROTOCOL", raising=False)
connect = _CapturingConnect()
with patch("websockets.connect", connect):
assert await realtime_main._realtime_health_check(
model="gpt-4o-realtime-preview",
custom_llm_provider="azure",
api_key="fake-key",
api_base="https://my-endpoint.openai.azure.com",
api_version="2024-10-01-preview",
model_params={"realtime_protocol": "beta"},
)
assert connect.url == _AZURE_BETA_HEALTH_URL
@pytest.mark.asyncio
async def test_azure_health_check_honors_env_realtime_protocol(monkeypatch):
monkeypatch.setenv("LITELLM_AZURE_REALTIME_PROTOCOL", "beta")
connect = _CapturingConnect()
with patch("websockets.connect", connect):
assert await realtime_main._realtime_health_check(
model="gpt-4o-realtime-preview",
custom_llm_provider="azure",
api_key="fake-key",
api_base="https://my-endpoint.openai.azure.com",
api_version="2024-10-01-preview",
)
assert connect.url == _AZURE_BETA_HEALTH_URL
@pytest.mark.asyncio
@pytest.mark.parametrize(
("first_event", "expected_exception", "expected_status", "expected_message"),
[
(_OPENAI_INVALID_API_KEY_EVENT, litellm.AuthenticationError, 401, "Incorrect API key provided"),
(_OPENAI_SERVER_ERROR_EVENT, litellm.InternalServerError, 500, "The server had an error"),
(_OPENAI_RATE_LIMIT_EVENT, litellm.RateLimitError, 429, "Rate limit reached"),
(_OPENAI_INSUFFICIENT_QUOTA_EVENT, litellm.RateLimitError, 429, "You exceeded your current quota"),
],
)
async def test_openai_health_check_reports_the_first_error_event_as_unhealthy(
first_event: str, expected_exception: type[openai.APIStatusError], expected_status: int, expected_message: str
):
connect: Final = _CapturingConnect(_ScriptedConnection(first_event))
with patch("websockets.connect", connect), pytest.raises(expected_exception) as raised:
await realtime_main.realtime_health_check(
model="gpt-realtime", custom_llm_provider="openai", api_key="sk-wrong"
)
assert raised.value.status_code == expected_status
assert expected_message in str(raised.value)
assert connect.url == "wss://api.openai.com/v1/realtime?model=gpt-realtime"
@pytest.mark.asyncio
async def test_openai_health_check_without_an_api_key_sends_no_auth_header_and_fails_authentication():
connect: Final = _CapturingConnect(_ScriptedConnection(_OPENAI_MISSING_AUTH_EVENT))
with patch("websockets.connect", connect), pytest.raises(litellm.AuthenticationError) as raised:
await realtime_main.realtime_health_check(model="gpt-realtime", custom_llm_provider="openai", api_key=None)
assert connect.kwargs["additional_headers"] == {}
assert raised.value.status_code == 401
assert "Missing bearer or basic authentication" in str(raised.value)
@pytest.mark.asyncio
async def test_openai_health_check_is_healthy_once_session_created_arrives():
connect: Final = _CapturingConnect(_ScriptedConnection(_OPENAI_SESSION_CREATED_EVENT))
with patch("websockets.connect", connect):
assert await realtime_main.realtime_health_check(
model="gpt-realtime", custom_llm_provider="openai", api_key="sk-real"
)
@pytest.mark.asyncio
@pytest.mark.parametrize("model", ["gpt-live-1", "gpt-4o-transcribe", "gpt-audio", "gpt-live-unreleased"])
async def test_openai_health_check_is_healthy_for_a_model_served_outside_realtime_once_the_key_is_accepted(
local_model_cost_map, model: str
):
connect: Final = _CapturingConnect(_ScriptedConnection(_OPENAI_LIVE_MODEL_ON_REALTIME_EVENT))
with patch("websockets.connect", connect):
assert await realtime_main.realtime_health_check(model=model, custom_llm_provider="openai", api_key="sk-real")
@pytest.mark.asyncio
async def test_openai_health_check_is_healthy_for_a_live_deployment_that_lists_the_live_sessions_endpoint(
local_model_cost_map,
):
_register_as_realtime_deployment("gpt-live-1", "/v1/live/sessions")
connect: Final = _CapturingConnect(_ScriptedConnection(_OPENAI_LIVE_MODEL_ON_REALTIME_EVENT))
with patch("websockets.connect", connect):
assert await realtime_main.realtime_health_check(
model="gpt-live-1", custom_llm_provider="openai", api_key="sk-real"
)
@pytest.mark.asyncio
@pytest.mark.parametrize(
("model", "first_event", "expected_exception", "expected_message"),
[
(
"gpt-5.5",
_OPENAI_CHAT_MODEL_ON_REALTIME_EVENT,
litellm.BadRequestError,
'Model "gpt-5.5" is not supported in realtime mode',
),
(
"gpt-realtime-2.1x",
_OPENAI_MODEL_NOT_FOUND_EVENT,
litellm.NotFoundError,
"The model gpt-realtime-2.1x does not exist",
),
],
)
async def test_openai_health_check_is_unhealthy_for_a_model_name_every_session_would_fail_on(
local_model_cost_map, model: str, first_event: str, expected_exception: type[Exception], expected_message: str
):
_register_as_realtime_deployment(model)
connect: Final = _CapturingConnect(_ScriptedConnection(first_event))
with patch("websockets.connect", connect), pytest.raises(expected_exception) as raised:
await realtime_main.realtime_health_check(model=model, custom_llm_provider="openai", api_key="sk-real")
assert expected_message in str(raised.value)
@pytest.mark.asyncio
@pytest.mark.parametrize(
("deployment_api_base", "global_api_base"),
[("https://realtime-gateway.example/v1", None), (None, "https://realtime-gateway.example/v1")],
)
async def test_openai_health_check_through_a_compatible_gateway_is_healthy_when_it_sends_no_first_event(
monkeypatch: pytest.MonkeyPatch, deployment_api_base: str | None, global_api_base: str | None
):
monkeypatch.setattr(litellm, "api_base", global_api_base)
connect: Final = _CapturingConnect(_SilentConnection())
with patch("websockets.connect", connect):
assert await asyncio.wait_for(
realtime_main.realtime_health_check(
model="gpt-realtime",
custom_llm_provider="openai",
api_key="sk-real",
api_base=deployment_api_base,
gateway_first_event_wait_seconds=0.01,
),
timeout=5,
)
assert connect.url == "wss://realtime-gateway.example/v1/realtime?model=gpt-realtime"
@pytest.mark.asyncio
@pytest.mark.parametrize(
("connection", "expected_exception"),
[
(_ScriptedConnection(_OPENAI_INVALID_API_KEY_EVENT), litellm.AuthenticationError),
(_ConnectionClosedBeforeAnyEvent(), ConnectionClosedError),
],
)
async def test_openai_health_check_through_a_compatible_gateway_is_unhealthy_when_it_rejects_the_session(
connection: _ServerEvents, expected_exception: type[Exception]
):
connect: Final = _CapturingConnect(connection)
with patch("websockets.connect", connect), pytest.raises(expected_exception):
await realtime_main.realtime_health_check(
model="gpt-realtime",
custom_llm_provider="openai",
api_key="sk-wrong",
api_base="https://realtime-gateway.example/v1",
)
assert connect.url == "wss://realtime-gateway.example/v1/realtime?model=gpt-realtime"
@pytest.mark.asyncio
async def test_openai_health_check_never_reports_healthy_before_the_first_event_arrives():
connect: Final = _CapturingConnect(_SilentConnection())
with patch("websockets.connect", connect), pytest.raises(asyncio.TimeoutError):
await asyncio.wait_for(
realtime_main.realtime_health_check(model="gpt-realtime", custom_llm_provider="openai", api_key="sk-real"),
timeout=0.05,
)
@pytest.mark.asyncio
async def test_openai_health_check_is_unhealthy_when_the_socket_closes_before_any_event():
connect: Final = _CapturingConnect(_ConnectionClosedBeforeAnyEvent())
with patch("websockets.connect", connect), pytest.raises(ConnectionClosedError):
await realtime_main.realtime_health_check(
model="gpt-realtime", custom_llm_provider="openai", api_key="sk-wrong"
)
@pytest.mark.asyncio
async def test_xai_health_check_trusts_the_handshake_without_reading_a_first_event():
connect: Final = _CapturingConnect(_ScriptedConnection())
with patch("websockets.connect", connect):
assert await realtime_main.realtime_health_check(
model="grok-voice-latest", custom_llm_provider="xai", api_key="sk-wrong"
)
assert connect.url == "wss://api.x.ai/v1/realtime?model=grok-voice-latest"
class _ConnectThatStopsAfterCapturingTheUrl:
url: str | None = None
def __call__(self, url: str, **kwargs: object) -> "_ConnectThatStopsAfterCapturingTheUrl":
self.url = url
return self
async def __aenter__(self) -> None:
raise RuntimeError("backend url captured, nothing to bridge")
async def __aexit__(
self,
exc_type: type[BaseException] | None,
exc: BaseException | None,
tb: TracebackType | None,
) -> None:
return None
@pytest.mark.asyncio
async def test_arealtime_azure_ai_on_a_foundry_host_connects_to_the_azure_openai_realtime_route():
connect: Final = _ConnectThatStopsAfterCapturingTheUrl()
with patch("websockets.connect", connect):
await realtime_main._arealtime.__wrapped__(
model="azure_ai/gpt-realtime-mini",
websocket=MagicMock(),
api_base="https://my-project.services.ai.azure.com",
api_key="fake-key",
litellm_logging_obj=FakeLogging(),
)
assert connect.url == "wss://my-project.services.ai.azure.com/openai/v1/realtime?model=gpt-realtime-mini"
@pytest.mark.asyncio
async def test_arealtime_routes_a_bare_model_name_to_its_explicit_provider():
connect: Final = _ConnectThatStopsAfterCapturingTheUrl()
with patch("websockets.connect", connect):
await realtime_main._arealtime.__wrapped__(
model="my-realtime",
websocket=MagicMock(close=AsyncMock()),
custom_llm_provider="openai",
api_base="https://realtime-gateway.example/v1",
api_key="fake-key",
query_params={"model": "my-realtime"},
litellm_logging_obj=FakeLogging(),
)
assert connect.url == "wss://realtime-gateway.example/v1/realtime?model=my-realtime"
class _GatewayThatRecordsTheUrl:
url: str | None = None
def __call__(self, request: httpx.Request) -> httpx.Response:
self.url = str(request.url)
return httpx.Response(200, json={"value": "ek_test"})
@pytest.mark.asyncio
@pytest.mark.parametrize(
("route", "route_kwargs", "path"),
[
(realtime_main.acreate_realtime_client_secret, {}, "client_secrets"),
(realtime_main.acreate_realtime_transcription_session, {}, "transcription_sessions"),
(realtime_main.arealtime_calls, {"openai_ephemeral_key": "ek_test", "sdp_body": b"v=0"}, "calls"),
],
)
async def test_realtime_http_routes_send_a_bare_model_name_to_its_explicit_provider(
route: Callable[..., Awaitable[object]], route_kwargs: Mapping[str, object], path: str
) -> None:
gateway: Final = _GatewayThatRecordsTheUrl()
await route.__wrapped__(
model="my-realtime",
custom_llm_provider="openai",
api_base="https://realtime-gateway.example/v1",
api_key="fake-key",
litellm_logging_obj=FakeLogging(),
client=AsyncHTTPHandler(transport=httpx.MockTransport(gateway)),
**route_kwargs,
)
assert gateway.url == f"https://realtime-gateway.example/v1/realtime/{path}"
class _GatewayThatRecordsTheBody:
body: object = None
def __call__(self, request: httpx.Request) -> httpx.Response:
self.body = json.loads(request.content)
return httpx.Response(200, json={"client_secret": {"value": "ek_test"}})
@pytest.mark.asyncio
async def test_transcription_session_sends_the_deployment_model_upstream_in_every_transcription_shape() -> None:
gateway: Final = _GatewayThatRecordsTheBody()
await realtime_main.acreate_realtime_transcription_session.__wrapped__(
model="openai/gpt-transcribe",
transcription_session={
"audio": {"input": {"transcription": {"model": "team-transcriber", "language": "en"}}},
"input_audio_transcription": {"model": "team-transcriber"},
"input_audio_format": "pcm16",
},
api_base="https://realtime-gateway.example/v1",
api_key="fake-key",
litellm_logging_obj=FakeLogging(),
client=AsyncHTTPHandler(transport=httpx.MockTransport(gateway)),
)
assert gateway.body == {
"audio": {"input": {"transcription": {"model": "gpt-transcribe", "language": "en"}}},
"input_audio_transcription": {"model": "gpt-transcribe"},
"input_audio_format": "pcm16",
}
def _create_realtime_session_with_an_explicit_null(route: str, client: AsyncHTTPHandler) -> Awaitable[httpx.Response]:
if route == "client_secrets":
return realtime_main.acreate_realtime_client_secret.__wrapped__(
session={
"type": "realtime",
"model": "gpt-realtime",
"instructions": None,
"tools": None,
"audio": {"input": {"turn_detection": None}},
},
custom_llm_provider="openai",
api_base="https://realtime-gateway.example/v1",
api_key="fake-key",
litellm_logging_obj=FakeLogging(),
client=client,
)
if route == "translation_client_secrets":
return realtime_main.acreate_realtime_translation_client_secret.__wrapped__(
model="gpt-realtime-translate",
session={"include": None, "audio": {"input": {"noise_reduction": None}}},
custom_llm_provider="openai",
api_base="https://realtime-gateway.example/v1",
api_key="fake-key",
litellm_logging_obj=_realtime_http_logging(route),
client=client,
)
return realtime_main.acreate_realtime_transcription_session.__wrapped__(
model="openai/gpt-transcribe",
transcription_session={"include": None, "audio": {"input": {"turn_detection": None}}},
api_base="https://realtime-gateway.example/v1",
api_key="fake-key",
litellm_logging_obj=FakeLogging(),
client=client,
)
@pytest.mark.asyncio
@pytest.mark.parametrize(
("route", "body_path"),
(
("client_secrets", ("session", "audio", "input", "turn_detection")),
("translation_client_secrets", ("session", "audio", "input", "noise_reduction")),
("transcription_sessions", ("audio", "input", "turn_detection")),
),
)
async def test_realtime_session_requests_forward_an_explicit_null_upstream(
route: str, body_path: tuple[str, ...]
) -> None:
gateway: Final = _GatewayThatRecordsTheBody()
await _create_realtime_session_with_an_explicit_null(
route, AsyncHTTPHandler(transport=httpx.MockTransport(gateway))
)
parent: Final = functools.reduce(lambda node, key: node[key], body_path[:-1], gateway.body)
assert body_path[-1] in parent
assert parent[body_path[-1]] is None
session_fields: Final = gateway.body["session"] if body_path[0] == "session" else gateway.body
assert [key for key, value in session_fields.items() if value is None] == []
def _create_translation_client_secret(expires_after: object, gateway: _GatewayThatRecordsTheBody) -> Awaitable[object]:
return realtime_main.acreate_realtime_translation_client_secret.__wrapped__(
model="gpt-realtime-translate",
expires_after=expires_after,
custom_llm_provider="openai",
api_base="https://realtime-gateway.example/v1",
api_key="fake-key",
litellm_logging_obj=_realtime_http_logging("translation_client_secrets"),
client=AsyncHTTPHandler(transport=httpx.MockTransport(gateway)),
)
@pytest.mark.asyncio
@pytest.mark.parametrize(
"expires_after",
[
pytest.param({"anchor": "created_at", "seconds": 600}, id="dict"),
pytest.param(MappingProxyType({"anchor": "created_at", "seconds": 600}), id="read-only-mapping"),
],
)
async def test_translation_client_secret_forwards_a_mapping_expires_after(expires_after: Mapping[str, object]) -> None:
gateway: Final = _GatewayThatRecordsTheBody()
await _create_translation_client_secret(expires_after, gateway)
assert gateway.body["expires_after"] == {"anchor": "created_at", "seconds": 600}
@pytest.mark.asyncio
@pytest.mark.parametrize(
"expires_after",
[
pytest.param([("anchor", "created_at"), ("seconds", 600)], id="list-of-pairs"),
pytest.param("seconds=600", id="string"),
],
)
async def test_translation_client_secret_rejects_an_expires_after_that_is_not_a_mapping(expires_after: object) -> None:
gateway: Final = _GatewayThatRecordsTheBody()
with pytest.raises(TypeError, match=r"realtime\.RealtimeExpiresAfter\(\) argument after \*\* must be a mapping"):
await _create_translation_client_secret(expires_after, gateway)
assert gateway.body is None
@pytest.mark.asyncio
@pytest.mark.parametrize(
"transcription_session",
[
pytest.param([("input_audio_format", "pcm16")], id="list-of-pairs"),
pytest.param("input_audio_format=pcm16", id="string"),
],
)
async def test_transcription_session_rejects_a_session_that_is_not_a_mapping(transcription_session: object) -> None:
gateway: Final = _GatewayThatRecordsTheBody()
with pytest.raises(
TypeError, match=r"realtime\.RealtimeTranscriptionSessionRequest\(\) argument after \*\* must be a mapping"
):
await realtime_main.acreate_realtime_transcription_session.__wrapped__(
model="openai/gpt-transcribe",
transcription_session=transcription_session,
api_base="https://realtime-gateway.example/v1",
api_key="fake-key",
litellm_logging_obj=FakeLogging(),
client=AsyncHTTPHandler(transport=httpx.MockTransport(gateway)),
)
assert gateway.body is None
@pytest.mark.asyncio
async def test_transcription_session_accepts_a_read_only_mapping() -> None:
gateway: Final = _GatewayThatRecordsTheBody()
await realtime_main.acreate_realtime_transcription_session.__wrapped__(
model="openai/gpt-transcribe",
transcription_session=MappingProxyType({"input_audio_format": "pcm16"}),
api_base="https://realtime-gateway.example/v1",
api_key="fake-key",
litellm_logging_obj=FakeLogging(),
client=AsyncHTTPHandler(transport=httpx.MockTransport(gateway)),
)
assert gateway.body["input_audio_format"] == "pcm16"
@pytest.mark.asyncio
async def test_transcription_client_secret_sends_the_deployment_model_as_the_session_model() -> None:
gateway: Final = _GatewayThatRecordsTheSession()
await realtime_main.acreate_realtime_client_secret.__wrapped__(
session={"type": "transcription", "model": "openai/gpt-live-transcribe"},
api_key="sk-test",
api_base="https://upstream.example",
litellm_logging_obj=_realtime_http_logging("client_secrets"),
client=AsyncHTTPHandler(transport=httpx.MockTransport(gateway)),
)
assert gateway.session is not None
assert gateway.session["model"] == "gpt-live-transcribe"
assert gateway.session["audio"] == {"input": {"transcription": {"model": "gpt-live-transcribe"}}}
class _ClientWebSocketWithHeaders:
def __init__(self, headers: tuple[tuple[bytes, bytes], ...]) -> None:
self.scope: Final = {"headers": headers}
_GA_CLIENT: Final = _ClientWebSocketWithHeaders(headers=())
_BETA_CLIENT: Final = _ClientWebSocketWithHeaders(headers=((b"openai-beta", b"realtime=v1"),))
_GA_DEFAULT_SESSIONS: Final = (
pytest.param("gpt-realtime-2.1", "realtime", None, id="ga-only-model"),
pytest.param("gpt-realtime-2", "realtime", None, id="ga-only-alias"),
pytest.param("gpt-realtime-2-2026-05-06", "realtime", None, id="ga-only-snapshot"),
# Azure answers beta sessions on these with "The Realtime preview API is no longer available"
# (https://aka.ms/aoai/realtime-ga), checked live 2026-10-28
pytest.param("gpt-realtime", "realtime", None, id="beta-retired-gpt-realtime"),
pytest.param("gpt-realtime-mini", "realtime", None, id="beta-retired-gpt-realtime-mini"),
pytest.param("gpt-realtime-1.5", "realtime", None, id="beta-retired-gpt-realtime-1.5"),
pytest.param("my-transcribe-deployment", "realtime", {"intent": "transcription"}, id="transcription-intent"),
pytest.param("my-translate-deployment", "translation", None, id="translation-mode"),
)
@pytest.mark.parametrize(("model", "realtime_mode", "query_params"), _GA_DEFAULT_SESSIONS)
def test_azure_realtime_protocol_defaults_to_ga_when_none_is_configured(
local_model_cost_map, model: str, realtime_mode: str, query_params: RealtimeQueryParams | None
) -> None:
from litellm.llms.azure.realtime.handler import azure_realtime_protocol_for_client
protocol: Final = azure_realtime_protocol_for_client(
None, model=model, realtime_mode=realtime_mode, query_params=query_params, websocket=_BETA_CLIENT
)
assert protocol.name == "GA"
@pytest.mark.parametrize(("model", "realtime_mode", "query_params"), _GA_DEFAULT_SESSIONS)
def test_azure_realtime_protocol_configured_by_the_operator_wins(
local_model_cost_map, model: str, realtime_mode: str, query_params: RealtimeQueryParams | None
) -> None:
from litellm.llms.azure.realtime.handler import azure_realtime_protocol_for_client
protocol: Final = azure_realtime_protocol_for_client(
"beta", model=model, realtime_mode=realtime_mode, query_params=query_params, websocket=_GA_CLIENT
)
assert protocol.name == "beta"
async def _azure_backend_url_dialed_for(websocket: _ClientWebSocketWithHeaders, **kwargs: object) -> str | None:
connect: Final = _ConnectThatStopsAfterCapturingTheUrl()
with patch("websockets.connect", connect):
await realtime_main._arealtime.__wrapped__(
model="azure/my-realtime-deployment",
websocket=websocket,
api_base="https://my-endpoint.openai.azure.com",
api_key="fake-key",
litellm_logging_obj=FakeLogging(),
**kwargs,
)
return connect.url
@pytest.mark.asyncio
async def test_arealtime_azure_ga_client_without_beta_header_dials_the_ga_upstream(monkeypatch):
monkeypatch.delenv("LITELLM_AZURE_REALTIME_PROTOCOL", raising=False)
assert (
await _azure_backend_url_dialed_for(_GA_CLIENT)
== "wss://my-endpoint.openai.azure.com/openai/v1/realtime?model=my-realtime-deployment"
)
@pytest.mark.asyncio
async def test_arealtime_azure_beta_header_client_keeps_the_beta_upstream(monkeypatch):
monkeypatch.delenv("LITELLM_AZURE_REALTIME_PROTOCOL", raising=False)
assert await _azure_backend_url_dialed_for(_BETA_CLIENT) == (
"wss://my-endpoint.openai.azure.com/openai/realtime?api-version=2024-10-01-preview&deployment=my-realtime-deployment"
)
@pytest.mark.asyncio
async def test_arealtime_azure_explicit_beta_protocol_wins_over_a_ga_client(monkeypatch):
monkeypatch.delenv("LITELLM_AZURE_REALTIME_PROTOCOL", raising=False)
assert await _azure_backend_url_dialed_for(_GA_CLIENT, realtime_protocol="beta") == (
"wss://my-endpoint.openai.azure.com/openai/realtime?api-version=2024-10-01-preview&deployment=my-realtime-deployment"
)
@pytest.mark.asyncio
async def test_arealtime_azure_env_beta_protocol_wins_over_a_ga_client(monkeypatch):
monkeypatch.setenv("LITELLM_AZURE_REALTIME_PROTOCOL", "beta")
assert await _azure_backend_url_dialed_for(_GA_CLIENT) == (
"wss://my-endpoint.openai.azure.com/openai/realtime?api-version=2024-10-01-preview&deployment=my-realtime-deployment"
)
class _BetaClientThatStaysConnected:
def __init__(self) -> None:
self.scope: Final = {"headers": ((b"openai-beta", b"realtime=v1"),)}
self.sent: tuple[str, ...] = ()
async def receive_text(self) -> str:
return await asyncio.get_running_loop().create_future()
async def send_text(self, data: str) -> None:
self.sent = (*self.sent, data)
async def close(self, code: int = 1000, reason: str = "") -> None:
return None
class _ConnectToABackendThatSendsOneFrame:
def __init__(self, frame: Mapping[str, object]) -> None:
self.url: str | None = None
self._backend: Final = MagicMock(recv=AsyncMock(side_effect=[json.dumps(frame), ConnectionClosed(None, None)]))
def __call__(
self, url: str, *, additional_headers: Mapping[str, str], max_size: int, ssl: object
) -> "_ConnectToABackendThatSendsOneFrame":
self.url = url
return self
async def __aenter__(self) -> MagicMock:
return self._backend
async def __aexit__(
self,
exc_type: type[BaseException] | None,
exc: BaseException | None,
tb: TracebackType | None,
) -> None:
return None
def _realtime_websocket_logging(*failure_callbacks: CustomLogger) -> Logging:
return Logging(
model="gpt-realtime",
messages=[],
stream=False,
call_type="_arealtime",
start_time=datetime(2026, 1, 1, tzinfo=timezone.utc),
litellm_call_id="test-realtime-websocket",
function_id="test-realtime-websocket",
dynamic_async_failure_callbacks=list(failure_callbacks),
)
_GA_SESSION_CREATED: Final = {
"type": "session.created",
"event_id": "event_1",
"session": {
"type": "realtime",
"id": "sess_1",
"model": "my-realtime-deployment",
"output_modalities": ["text"],
"audio": {"output": {"voice": "marin"}},
},
}
async def _azure_session_created_seen_by_a_beta_client(
session_kwargs: Mapping[str, object],
) -> tuple[str | None, object]:
client: Final = _BetaClientThatStaysConnected()
connect: Final = _ConnectToABackendThatSendsOneFrame(_GA_SESSION_CREATED)
with patch("websockets.connect", connect):
await realtime_main._arealtime.__wrapped__(
model="azure/my-realtime-deployment",
websocket=client,
api_base="https://my-endpoint.openai.azure.com",
api_key="fake-key",
litellm_logging_obj=_realtime_websocket_logging(),
**session_kwargs,
)
return connect.url, json.loads(client.sent[0])
@pytest.mark.asyncio
@pytest.mark.parametrize(
("session_kwargs", "expected_url"),
(
pytest.param(
{"query_params": {"intent": "transcription"}},
"wss://my-endpoint.openai.azure.com/openai/v1/realtime?intent=transcription",
id="transcription-intent",
),
),
)
async def test_arealtime_azure_beta_client_on_a_ga_session_it_did_not_need_bridging_for_keeps_ga_session_events(
monkeypatch: pytest.MonkeyPatch, session_kwargs: Mapping[str, object], expected_url: str
) -> None:
monkeypatch.delenv("LITELLM_AZURE_REALTIME_PROTOCOL", raising=False)
assert await _azure_session_created_seen_by_a_beta_client(session_kwargs) == (expected_url, _GA_SESSION_CREATED)
@pytest.mark.asyncio
@pytest.mark.parametrize(
("env_protocol", "session_kwargs"),
(
pytest.param(None, {"realtime_protocol": "GA"}, id="operator-set-ga"),
pytest.param("GA", {}, id="env-set-ga"),
),
)
async def test_arealtime_azure_bridges_a_beta_client_to_ga_when_the_operator_pins_ga(
monkeypatch: pytest.MonkeyPatch, env_protocol: str | None, session_kwargs: Mapping[str, object]
) -> None:
if env_protocol is None:
monkeypatch.delenv("LITELLM_AZURE_REALTIME_PROTOCOL", raising=False)
else:
monkeypatch.setenv("LITELLM_AZURE_REALTIME_PROTOCOL", env_protocol)
assert await _azure_session_created_seen_by_a_beta_client(session_kwargs) == (
"wss://my-endpoint.openai.azure.com/openai/v1/realtime?model=my-realtime-deployment",
_GA_SESSION_CREATED_BRIDGED_TO_BETA,
)
@pytest.mark.asyncio
@pytest.mark.parametrize(
"base_model_kwargs",
(
pytest.param({"base_model": "azure/gpt-realtime-2.1"}, id="base-model-param"),
pytest.param({"metadata": {"model_info": {"base_model": "azure/gpt-realtime-2.1"}}}, id="model-info"),
),
)
async def test_arealtime_azure_bridges_a_beta_client_to_ga_when_the_deployment_base_model_is_ga_only(
local_model_cost_map, monkeypatch: pytest.MonkeyPatch, base_model_kwargs: Mapping[str, object]
) -> None:
monkeypatch.delenv("LITELLM_AZURE_REALTIME_PROTOCOL", raising=False)
assert await _azure_session_created_seen_by_a_beta_client(base_model_kwargs) == (
"wss://my-endpoint.openai.azure.com/openai/v1/realtime?model=my-realtime-deployment",
_GA_SESSION_CREATED_BRIDGED_TO_BETA,
)
_GA_SESSION_CREATED_BRIDGED_TO_BETA: Final = {
"type": "session.created",
"event_id": "event_1",
"session": {"id": "sess_1", "model": "my-realtime-deployment", "modalities": ["text"], "voice": "marin"},
}
_ROUTER_REQUEST_SHAPES: Final = (
pytest.param({}, id="sdk-router-call"),
pytest.param({"metadata": {"user_api_key_hash": "hashed-key"}}, id="proxy-request-metadata"),
)
def _router_with_an_azure_deployment_aliasing(deployment_model: str, base_model: str) -> litellm.Router:
return litellm.Router(
model_list=[
{
"model_name": "team-alias",
"litellm_params": {
"model": deployment_model,
"api_base": "https://my-endpoint.openai.azure.com",
"api_key": "fake-key",
},
"model_info": {"base_model": base_model},
}
]
)
@pytest.mark.asyncio
@pytest.mark.parametrize("request_kwargs", _ROUTER_REQUEST_SHAPES)
async def test_router_realtime_bridges_a_beta_client_to_ga_when_model_info_base_model_is_ga_only(
local_model_cost_map, monkeypatch: pytest.MonkeyPatch, request_kwargs: Mapping[str, object]
) -> None:
monkeypatch.delenv("LITELLM_AZURE_REALTIME_PROTOCOL", raising=False)
router: Final = _router_with_an_azure_deployment_aliasing(
"azure/my-realtime-deployment", "azure/gpt-realtime-2.1-mini"
)
client: Final = _BetaClientThatStaysConnected()
connect: Final = _ConnectToABackendThatSendsOneFrame(_GA_SESSION_CREATED)
with patch("websockets.connect", connect):
await router._arealtime(model="team-alias", websocket=client, **request_kwargs)
assert (connect.url, json.loads(client.sent[0])) == (
"wss://my-endpoint.openai.azure.com/openai/v1/realtime?model=my-realtime-deployment",
_GA_SESSION_CREATED_BRIDGED_TO_BETA,
)
async def _vertex_provider_config_for(monkeypatch, model: str, vertex_location: str | None):
from litellm.llms.vertex_ai.audio_transcription.realtime_transformation import VertexChirpRealtimeConfig
from litellm.llms.vertex_ai.realtime.transformation import VertexAIRealtimeConfig
captured: dict[str, object] = {}
def mock_get_llm_provider(model, api_base, api_key, custom_llm_provider: str | None = None):
return model.removeprefix("vertex_ai/"), "vertex_ai", None, api_base
async def mock_token_resolver(**kwargs):
return "access-token", kwargs["project_id"]
async def mock_async_realtime(**kwargs):
captured.update(kwargs)
monkeypatch.setattr(realtime_main, "get_llm_provider", mock_get_llm_provider)
monkeypatch.setattr(realtime_main, "vertex_access_token_resolver", mock_token_resolver)
monkeypatch.setattr(realtime_main.base_llm_http_handler, "async_realtime", mock_async_realtime)
monkeypatch.setattr(litellm, "vertex_location", None)
monkeypatch.delenv("VERTEXAI_LOCATION", raising=False)
await realtime_main._arealtime.__wrapped__(
model=model,
websocket=MagicMock(),
litellm_logging_obj=FakeLogging(),
query_params={"model": model, "intent": "transcription"},
vertex_credentials="fake-credentials",
vertex_project="proj-1",
vertex_location=vertex_location,
)
provider_config = captured["provider_config"]
assert isinstance(provider_config, (VertexAIRealtimeConfig, VertexChirpRealtimeConfig))
return provider_config, captured["model"]
@pytest.mark.asyncio
async def test_arealtime_routes_chirp_models_to_the_speech_to_text_backend(monkeypatch):
from litellm.llms.vertex_ai.audio_transcription.realtime_transformation import VertexChirpRealtimeConfig
provider_config, model = await _vertex_provider_config_for(monkeypatch, "vertex_ai/chirp_3", None)
assert isinstance(provider_config, VertexChirpRealtimeConfig)
assert model == "chirp_3"
assert provider_config.get_complete_url(None, model) == "us-speech.googleapis.com"
assert provider_config.validate_environment({}, model, "https://us-speech.googleapis.com") == {}
@pytest.mark.asyncio
async def test_arealtime_routes_chirp_models_to_the_configured_speech_region(monkeypatch):
provider_config, model = await _vertex_provider_config_for(monkeypatch, "vertex_ai/chirp_3", "europe-west4")
assert provider_config.get_complete_url(None, model) == "europe-west4-speech.googleapis.com"
@pytest.mark.asyncio
async def test_arealtime_keeps_gemini_live_on_the_vertex_realtime_websocket(monkeypatch):
from litellm.llms.vertex_ai.realtime.transformation import VertexAIRealtimeConfig
provider_config, model = await _vertex_provider_config_for(monkeypatch, "vertex_ai/gemini-live-2.5-flash", None)
assert isinstance(provider_config, VertexAIRealtimeConfig)
assert provider_config.get_complete_url(None, model).startswith("wss://us-central1-aiplatform.googleapis.com/")
@pytest.mark.asyncio
async def test_realtime_health_check_names_the_batch_mode_for_chirp_models():
with pytest.raises(ValueError, match="mode audio_transcription"):
await realtime_main._realtime_health_check(model="chirp_3", custom_llm_provider="vertex_ai", api_key=None)
class _ClosableGaClientWebSocket:
def __init__(self) -> None:
self.scope: Final = {"headers": ()}
async def close(self, code: int = 1000, reason: str = "") -> None:
return None
@pytest.mark.asyncio
async def test_arealtime_openai_forwards_the_intent_query_param_to_the_upstream_url():
connect: Final = _ConnectThatStopsAfterCapturingTheUrl()
with patch("websockets.connect", connect):
await realtime_main._arealtime.__wrapped__(
model="openai/gpt-realtime",
websocket=_ClosableGaClientWebSocket(),
api_key="fake-key",
query_params={"model": "openai/gpt-realtime", "intent": "chat"},
litellm_logging_obj=FakeLogging(),
)
assert connect.url == "wss://api.openai.com/v1/realtime?model=gpt-realtime&intent=chat"
@pytest.mark.asyncio
@pytest.mark.parametrize("intent", (None, "transcription"))
async def test_arealtime_openai_connects_to_api_base_rather_than_the_callers_client_base_url(
intent: str | None,
) -> None:
connect: Final = _ConnectThatStopsAfterCapturingTheUrl()
query_params: Final[RealtimeQueryParams] = (
{"model": "openai/gateway-realtime", "intent": intent} if intent else {"model": "openai/gateway-realtime"}
)
with patch("websockets.connect", connect), patch("websockets.asyncio.client.connect", connect):
await realtime_main._arealtime.__wrapped__(
model="openai/gateway-realtime",
websocket=_ClosableGaClientWebSocket(),
api_key="fake-key",
api_base="https://configured-gateway.example/",
client=AsyncOpenAI(api_key="caller-key", base_url="https://caller-client.example/v1"),
query_params=query_params,
litellm_logging_obj=FakeLogging(),
)
expected_query: Final = "model=gateway-realtime&intent=transcription" if intent else "model=gateway-realtime"
assert connect.url == f"wss://configured-gateway.example/v1/realtime?{expected_query}"
@pytest.mark.parametrize(
("model", "api_base", "client_query_params", "expected_backend_url"),
[
(
"openai/gpt-live-transcribe",
"https://api.openai.com/",
{"model": "my-transcribe-alias", "intent": "transcription"},
"wss://api.openai.com/v1/realtime?intent=transcription",
),
(
"openai/gpt-live-transcribe",
"https://eu.api.openai.com/",
{"model": "my-transcribe-alias", "intent": "transcription"},
"wss://eu.api.openai.com/v1/realtime?intent=transcription",
),
(
"openai/gpt-live-transcribe",
"https://api.openai.com/",
{"model": "my-realtime-alias"},
"wss://api.openai.com/v1/realtime?model=gpt-live-transcribe",
),
(
"openai/gateway-transcribe-alias",
"https://gateway.example/",
{"model": "my-transcribe-alias", "intent": "transcription"},
"wss://gateway.example/v1/realtime?model=gateway-transcribe-alias&intent=transcription",
),
(
"xai/grok-voice",
"https://api.x.ai/v1",
{"model": "my-voice-alias", "intent": "transcription"},
"wss://api.x.ai/v1/realtime?model=grok-voice&intent=transcription",
),
],
)
@pytest.mark.asyncio
async def test_arealtime_drops_model_from_the_upstream_url_only_for_transcription_sessions_on_openai_hosts(
model, api_base, client_query_params, expected_backend_url
):
connect: Final = _ConnectThatStopsAfterCapturingTheUrl()
with patch("websockets.connect", connect):
await realtime_main._arealtime.__wrapped__(
model=model,
websocket=_ClosableGaClientWebSocket(),
api_base=api_base,
api_key="fake-key",
litellm_logging_obj=FakeLogging(),
query_params=client_query_params,
)
assert connect.url == expected_backend_url
async def _vertex_health_check_connect_for(monkeypatch: pytest.MonkeyPatch, api_base: str | None) -> _CapturingConnect:
async def fake_token_resolver(
credentials: object, project_id: str | None, custom_llm_provider: str
) -> tuple[str, str]:
return "access-token", project_id or ""
monkeypatch.setattr(realtime_main, "vertex_access_token_resolver", fake_token_resolver)
connect: Final = _CapturingConnect()
with patch("websockets.connect", connect):
assert await realtime_main._realtime_health_check(
model="gemini-3.8-live",
custom_llm_provider="vertex_ai",
api_key=None,
api_base=api_base,
model_params={
"vertex_project": "proj-1",
"vertex_credentials": "fake-credentials",
"vertex_location": "us",
},
)
return connect
@pytest.mark.asyncio
async def test_vertex_health_check_sends_no_tls_argument_to_a_plain_ws_api_base(
monkeypatch: pytest.MonkeyPatch,
) -> None:
connect: Final = await _vertex_health_check_connect_for(monkeypatch, "http://127.0.0.1:8080")
assert connect.url == "ws://127.0.0.1:8080/ws/google.cloud.aiplatform.v1.LlmBidiService/BidiGenerateContent"
assert connect.kwargs["ssl"] is None
@pytest.mark.asyncio
async def test_vertex_health_check_keeps_tls_for_the_multi_region_host(monkeypatch: pytest.MonkeyPatch) -> None:
connect: Final = await _vertex_health_check_connect_for(monkeypatch, None)
assert connect.url == (
"wss://aiplatform.us.rep.googleapis.com/ws/google.cloud.aiplatform.v1.LlmBidiService/BidiGenerateContent"
)
assert connect.kwargs["ssl"] is not None
BLOCKED_PHRASE = "XSECRETBLOCKTESTPHRASEX"
class PhraseBlockingGuardrail(CustomGuardrail):
async def apply_guardrail(self, inputs, request_data, input_type, logging_obj=None):
for text in inputs.get("texts", []):
if BLOCKED_PHRASE in text:
raise ValueError("Content blocked: contains forbidden test phrase.")
return inputs
def _make_guardrail(event_hook=GuardrailEventHooks.pre_call):
return PhraseBlockingGuardrail(
guardrail_name="integration-test-guard",
event_hook=event_hook,
default_on=True,
)
async def _build_streaming(client_events, backend_ws):
client_ws = MagicMock()
input_queue: asyncio.Queue = asyncio.Queue()
async def send_text(data: str):
client_events.append(json.loads(data))
client_ws.send_text = send_text
client_ws.receive_text = input_queue.get
logging_obj: Final = MagicMock(
pre_call=MagicMock(),
async_success_handler=AsyncMock(),
success_handler=MagicMock(),
dispatch_success_handlers=AsyncMock(),
dispatch_failure_handlers=AsyncMock(),
model_call_details={},
)
streaming = RealTimeStreaming(
websocket=client_ws,
backend_ws=backend_ws,
logging_obj=logging_obj,
request_data={"guardrails": ["integration-test-guard"]},
)
return streaming, input_queue
@pytest.mark.asyncio
async def test_voice_transcript_blocked_by_guardrail():
"""
Simulate a backend-side voice transcription event containing the blocked phrase.
Guardrail must block it - no response.create sent to OpenAI.
"""
from websockets.exceptions import ConnectionClosed
guardrail = _make_guardrail(GuardrailEventHooks.realtime_input_transcription)
litellm.callbacks = [guardrail]
client_events: list[dict] = []
# Build the transcript event that would come from the OpenAI backend
transcript_event = json.dumps(
{
"type": "conversation.item.input_audio_transcription.completed",
"transcript": f"This is {BLOCKED_PHRASE} in my voice message",
"item_id": "item_integ_test",
}
).encode()
# Mock backend that delivers the transcript then closes
backend_ws = MagicMock()
backend_ws.recv = AsyncMock(
side_effect=[
transcript_event,
ConnectionClosed(None, None),
]
)
backend_ws.send = AsyncMock()
try:
streaming, _ = await _build_streaming(client_events, backend_ws)
await streaming.backend_to_client_send_messages()
event_types = [e.get("type") for e in client_events]
# 1. Error event must be sent to client
error_events = [e for e in client_events if e.get("type") == "error"]
assert len(error_events) >= 1, f"Expected guardrail error event, got: {event_types}"
assert error_events[0]["error"]["type"] == "guardrail_violation"
# 2. Check what was sent to backend.
# The guardrail may send response.cancel + conversation.item.create (block msg)
# + response.create (to speak the block message). That's acceptable.
# What we assert is that a response.cancel was sent (blocking the original).
sent_to_backend = [
json.loads(c.args[0]) for c in backend_ws.send.call_args_list if c.args and isinstance(c.args[0], str)
]
response_cancels = [e for e in sent_to_backend if e.get("type") == "response.cancel"]
assert len(response_cancels) >= 1 or len(sent_to_backend) == 0, (
f"Guardrail should have sent response.cancel or nothing, got: {sent_to_backend}"
)
# Note: The guardrail may or may not send transcript deltas; the error event
# (assertion #1) is the primary signal that the blocked content was handled.
finally:
litellm.callbacks = []
@pytest.fixture
def federated_deployment(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> dict[str, str]:
token_file: Final = tmp_path / "subject_token.jwt"
token_file.write_text("subject-token")
for name in (
"OPENAI_API_KEY",
"OPENAI_BASE_URL",
"OPENAI_API_BASE",
"OPENAI_IDENTITY_PROVIDER_ID",
"OPENAI_SERVICE_ACCOUNT_ID",
"OPENAI_IDENTITY_TOKEN_FILE",
):
monkeypatch.delenv(name, raising=False)
for attribute in ("api_key", "openai_key", "api_base"):
monkeypatch.setattr(litellm, attribute, None)
_workload_identity_auth.cache_clear()
return {
"openai_identity_provider_id": "idp_realtime",
"openai_service_account_id": "user-realtime",
"openai_identity_token_file": str(token_file),
}
def _mock_token_exchange() -> None:
respx.post("https://auth.openai.com/oauth/token").mock(
return_value=httpx.Response(200, json={"access_token": "federated-bearer", "expires_in": 3600})
)
class _ConnectThatStopsAfterCapturingTheAuthorization:
authorization: str | None = None
def __call__(
self, url: str, *, additional_headers: Mapping[str, str], **kwargs: object
) -> "_ConnectThatStopsAfterCapturingTheAuthorization":
self.authorization = additional_headers.get("Authorization")
return self
async def __aenter__(self) -> None:
raise RuntimeError("backend headers captured, nothing to bridge")
async def __aexit__(
self,
exc_type: type[BaseException] | None,
exc: BaseException | None,
tb: TracebackType | None,
) -> None:
return None
@pytest.mark.asyncio
@pytest.mark.parametrize(
"live_session_start",
[None, {"type": "session.start", "session": {"model": "gpt-live-1"}}],
ids=["realtime", "live_session"],
)
@respx.mock
async def test_openai_websocket_sessions_send_the_federated_bearer_for_a_keyless_deployment(
federated_deployment: dict[str, str], live_session_start: Mapping[str, object] | None
) -> None:
_mock_token_exchange()
connect: Final = _ConnectThatStopsAfterCapturingTheAuthorization()
with patch("websockets.connect", connect):
await realtime_main._arealtime.__wrapped__(
model="gpt-realtime",
custom_llm_provider="openai",
websocket=MagicMock(close=AsyncMock()),
litellm_logging_obj=FakeLogging(),
live_session_start=live_session_start,
**federated_deployment,
)
assert connect.authorization == "Bearer federated-bearer"
class _GatewayThatRecordsTheAuthorization:
authorization: str | None = None
def __call__(self, request: httpx.Request) -> httpx.Response:
self.authorization = request.headers.get("Authorization")
return httpx.Response(200, json={"value": "ek_test"})
@pytest.mark.asyncio
@pytest.mark.parametrize(
("route", "model"),
[
(realtime_main.acreate_realtime_client_secret, "gpt-realtime"),
(realtime_main.acreate_realtime_transcription_session, "gpt-realtime"),
(realtime_main.acreate_realtime_translation_client_secret, "gpt-realtime-translate"),
],
ids=["client_secrets", "transcription_sessions", "translation_client_secrets"],
)
@respx.mock
async def test_openai_realtime_http_routes_send_the_federated_bearer_for_a_keyless_deployment(
federated_deployment: dict[str, str], route: Callable[..., Awaitable[object]], model: str
) -> None:
_mock_token_exchange()
gateway: Final = _GatewayThatRecordsTheAuthorization()
await route.__wrapped__(
model=model,
custom_llm_provider="openai",
litellm_logging_obj=Logging(
model=model,
messages=[],
stream=False,
call_type="acreate_realtime_client_secret",
start_time=datetime(2026, 1, 1, tzinfo=timezone.utc),
litellm_call_id="test-federated-realtime",
function_id="test-federated-realtime",
),
client=AsyncHTTPHandler(transport=httpx.MockTransport(gateway)),
**federated_deployment,
)
assert gateway.authorization == "Bearer federated-bearer"
@pytest.mark.asyncio
@pytest.mark.parametrize("flow", ("client_secret", "calls"))
async def test_translation_http_routes_preserve_explicit_azure_provider(flow: str) -> None:
expected_path: Final = (
"/openai/v1/realtime/translations/"
+ {
"client_secret": "client_secrets",
"calls": "calls",
}[flow]
)
def send_response(request: httpx.Request) -> httpx.Response:
assert request.url.host == "azure-deployment.example"
assert request.url.path == expected_path
return httpx.Response(200, json={"value": "ek-test", "routed_path": request.url.path})
params: Final[Mapping[str, object]] = {
"model": "translation-deployment",
"custom_llm_provider": "azure",
"api_base": "https://azure-deployment.example/openai/v1/",
"api_key": "test-key",
"client": AsyncHTTPHandler(transport=httpx.MockTransport(send_response)),
"litellm_logging_obj": Logging(
model="translation-deployment",
messages=[],
stream=False,
call_type="acreate_realtime_translation_client_secret",
start_time=datetime(2026, 1, 1, tzinfo=timezone.utc),
litellm_call_id="test-translation-routing",
function_id="test-translation-routing",
),
}
response: Final = await (
realtime_main.acreate_realtime_translation_client_secret.__wrapped__(**params)
if flow == "client_secret"
else realtime_main.arealtime_translation_calls.__wrapped__(
openai_ephemeral_key="ek-test", sdp_body=b"v=0", **params
)
)
assert response.status_code == 200
assert response.json() == {"value": "ek-test", "routed_path": expected_path}
@pytest.mark.asyncio
@respx.mock
async def test_openai_realtime_health_check_sends_the_federated_bearer_for_a_keyless_deployment(
federated_deployment: dict[str, str],
) -> None:
_mock_token_exchange()
connect: Final = _CapturingConnect()
with patch("websockets.connect", connect):
assert await realtime_main.realtime_health_check(
model="gpt-realtime",
custom_llm_provider="openai",
api_key=None,
model_params={"model": "openai/gpt-realtime", **federated_deployment},
)
assert connect.kwargs["additional_headers"] == {"Authorization": "Bearer federated-bearer"}
def _realtime_http_route_call(
route: str,
logging_obj: Logging,
custom_llm_provider: str,
api_base: str,
client: AsyncHTTPHandler | None = None,
metadata: Mapping[str, object] | None = None,
) -> Awaitable[httpx.Response]:
if route == "client_secrets":
return realtime_main.acreate_realtime_client_secret.__wrapped__(
model="gpt-realtime",
custom_llm_provider=custom_llm_provider,
api_key="sk-test",
api_base=api_base,
client=client,
litellm_logging_obj=logging_obj,
)
if route == "translation_client_secrets":
return realtime_main.acreate_realtime_translation_client_secret.__wrapped__(
model="gpt-realtime-translate",
custom_llm_provider=custom_llm_provider,
api_key="sk-test",
api_base=api_base,
client=client,
litellm_logging_obj=logging_obj,
metadata=metadata,
)
if route == "calls":
return realtime_main.arealtime_calls.__wrapped__(
openai_ephemeral_key="ek_test",
sdp_body=b"v=0",
model="gpt-realtime",
custom_llm_provider=custom_llm_provider,
api_base=api_base,
client=client,
litellm_logging_obj=logging_obj,
)
return realtime_main.arealtime_translation_calls.__wrapped__(
openai_ephemeral_key="ek_test",
sdp_body=b"v=0",
model="gpt-realtime-translate",
custom_llm_provider=custom_llm_provider,
api_base=api_base,
client=client,
litellm_logging_obj=logging_obj,
metadata=metadata,
)
def _realtime_http_logging(route: str) -> Logging:
return Logging(
model="gpt-realtime",
messages=[],
stream=False,
call_type="acreate_realtime_client_secret",
start_time=datetime(2026, 1, 1, tzinfo=timezone.utc),
litellm_call_id=f"test-realtime-{route}",
function_id=f"test-realtime-{route}",
)
_OPENAI_REALTIME_HTTP_ROUTE_PATHS: Final = {
"client_secrets": "/v1/realtime/client_secrets",
"translation_client_secrets": "/v1/realtime/translations/client_secrets",
"calls": "/v1/realtime/calls",
"translation_calls": "/v1/realtime/translations/calls",
}
@pytest.mark.asyncio
@pytest.mark.parametrize("route", tuple(_OPENAI_REALTIME_HTTP_ROUTE_PATHS))
async def test_openai_realtime_http_routes_surface_the_upstream_error_body_verbatim(route: str) -> None:
upstream_error: Final = {
"error": {
"message": "Invalid model",
"type": "invalid_request_error",
"param": "session.model",
"code": "invalid_value",
}
}
with pytest.raises(BaseLLMException) as raised:
await _realtime_http_route_call(
route,
_realtime_http_logging(route),
custom_llm_provider="openai",
api_base="https://upstream.example",
client=AsyncHTTPHandler(
transport=httpx.MockTransport(lambda request: httpx.Response(400, json=upstream_error))
),
)
assert raised.value.status_code == 400
assert json.loads(raised.value.message) == upstream_error
class _GatewayThatRecordsTheSession:
session: Mapping[str, object] | None = None
def __call__(self, request: httpx.Request) -> httpx.Response:
self.session = json.loads(request.content)["session"]
return httpx.Response(200, json={"value": "ek_test"})
@pytest.mark.asyncio
async def test_client_secret_keeps_the_realtime_session_model_when_input_transcription_names_another() -> None:
gateway: Final = _GatewayThatRecordsTheSession()
logging_obj: Final = _realtime_http_logging("client_secrets")
await realtime_main.acreate_realtime_client_secret.__wrapped__(
model="gpt-realtime",
session={
"type": "realtime",
"model": "gpt-realtime",
"audio": {"input": {"transcription": {"model": "gpt-realtime-whisper"}}},
},
custom_llm_provider="openai",
api_key="sk-test",
api_base="https://upstream.example",
litellm_logging_obj=logging_obj,
client=AsyncHTTPHandler(transport=httpx.MockTransport(gateway)),
)
assert gateway.session is not None
assert gateway.session["model"] == "gpt-realtime"
assert gateway.session["audio"] == {"input": {"transcription": {"model": "gpt-realtime-whisper"}}}
assert logging_obj.model == "gpt-realtime"
@pytest.mark.asyncio
async def test_client_secret_session_model_wins_over_a_provider_qualified_top_level_model(
local_model_cost_map: None,
) -> None:
gateway: Final = _GatewayThatRecordsTheSession()
logging_obj: Final = _realtime_http_logging("client_secrets")
await realtime_main.acreate_realtime_client_secret.__wrapped__(
model="openai/gpt-realtime-2.1",
session={"type": "realtime", "model": "gpt-realtime-2.1-mini"},
api_key="sk-test",
api_base="https://upstream.example",
litellm_logging_obj=logging_obj,
client=AsyncHTTPHandler(transport=httpx.MockTransport(gateway)),
)
assert gateway.session is not None
assert gateway.session["model"] == "gpt-realtime-2.1-mini"
assert logging_obj.model == "gpt-realtime-2.1-mini"
class _UpstreamThatMustNotBeCalled:
called: bool = False
def __call__(self, request: httpx.Request) -> httpx.Response:
self.called = True
return httpx.Response(200, json={"value": "ek_test"})
class _RecordingClientWebSocket:
def __init__(self) -> None:
self.scope: Final = {"headers": ()}
self.sent: tuple[str, ...] = ()
self.close_code: int | None = None
async def send_text(self, data: str) -> None:
self.sent = (*self.sent, data)
async def close(self, code: int = 1000, reason: str = "") -> None:
self.close_code = code
def _block_every_request_with_a_guardrail(
monkeypatch: pytest.MonkeyPatch, event_hook: GuardrailEventHooks = GuardrailEventHooks.pre_call
) -> None:
monkeypatch.setattr(
litellm,
"callbacks",
[CustomGuardrail(guardrail_name="translation-blocker", event_hook=event_hook, default_on=True)],
)
@pytest.mark.asyncio
@pytest.mark.parametrize("route", ("translation_client_secrets", "translation_calls"))
@pytest.mark.parametrize("event_hook", (GuardrailEventHooks.pre_call, GuardrailEventHooks.post_call))
async def test_translation_http_routes_refuse_a_guarded_request_before_any_upstream_call(
route: str,
monkeypatch: pytest.MonkeyPatch,
event_hook: GuardrailEventHooks,
) -> None:
_block_every_request_with_a_guardrail(monkeypatch, event_hook)
upstream: Final = _UpstreamThatMustNotBeCalled()
with pytest.raises(litellm.BadRequestError, match=TRANSLATION_GUARDRAILS_UNSUPPORTED_MESSAGE) as raised:
await _realtime_http_route_call(
route,
_realtime_http_logging(route),
custom_llm_provider="openai",
api_base="https://upstream.example",
client=AsyncHTTPHandler(transport=httpx.MockTransport(upstream)),
)
assert raised.value.status_code == 400
assert upstream.called is False
_DEFAULT_ON_GUARDRAIL_OPT_OUTS: Final = (
pytest.param({"user_api_key_metadata": {"disable_global_guardrails": True}}, id="key-disables-global"),
pytest.param(
{"user_api_key_team_metadata": {"opted_out_global_guardrails": ["translation-blocker"]}}, id="team-opts-out"
),
)
@pytest.mark.asyncio
@pytest.mark.parametrize(
("route", "path"),
(("translation_client_secrets", "client_secrets"), ("translation_calls", "calls")),
)
@pytest.mark.parametrize("key_metadata", _DEFAULT_ON_GUARDRAIL_OPT_OUTS)
@pytest.mark.parametrize("event_hook", (GuardrailEventHooks.pre_call, GuardrailEventHooks.post_call))
async def test_translation_http_routes_serve_a_key_opted_out_of_the_default_on_guardrail(
route: str,
path: str,
key_metadata: Mapping[str, object],
monkeypatch: pytest.MonkeyPatch,
event_hook: GuardrailEventHooks,
) -> None:
_block_every_request_with_a_guardrail(monkeypatch, event_hook)
gateway: Final = _GatewayThatRecordsTheUrl()
await _realtime_http_route_call(
route,
_realtime_http_logging(route),
custom_llm_provider="openai",
api_base="https://upstream.example",
client=AsyncHTTPHandler(transport=httpx.MockTransport(gateway)),
metadata=key_metadata,
)
assert gateway.url == f"https://upstream.example/v1/realtime/translations/{path}"
@pytest.mark.asyncio
@pytest.mark.parametrize("route", ("translation_client_secrets", "translation_calls"))
async def test_translation_http_routes_refuse_a_provider_other_than_openai_and_azure_before_any_upstream_call(
route: str,
) -> None:
upstream: Final = _UpstreamThatMustNotBeCalled()
with pytest.raises(litellm.BadRequestError, match="supports only the azure, openai providers, got xai") as raised:
await _realtime_http_route_call(
route,
_realtime_http_logging(route),
custom_llm_provider="xai",
api_base="https://upstream.example",
client=AsyncHTTPHandler(transport=httpx.MockTransport(upstream)),
)
assert raised.value.status_code == 400
assert upstream.called is False
@pytest.mark.asyncio
@pytest.mark.parametrize("event_hook", (GuardrailEventHooks.pre_call, GuardrailEventHooks.post_call))
async def test_translation_websocket_closes_1008_for_a_guarded_request_before_the_upstream_connect(
monkeypatch: pytest.MonkeyPatch,
event_hook: GuardrailEventHooks,
) -> None:
_block_every_request_with_a_guardrail(monkeypatch, event_hook)
websocket: Final = _RecordingClientWebSocket()
connect: Final = _CapturingConnect()
with patch("websockets.connect", connect):
await realtime_main._arealtime.__wrapped__(
model="openai/gpt-realtime-translate",
websocket=websocket,
api_key="fake-key",
realtime_mode="translation",
litellm_logging_obj=_realtime_websocket_logging(),
)
assert connect.url is None
assert websocket.close_code == 1008
assert [json.loads(event)["error"]["message"] for event in websocket.sent] == [
TRANSLATION_GUARDRAILS_UNSUPPORTED_MESSAGE
]
@pytest.mark.asyncio
@pytest.mark.parametrize("key_metadata", _DEFAULT_ON_GUARDRAIL_OPT_OUTS)
@pytest.mark.parametrize("event_hook", (GuardrailEventHooks.pre_call, GuardrailEventHooks.post_call))
async def test_translation_websocket_opens_for_a_key_opted_out_of_the_default_on_guardrail(
key_metadata: Mapping[str, object],
monkeypatch: pytest.MonkeyPatch,
event_hook: GuardrailEventHooks,
) -> None:
_block_every_request_with_a_guardrail(monkeypatch, event_hook)
websocket: Final = _RecordingClientWebSocket()
connect: Final = _CapturingConnect()
with patch("websockets.connect", connect):
await realtime_main._arealtime.__wrapped__(
model="openai/gpt-realtime-translate",
websocket=websocket,
api_key="fake-key",
realtime_mode="translation",
litellm_logging_obj=FakeLogging(),
metadata=key_metadata,
)
assert connect.url == "wss://api.openai.com/v1/realtime/translations?model=gpt-realtime-translate"
assert all(TRANSLATION_GUARDRAILS_UNSUPPORTED_MESSAGE not in event for event in websocket.sent)
class _FailureRecorder(CustomLogger):
def __init__(self) -> None:
super().__init__()
self.failures: tuple[object, ...] = ()
async def async_log_failure_event(
self, kwargs: Mapping[str, object], response_obj: object, start_time: object, end_time: object
) -> None:
self.failures = (*self.failures, kwargs.get("exception"))
@pytest.mark.asyncio
async def test_translation_websocket_refusal_logs_the_session_as_a_failure(monkeypatch: pytest.MonkeyPatch) -> None:
_block_every_request_with_a_guardrail(monkeypatch)
recorder: Final = _FailureRecorder()
logging_obj: Final = _realtime_websocket_logging(recorder)
with patch("websockets.connect", _CapturingConnect()):
await realtime_main._arealtime.__wrapped__(
model="openai/gpt-realtime-translate",
websocket=_RecordingClientWebSocket(),
api_key="fake-key",
realtime_mode="translation",
litellm_logging_obj=logging_obj,
)
assert [(type(failure), getattr(failure, "status_code", None)) for failure in recorder.failures] == [
(litellm.BadRequestError, 400)
]
assert TRANSLATION_GUARDRAILS_UNSUPPORTED_MESSAGE in str(recorder.failures[0])
assert logging_obj.model_call_details[REALTIME_SESSION_FAILURE_LOGGED_KEY] is True
class _NativeLifecycleGuardrail(CustomGuardrail):
use_native_lifecycle_hooks: ClassVar[bool] = True
@pytest.mark.asyncio
async def test_translation_websocket_opens_under_a_native_pre_call_guardrail(
monkeypatch: pytest.MonkeyPatch,
) -> None:
monkeypatch.setattr(
litellm,
"callbacks",
[
_NativeLifecycleGuardrail(
guardrail_name="runner-skipped", event_hook=GuardrailEventHooks.pre_call, default_on=True
)
],
)
websocket: Final = _RecordingClientWebSocket()
connect: Final = _CapturingConnect()
with patch("websockets.connect", connect):
await realtime_main._arealtime.__wrapped__(
model="openai/gpt-realtime-translate",
websocket=websocket,
api_key="fake-key",
realtime_mode="translation",
litellm_logging_obj=FakeLogging(),
)
assert connect.url == "wss://api.openai.com/v1/realtime/translations?model=gpt-realtime-translate"
assert all(TRANSLATION_GUARDRAILS_UNSUPPORTED_MESSAGE not in event for event in websocket.sent)
@pytest.mark.asyncio
@pytest.mark.parametrize("route", ("translation_client_secrets", "translation_calls", "websocket"))
@pytest.mark.parametrize(
("key_metadata", "blocked"),
(
pytest.param({"user_api_key_metadata": {"disable_global_guardrails": True}}, False, id="key-disables-global"),
pytest.param({}, True, id="no-opt-out"),
),
)
async def test_translation_lets_only_an_opted_out_key_past_a_legacy_string_callback_guardrail(
route: str,
key_metadata: Mapping[str, object],
blocked: bool,
monkeypatch: pytest.MonkeyPatch,
) -> None:
monkeypatch.setattr(litellm, "callbacks", [])
initialize_callbacks_on_proxy(
value=["lakera_prompt_injection"], premium_user=False, config_file_path="", litellm_settings={}
)
if route == "websocket":
websocket: Final = _RecordingClientWebSocket()
connect: Final = _CapturingConnect()
with patch("websockets.connect", connect):
await realtime_main._arealtime.__wrapped__(
model="openai/gpt-realtime-translate",
websocket=websocket,
api_key="fake-key",
realtime_mode="translation",
litellm_logging_obj=_realtime_websocket_logging(),
metadata=key_metadata,
)
if blocked:
assert connect.url is None
assert websocket.close_code == 1008
assert TRANSLATION_GUARDRAILS_UNSUPPORTED_MESSAGE in websocket.sent[0]
else:
assert connect.url == "wss://api.openai.com/v1/realtime/translations?model=gpt-realtime-translate"
assert all(TRANSLATION_GUARDRAILS_UNSUPPORTED_MESSAGE not in event for event in websocket.sent)
return
gateway: Final = _GatewayThatRecordsTheUrl()
call: Final = _realtime_http_route_call(
route,
_realtime_http_logging(route),
custom_llm_provider="openai",
api_base="https://upstream.example",
client=AsyncHTTPHandler(transport=httpx.MockTransport(gateway)),
metadata=key_metadata,
)
if blocked:
with pytest.raises(litellm.BadRequestError, match=TRANSLATION_GUARDRAILS_UNSUPPORTED_MESSAGE):
await call
assert gateway.url is None
else:
await call
assert gateway.url is not None
@pytest.mark.asyncio
async def test_translation_websocket_closes_1008_for_a_provider_other_than_openai_and_azure() -> None:
websocket: Final = _RecordingClientWebSocket()
connect: Final = _CapturingConnect()
with patch("websockets.connect", connect):
await realtime_main._arealtime.__wrapped__(
model="xai/grok-voice",
websocket=websocket,
api_key="fake-key",
realtime_mode="translation",
litellm_logging_obj=_realtime_websocket_logging(),
)
assert connect.url is None
assert websocket.close_code == 1008
assert [json.loads(event)["error"]["message"] for event in websocket.sent] == [
"Realtime translation supports only the azure, azure_ai, openai providers, got xai"
]
@pytest.mark.asyncio
@pytest.mark.parametrize("route", ("translation_client_secrets", "translation_calls", "websocket"))
@pytest.mark.parametrize("mode", ("pre_call", "post_call"))
@pytest.mark.parametrize("metadata_bucket", ("metadata", "litellm_metadata"))
@pytest.mark.parametrize("selected,condition_matches", ((True, True), (True, False), (False, True)))
async def test_translation_rejects_only_applicable_policy_pipelines(
route: str,
mode: Literal["pre_call", "post_call"],
metadata_bucket: str,
selected: bool,
condition_matches: bool,
monkeypatch: pytest.MonkeyPatch,
) -> None:
monkeypatch.setattr(litellm, "callbacks", [])
request_data: Final[dict[str, object]] = {"metadata": {}, "litellm_metadata": {}}
_apply_resolved_guardrails_to_metadata(
data=request_data,
metadata_variable_name=metadata_bucket,
context=PolicyMatchContext(model="translation-deployment"),
policy_names=["translation-policy"] if selected else [],
policies={
"translation-policy": Policy(
condition=PolicyCondition(model="translation-deployment" if condition_matches else "unrelated-model"),
pipeline=GuardrailPipeline(mode=mode, steps=[PipelineStep(guardrail="speech-policy", on_fail="block")]),
)
},
)
metadata: Final = realtime_main._build_litellm_metadata(request_data)
blocked: Final = selected and condition_matches
if route == "websocket":
websocket: Final = _RecordingClientWebSocket()
connect: Final = _CapturingConnect()
with patch("websockets.connect", connect):
await realtime_main._arealtime.__wrapped__(
model="openai/gpt-realtime-translate",
websocket=websocket,
api_key="fake-key",
realtime_mode="translation",
litellm_logging_obj=_realtime_websocket_logging(),
litellm_metadata=metadata,
)
if blocked:
assert connect.url is None
assert websocket.close_code == 1008
assert TRANSLATION_GUARDRAILS_UNSUPPORTED_MESSAGE in websocket.sent[0]
else:
assert connect.url == "wss://api.openai.com/v1/realtime/translations?model=gpt-realtime-translate"
return
upstream: Final = _UpstreamThatMustNotBeCalled()
call: Final = _realtime_http_route_call(
route,
_realtime_http_logging(route),
custom_llm_provider="openai",
api_base="https://upstream.example",
client=AsyncHTTPHandler(transport=httpx.MockTransport(upstream)),
metadata=metadata,
)
if blocked:
with pytest.raises(litellm.BadRequestError, match=TRANSLATION_GUARDRAILS_UNSUPPORTED_MESSAGE) as raised:
await call
assert raised.value.status_code == 400
assert upstream.called is False
else:
response: Final = await call
assert response.status_code == 200
assert upstream.called is True
@pytest.mark.asyncio
async def test_arealtime_routes_soniox_with_the_deployment_options(monkeypatch):
from litellm.llms.soniox.realtime.transformation import SonioxRealtimeConfig
captured: dict[str, object] = {}
async def mock_async_realtime(**kwargs):
captured.update(kwargs)
monkeypatch.setattr(realtime_main.base_llm_http_handler, "async_realtime", mock_async_realtime)
await realtime_main._arealtime.__wrapped__(
model="soniox/stt-rt-v5",
websocket=MagicMock(),
litellm_logging_obj=FakeLogging(),
query_params={"model": "soniox/stt-rt-v5", "intent": "transcription"},
api_key="deployment-key",
language_hints=["de"],
)
provider_config = captured["provider_config"]
assert isinstance(provider_config, SonioxRealtimeConfig)
assert captured["model"] == "stt-rt-v5"
assert captured["api_key"] == "deployment-key"
start_request = provider_config.transform_realtime_request(
json.dumps({"type": "input_audio_buffer.commit"}), "stt-rt-v5"
)[0]
assert json.loads(start_request)["language_hints"] == ["de"]
async def _serve_soniox(reply: str | None) -> tuple[object, list[object]]:
import websockets
received: list[object] = []
async def handler(connection) -> None:
received.append(connection.request.headers["Authorization"])
received.append(json.loads(await connection.recv()))
received.append(await connection.recv())
received.append(await connection.recv())
if reply is not None:
await connection.send(reply)
return await websockets.serve(handler, "127.0.0.1", 0), received
@pytest.mark.asyncio
async def test_soniox_realtime_health_check_starts_and_ends_a_stream_on_the_deployment_websocket():
server, received = await _serve_soniox(json.dumps({"tokens": [], "finished": True}))
port = next(iter(server.sockets)).getsockname()[1]
async with server:
healthy = await realtime_main.realtime_health_check(
model="stt-rt-v5",
custom_llm_provider="soniox",
api_key="soniox-key",
api_base=f"http://127.0.0.1:{port}",
model_params={"model": "soniox/stt-rt-v5", "language_hints": ["fa", "en"]},
)
assert healthy is True
assert received[0] == "Bearer soniox-key"
assert received[1]["model"] == "stt-rt-v5"
assert received[1]["language_hints"] == ["fa", "en"]
assert received[2] == bytes(len(received[2]))
assert len(received[2]) > 0
assert received[3] == ""
@pytest.mark.asyncio
async def test_soniox_realtime_health_check_fails_when_soniox_refuses_the_stream():
from litellm.llms.soniox.common_utils import SonioxException
server, _ = await _serve_soniox(json.dumps({"error_code": 401, "error_message": "Invalid API key."}))
port = next(iter(server.sockets)).getsockname()[1]
async with server:
with pytest.raises(SonioxException) as refused:
await realtime_main.realtime_health_check(
model="stt-rt-v5", custom_llm_provider="soniox", api_key="bad-key", api_base=f"ws://127.0.0.1:{port}"
)
assert refused.value.status_code == 401
@pytest.mark.asyncio
async def test_soniox_realtime_health_check_fails_when_soniox_closes_without_finishing_the_stream():
from litellm.llms.soniox.common_utils import SonioxException
server, _ = await _serve_soniox(None)
port = next(iter(server.sockets)).getsockname()[1]
async with server:
with pytest.raises(SonioxException):
await realtime_main.realtime_health_check(
model="stt-rt-v5", custom_llm_provider="soniox", api_key="soniox-key", api_base=f"ws://127.0.0.1:{port}"
)
Reference in a new issue View git blame Copy permalink