mirror of
https://github.com/BerriAI/litellm.git
synced 2026-10-29 06:33:28 +00:00
86 lines
3.6 KiB
Python
86 lines
3.6 KiB
Python
from collections.abc import Mapping
|
|
from typing import Any, Final, cast
|
|
|
|
import pytest
|
|
from pydantic import TypeAdapter
|
|
|
|
from litellm.proxy.common_utils.realtime_utils import realtime_request_body
|
|
from litellm.proxy.proxy_server import _realtime_query_params_template
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def clear_realtime_caches():
|
|
realtime_request_body.cache_clear()
|
|
_realtime_query_params_template.cache_clear()
|
|
yield
|
|
realtime_request_body.cache_clear()
|
|
_realtime_query_params_template.cache_clear()
|
|
|
|
|
|
def test_realtime_request_body_returns_immutable_bytes():
|
|
cached_body = realtime_request_body("gpt-4o")
|
|
|
|
with pytest.raises(TypeError):
|
|
cast(Any, cached_body)[0] = ord("x")
|
|
|
|
|
|
def test_realtime_query_params_template_returns_immutable_tuples():
|
|
cached_tuple = _realtime_query_params_template("gpt-4o", "intent-a")
|
|
|
|
with pytest.raises(TypeError):
|
|
cast(Any, cached_tuple)[0] = ("model", "mutated")
|
|
|
|
|
|
def test_realtime_request_body_caches_each_model_separately():
|
|
gpt4o_body_first = realtime_request_body("gpt-4o")
|
|
gpt4o_body_second = realtime_request_body("gpt-4o")
|
|
gpt4o_mini_body = realtime_request_body("gpt-4o-mini")
|
|
|
|
assert gpt4o_body_first is gpt4o_body_second
|
|
assert gpt4o_body_first == b'{"model": "gpt-4o"}'
|
|
assert gpt4o_mini_body == b'{"model": "gpt-4o-mini"}'
|
|
assert gpt4o_body_first is not gpt4o_mini_body
|
|
|
|
|
|
@pytest.mark.parametrize("is_translation", (False, True))
|
|
def test_realtime_request_body_preserves_default_model_authorization(is_translation: bool) -> None:
|
|
from litellm.proxy.agent_endpoints.auth.managed_authorization import managed_inference_request
|
|
from litellm.proxy.common_utils.http_parsing_utils import resolve_realtime_route_model
|
|
|
|
body: Final = TypeAdapter(Mapping[str, object]).validate_json(realtime_request_body(None))
|
|
route: Final = "/v1/realtime/translations" if is_translation else "/v1/realtime"
|
|
intent: Final = None if is_translation else "transcription"
|
|
admitted: Final = managed_inference_request(route, body, {}, None, intent=intent)
|
|
|
|
assert admitted["model"] == resolve_realtime_route_model(None, intent, is_translation)
|
|
assert isinstance(admitted["model"], str) and admitted["model"]
|
|
|
|
|
|
@pytest.mark.parametrize("model", (None, "requested", 'model"with\\quotes', "model\nwith-newline"))
|
|
def test_realtime_request_body_preserves_model_names(model: str | None) -> None:
|
|
body: Final = TypeAdapter(Mapping[str, object]).validate_json(realtime_request_body(model))
|
|
|
|
assert body["model"] == model
|
|
|
|
|
|
def test_realtime_query_params_template_caches_each_pair_separately():
|
|
params_with_intent_first = _realtime_query_params_template("gpt-4o", "intent-a")
|
|
params_with_intent_second = _realtime_query_params_template("gpt-4o", "intent-a")
|
|
params_without_intent = _realtime_query_params_template("gpt-4o", None)
|
|
params_transcription_without_model = _realtime_query_params_template(None, "transcription")
|
|
|
|
assert params_with_intent_first is params_with_intent_second
|
|
assert params_with_intent_first == (("model", "gpt-4o"), ("intent", "intent-a"))
|
|
assert params_without_intent == (("model", "gpt-4o"),)
|
|
assert params_transcription_without_model == (("intent", "transcription"),)
|
|
assert params_with_intent_first is not params_without_intent
|
|
|
|
|
|
def test_realtime_query_params_dict_copies_do_not_leak_state():
|
|
params_dict_one = dict(_realtime_query_params_template("gpt-4o", "intent-a"))
|
|
params_dict_one["new"] = "value"
|
|
|
|
params_dict_two = dict(_realtime_query_params_template("gpt-4o", "intent-a"))
|
|
|
|
assert "new" not in params_dict_two
|
|
assert params_dict_two == {"model": "gpt-4o", "intent": "intent-a"}
|