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"}