import os
from unittest.mock import AsyncMock, MagicMock, patch

import pytest

from litellm.llms.custom_httpx.http_handler import get_shared_realtime_ssl_context



@pytest.mark.asyncio
async def test_async_realtime_uses_max_size_parameter():
    """
    Test that Azure's async_realtime method uses the REALTIME_WEBSOCKET_MAX_MESSAGE_SIZE_BYTES
    constant for the max_size parameter to handle large base64 audio payloads.

    This verifies the fix for: https://github.com/BerriAI/litellm/issues/15747
    """
    from litellm.constants import REALTIME_WEBSOCKET_MAX_MESSAGE_SIZE_BYTES
    from litellm.llms.azure.realtime.handler import AzureOpenAIRealtime

    handler = AzureOpenAIRealtime()
    api_base = "https://my-endpoint.openai.azure.com"
    api_key = "test-key"
    api_version = "2024-10-01-preview"
    model = "gpt-4o-realtime-preview"

    dummy_websocket = AsyncMock()
    dummy_logging_obj = MagicMock()
    mock_backend_ws = AsyncMock()

    class DummyAsyncContextManager:
        def __init__(self, value):
            self.value = value

        async def __aenter__(self):
            return self.value

        async def __aexit__(self, exc_type, exc, tb):
            return None

    shared_context = get_shared_realtime_ssl_context()
    with (
        patch(
            "websockets.connect", return_value=DummyAsyncContextManager(mock_backend_ws)
        ) as mock_ws_connect,
        patch(
            "litellm.llms.azure.realtime.handler.RealTimeStreaming"
        ) as mock_realtime_streaming,
    ):

        mock_streaming_instance = MagicMock()
        mock_realtime_streaming.return_value = mock_streaming_instance
        mock_streaming_instance.bidirectional_forward = AsyncMock()

        await handler.async_realtime(
            model=model,
            websocket=dummy_websocket,
            logging_obj=dummy_logging_obj,
            api_base=api_base,
            api_key=api_key,
            api_version=api_version,
        )

        # Verify websockets.connect was called with the max_size parameter
        mock_ws_connect.assert_called_once()
        called_kwargs = mock_ws_connect.call_args[1]

        # Verify max_size is set (default None for unlimited, matching OpenAI's SDK)
        assert "max_size" in called_kwargs
        assert called_kwargs["max_size"] is None
        assert called_kwargs["ssl"] is shared_context
        # Default should be None (unlimited) to match OpenAI's official agents SDK
        # https://github.com/openai/openai-agents-python/blob/cf1b933660e44fd37b4350c41febab8221801409/src/agents/realtime/openai_realtime.py#L235

        mock_realtime_streaming.assert_called_once()
        mock_streaming_instance.bidirectional_forward.assert_awaited_once()


@pytest.mark.asyncio
async def test_construct_url_default_beta_protocol():
    """
    Test that _construct_url uses /openai/realtime (beta) by default.
    This maintains backwards compatibility.
    """
    from litellm.llms.azure.realtime.handler import AzureOpenAIRealtime

    handler = AzureOpenAIRealtime()
    url = handler._construct_url(
        api_base="https://my-endpoint.openai.azure.com",
        model="gpt-4o-realtime-preview",
        api_version="2024-10-01-preview",
    )

    assert url.startswith("wss://my-endpoint.openai.azure.com/openai/realtime?")
    assert "/openai/realtime?" in url
    assert "/openai/v1/realtime" not in url
    assert "api-version=2024-10-01-preview" in url
    assert "deployment=gpt-4o-realtime-preview" in url


@pytest.mark.asyncio
async def test_construct_url_beta_protocol_explicit():
    """
    Test that realtime_protocol='beta' explicitly uses /openai/realtime.
    """
    from litellm.llms.azure.realtime.handler import AzureOpenAIRealtime

    handler = AzureOpenAIRealtime()
    url = handler._construct_url(
        api_base="https://my-endpoint.openai.azure.com",
        model="gpt-4o-realtime-preview",
        api_version="2024-10-01-preview",
        realtime_protocol="beta",
    )

    assert "/openai/realtime?" in url
    assert "/openai/v1/realtime" not in url


@pytest.mark.asyncio
async def test_construct_url_ga_protocol():
    """
    Test that realtime_protocol='GA' uses /openai/v1/realtime (GA path).
    GA path uses ?model= instead of ?api-version=&deployment= format.
    """
    from litellm.llms.azure.realtime.handler import AzureOpenAIRealtime

    handler = AzureOpenAIRealtime()
    url = handler._construct_url(
        api_base="https://my-endpoint.openai.azure.com",
        model="gpt-4o-realtime-preview",
        api_version="2024-10-01-preview",
        realtime_protocol="GA",
    )

    assert url.startswith("wss://my-endpoint.openai.azure.com/openai/v1/realtime?")
    assert "/openai/v1/realtime?" in url
    # Ensure it doesn't have both paths
    assert url.count("/realtime") == 1
    # GA path uses model= query param, not api-version and deployment
    assert "model=gpt-4o-realtime-preview" in url
    assert "api-version" not in url
    assert "deployment" not in url


@pytest.mark.asyncio
async def test_construct_url_forwards_transcription_intent_ga():
    """
    Transcription sessions connect with intent=transcription. The Azure handler
    must forward that query param so gpt-realtime-whisper opens a transcription
    session instead of a normal realtime session.
    """
    from litellm.llms.azure.realtime.handler import AzureOpenAIRealtime

    handler = AzureOpenAIRealtime()
    url = handler._construct_url(
        api_base="https://my-endpoint.openai.azure.com",
        model="gpt-realtime-whisper",
        api_version="2025-04-01-preview",
        realtime_protocol="GA",
        query_params={"model": "gpt-realtime-whisper", "intent": "transcription"},
    )

    assert "/openai/v1/realtime?" in url
    assert "intent=transcription" in url
    assert "model=" not in url


@pytest.mark.asyncio
async def test_construct_url_forwards_transcription_intent_ga_without_model_query():
    """
    OpenAI-compatible transcription clients may connect with only
    intent=transcription and send the transcription model in session.update.
    Preserve that query shape instead of forcing model= into the upstream URL.
    """
    from litellm.llms.azure.realtime.handler import AzureOpenAIRealtime

    handler = AzureOpenAIRealtime()
    url = handler._construct_url(
        api_base="https://my-endpoint.openai.azure.com",
        model="gpt-realtime-whisper",
        api_version="2025-04-01-preview",
        realtime_protocol="GA",
        query_params={"intent": "transcription"},
    )

    assert url == (
        "wss://my-endpoint.openai.azure.com/openai/v1/realtime"
        "?intent=transcription"
    )


@pytest.mark.asyncio
async def test_construct_url_forwards_transcription_intent_beta():
    from litellm.llms.azure.realtime.handler import AzureOpenAIRealtime

    handler = AzureOpenAIRealtime()
    url = handler._construct_url(
        api_base="https://my-endpoint.openai.azure.com",
        model="whisper-deploy",
        api_version="2024-10-01-preview",
        query_params={"intent": "transcription"},
    )

    assert "/openai/realtime?" in url
    assert "deployment=whisper-deploy" in url
    assert "intent=transcription" in url


@pytest.mark.asyncio
async def test_construct_url_encodes_intent_value():
    """A crafted intent value must be URL-encoded, not injected as raw query params."""
    from litellm.llms.azure.realtime.handler import AzureOpenAIRealtime

    handler = AzureOpenAIRealtime()
    url = handler._construct_url(
        api_base="https://my-endpoint.openai.azure.com",
        model="gpt-realtime-whisper",
        api_version="2025-04-01-preview",
        realtime_protocol="GA",
        query_params={"intent": "transcription&foo=bar"},
    )
    assert "intent=transcription%26foo%3Dbar" in url
    assert "&foo=bar" not in url


@pytest.mark.asyncio
async def test_construct_url_no_intent_when_absent():
    """No intent param leaks into the URL when not provided."""
    from litellm.llms.azure.realtime.handler import AzureOpenAIRealtime

    handler = AzureOpenAIRealtime()
    url = handler._construct_url(
        api_base="https://my-endpoint.openai.azure.com",
        model="gpt-4o-realtime-preview",
        api_version="2024-10-01-preview",
        realtime_protocol="GA",
        query_params={"model": "gpt-4o-realtime-preview"},
    )
    assert "intent=" not in url


@pytest.mark.asyncio
async def test_construct_url_v1_protocol():
    """
    Test that realtime_protocol='v1' also uses /openai/v1/realtime.
    """
    from litellm.llms.azure.realtime.handler import AzureOpenAIRealtime

    handler = AzureOpenAIRealtime()
    url = handler._construct_url(
        api_base="https://my-endpoint.openai.azure.com",
        model="gpt-4o-realtime-preview",
        api_version="2024-10-01-preview",
        realtime_protocol="v1",
    )

    assert "/openai/v1/realtime?" in url
    assert url.count("/realtime") == 1


@pytest.mark.asyncio
@pytest.mark.parametrize("protocol", ["ga", "Ga", "gA", "V1", "v1", "GA"])
async def test_construct_url_case_insensitive_protocol(protocol):
    """
    Test that realtime_protocol matching is case-insensitive.
    """
    from litellm.llms.azure.realtime.handler import AzureOpenAIRealtime

    handler = AzureOpenAIRealtime()
    url = handler._construct_url(
        api_base="https://my-endpoint.openai.azure.com",
        model="gpt-realtime-deployment",
        api_version=None,
        realtime_protocol=protocol,
    )

    assert "/openai/v1/realtime?" in url
    assert "model=gpt-realtime-deployment" in url
    assert "api-version" not in url


@pytest.mark.asyncio
async def test_async_realtime_uses_ga_protocol_end_to_end():
    """
    Test that realtime_protocol='GA' flows through async_realtime to construct the correct URL.
    This is the end-to-end test ensuring the parameter is properly used.
    """
    from litellm.llms.azure.realtime.handler import AzureOpenAIRealtime

    handler = AzureOpenAIRealtime()
    api_base = "https://my-endpoint.openai.azure.com"
    api_key = "test-key"
    api_version = "2024-10-01-preview"
    model = "gpt-4o-realtime-preview"

    dummy_websocket = AsyncMock()
    dummy_logging_obj = MagicMock()
    mock_backend_ws = AsyncMock()

    class DummyAsyncContextManager:
        def __init__(self, value):
            self.value = value

        async def __aenter__(self):
            return self.value

        async def __aexit__(self, exc_type, exc, tb):
            return None

    with (
        patch(
            "websockets.connect", return_value=DummyAsyncContextManager(mock_backend_ws)
        ) as mock_ws_connect,
        patch(
            "litellm.llms.azure.realtime.handler.RealTimeStreaming"
        ) as mock_realtime_streaming,
    ):

        mock_streaming_instance = MagicMock()
        mock_realtime_streaming.return_value = mock_streaming_instance
        mock_streaming_instance.bidirectional_forward = AsyncMock()

        await handler.async_realtime(
            model=model,
            websocket=dummy_websocket,
            logging_obj=dummy_logging_obj,
            api_base=api_base,
            api_key=api_key,
            api_version=api_version,
            realtime_protocol="GA",
        )

        # Verify websockets.connect was called with GA URL
        mock_ws_connect.assert_called_once()
        called_url = mock_ws_connect.call_args[0][0]
        assert "/openai/v1/realtime" in called_url
        assert called_url.startswith("wss://")
        # GA path uses model= query param, not api-version and deployment
        assert "model=gpt-4o-realtime-preview" in called_url
        assert "api-version" not in called_url
        assert "deployment" not in called_url
        assert (
            mock_realtime_streaming.call_args.kwargs["backend_uses_beta_protocol"]
            is False
        )


@pytest.mark.asyncio
async def test_async_realtime_ga_without_api_version():
    """
    Test that GA/v1 protocol works without api_version (which is not needed for the GA path).
    Fixes #22127: api_version check was unconditional, blocking GA path.
    """
    from litellm.llms.azure.realtime.handler import AzureOpenAIRealtime

    handler = AzureOpenAIRealtime()
    api_base = "https://my-endpoint.openai.azure.com"
    api_key = "test-key"
    model = "gpt-realtime-deployment"

    dummy_websocket = AsyncMock()
    dummy_logging_obj = MagicMock()
    mock_backend_ws = AsyncMock()

    class DummyAsyncContextManager:
        def __init__(self, value):
            self.value = value

        async def __aenter__(self):
            return self.value

        async def __aexit__(self, exc_type, exc, tb):
            return None

    with (
        patch(
            "websockets.connect", return_value=DummyAsyncContextManager(mock_backend_ws)
        ) as mock_ws_connect,
        patch(
            "litellm.llms.azure.realtime.handler.RealTimeStreaming"
        ) as mock_realtime_streaming,
    ):

        mock_streaming_instance = MagicMock()
        mock_realtime_streaming.return_value = mock_streaming_instance
        mock_streaming_instance.bidirectional_forward = AsyncMock()

        # GA protocol with api_version=None should NOT raise ValueError
        await handler.async_realtime(
            model=model,
            websocket=dummy_websocket,
            logging_obj=dummy_logging_obj,
            api_base=api_base,
            api_key=api_key,
            api_version=None,
            realtime_protocol="GA",
        )

        called_url = mock_ws_connect.call_args[0][0]
        assert "/openai/v1/realtime?" in called_url
        assert "model=gpt-realtime-deployment" in called_url
        assert "api-version" not in called_url


@pytest.mark.asyncio
async def test_async_realtime_beta_without_api_version_raises():
    """
    Test that beta protocol still requires api_version.
    """
    from litellm.llms.azure.realtime.handler import AzureOpenAIRealtime

    handler = AzureOpenAIRealtime()
    dummy_websocket = AsyncMock()
    dummy_logging_obj = MagicMock()

    with pytest.raises(ValueError, match="api_version is required"):
        await handler.async_realtime(
            model="gpt-4o-realtime-preview",
            websocket=dummy_websocket,
            logging_obj=dummy_logging_obj,
            api_base="https://my-endpoint.openai.azure.com",
            api_key="test-key",
            api_version=None,
            realtime_protocol="beta",
        )


@pytest.mark.asyncio
async def test_realtime_protocol_env_var_fallback():
    """
    Test that LITELLM_AZURE_REALTIME_PROTOCOL env var is used as fallback.
    Fixes #22127: no way to set realtime_protocol from config.
    """
    from litellm.realtime_api.main import _arealtime
    from litellm.types.router import GenericLiteLLMParams

    with patch.dict(os.environ, {"LITELLM_AZURE_REALTIME_PROTOCOL": "v1"}):
        # Create a GenericLiteLLMParams without realtime_protocol
        litellm_params = GenericLiteLLMParams()
        # The env var should be picked up as fallback
        realtime_protocol = (
            {}.get("realtime_protocol")
            or litellm_params.get("realtime_protocol")
            or os.environ.get("LITELLM_AZURE_REALTIME_PROTOCOL")
            or "beta"
        )
        assert realtime_protocol == "v1"


@pytest.mark.asyncio
async def test_realtime_protocol_from_litellm_params():
    """
    Test that realtime_protocol is read from litellm_params (config.yaml extra field).
    Fixes #22127: realtime_protocol in litellm_params was not used.
    """
    from litellm.types.router import GenericLiteLLMParams

    # Simulate config.yaml with realtime_protocol as an extra field
    litellm_params = GenericLiteLLMParams(realtime_protocol="GA")
    assert litellm_params.get("realtime_protocol") == "GA"


@pytest.mark.asyncio
async def test_arealtime_transcription_intent_defaults_to_ga(monkeypatch):
    """
    Azure gpt-realtime-whisper transcription connects on the GA /openai/v1/realtime
    path. If the DB model lacks realtime_protocol, infer GA from intent=transcription.
    """
    from litellm.realtime_api import main as realtime_main

    mock_async_realtime = AsyncMock()
    monkeypatch.setattr(
        realtime_main,
        "azure_realtime",
        MagicMock(async_realtime=mock_async_realtime),
    )

    def fake_get_llm_provider(model, api_base=None, api_key=None):
        return (
            "gpt-realtime-whisper",
            "azure",
            "test-key",
            "https://my-endpoint.openai.azure.com",
        )

    monkeypatch.setattr(realtime_main, "get_llm_provider", fake_get_llm_provider)

    await realtime_main._arealtime(
        model="azure/gpt-realtime-whisper",
        websocket=MagicMock(),
        api_key="test-key",
        api_version="2025-04-01-preview",
        query_params={"intent": "transcription"},
        litellm_logging_obj=MagicMock(),
    )

    called_kwargs = mock_async_realtime.call_args.kwargs
    assert called_kwargs["realtime_protocol"] == "GA"
    assert called_kwargs["query_params"] == {"intent": "transcription"}


@pytest.mark.asyncio
async def test_async_realtime_default_maintains_backwards_compatibility():
    """
    Test that not passing realtime_protocol maintains the original beta behavior.
    This ensures backwards compatibility for existing deployments.
    """
    from litellm.llms.azure.realtime.handler import AzureOpenAIRealtime

    handler = AzureOpenAIRealtime()
    api_base = "https://my-endpoint.openai.azure.com"
    api_key = "test-key"
    api_version = "2024-10-01-preview"
    model = "gpt-4o-realtime-preview"

    dummy_websocket = AsyncMock()
    dummy_logging_obj = MagicMock()
    mock_backend_ws = AsyncMock()

    class DummyAsyncContextManager:
        def __init__(self, value):
            self.value = value

        async def __aenter__(self):
            return self.value

        async def __aexit__(self, exc_type, exc, tb):
            return None

    with (
        patch(
            "websockets.connect", return_value=DummyAsyncContextManager(mock_backend_ws)
        ) as mock_ws_connect,
        patch(
            "litellm.llms.azure.realtime.handler.RealTimeStreaming"
        ) as mock_realtime_streaming,
    ):

        mock_streaming_instance = MagicMock()
        mock_realtime_streaming.return_value = mock_streaming_instance
        mock_streaming_instance.bidirectional_forward = AsyncMock()

        # Call without realtime_protocol parameter
        await handler.async_realtime(
            model=model,
            websocket=dummy_websocket,
            logging_obj=dummy_logging_obj,
            api_base=api_base,
            api_key=api_key,
            api_version=api_version,
        )

        # Verify it still uses the beta path
        called_url = mock_ws_connect.call_args[0][0]
        assert "/openai/realtime?" in called_url
        assert "/openai/v1/realtime" not in called_url
        assert (
            mock_realtime_streaming.call_args.kwargs["backend_uses_beta_protocol"]
            is True
        )


class _DummyAsyncContextManager:
    def __init__(self, value):
        self.value = value

    async def __aenter__(self):
        return self.value

    async def __aexit__(self, exc_type, exc, tb):
        return None


@pytest.mark.asyncio
async def test_async_realtime_uses_bearer_token_when_no_api_key():
    """
    Entra ID-only Azure realtime deployments have no static api-key, so the handshake must
    authenticate with `Authorization: Bearer <azure_ad_token>` and must not send `api-key`.

    Regression test for https://github.com/BerriAI/litellm/issues/34654
    """
    from litellm.llms.azure.realtime.handler import AzureOpenAIRealtime

    handler = AzureOpenAIRealtime()
    mock_backend_ws = AsyncMock()

    with (
        patch(
            "websockets.connect",
            return_value=_DummyAsyncContextManager(mock_backend_ws),
        ) as mock_ws_connect,
        patch(  # test-quality-ok: handler owns the streaming loop, only the handshake headers are under test
            "litellm.llms.azure.realtime.handler.RealTimeStreaming"
        ) as mock_realtime_streaming,
    ):
        mock_realtime_streaming.return_value.bidirectional_forward = AsyncMock()

        await handler.async_realtime(
            model="gpt-realtime-whisper",
            websocket=AsyncMock(),
            logging_obj=MagicMock(),
            api_base="https://my-endpoint.openai.azure.com",
            api_key=None,
            api_version="2024-10-01-preview",
            azure_ad_token="my-entra-token",
        )

    headers = mock_ws_connect.call_args.kwargs["additional_headers"]
    assert headers == {"Authorization": "Bearer my-entra-token"}


def test_get_auth_headers_prefers_api_key_and_never_sends_both():
    from litellm.llms.azure.realtime.handler import AzureOpenAIRealtime

    assert AzureOpenAIRealtime.get_auth_headers(api_key="test-key", azure_ad_token="my-entra-token") == {
        "api-key": "test-key"
    }


def test_get_auth_headers_without_credentials_raises():
    from litellm.llms.azure.realtime.handler import AzureOpenAIRealtime

    with pytest.raises(ValueError, match="Missing Azure credentials"):
        AzureOpenAIRealtime.get_auth_headers(api_key=None, azure_ad_token=None)


@pytest.mark.asyncio
async def test_arealtime_resolves_azure_ad_token_when_no_api_key(monkeypatch):
    """
    `_arealtime` must resolve an Azure AD token (managed identity, service principal, etc.)
    and forward it to the handler when the deployment has no api_key.

    Regression test for https://github.com/BerriAI/litellm/issues/34654
    """
    from litellm.realtime_api import main as realtime_main

    mock_async_realtime = AsyncMock()
    monkeypatch.setattr(realtime_main, "azure_realtime", MagicMock(async_realtime=mock_async_realtime))
    monkeypatch.setattr(
        realtime_main,
        "get_llm_provider",
        lambda model, api_base=None, api_key=None: (
            "gpt-realtime-whisper",
            "azure",
            None,
            "https://my-endpoint.openai.azure.com",
        ),
    )
    monkeypatch.delenv("AZURE_API_KEY", raising=False)

    captured_params = {}

    def fake_get_azure_ad_token(litellm_params):
        captured_params["tenant_id"] = litellm_params.get("tenant_id")
        return "my-entra-token"

    monkeypatch.setattr(realtime_main, "get_azure_ad_token", fake_get_azure_ad_token)

    await realtime_main._arealtime(
        model="azure/gpt-realtime-whisper",
        websocket=MagicMock(),
        api_version="2024-10-01-preview",
        litellm_logging_obj=MagicMock(),
        tenant_id="my-tenant",
        client_id="my-client",
        client_secret="my-secret",
    )

    assert mock_async_realtime.call_args.kwargs["azure_ad_token"] == "my-entra-token"
    assert captured_params["tenant_id"] == "my-tenant"


@pytest.mark.asyncio
async def test_arealtime_does_not_resolve_azure_ad_token_when_api_key_present(monkeypatch):
    from litellm.realtime_api import main as realtime_main

    mock_async_realtime = AsyncMock()
    monkeypatch.setattr(realtime_main, "azure_realtime", MagicMock(async_realtime=mock_async_realtime))
    monkeypatch.setattr(
        realtime_main,
        "get_llm_provider",
        lambda model, api_base=None, api_key=None: (
            "gpt-realtime-whisper",
            "azure",
            "test-key",
            "https://my-endpoint.openai.azure.com",
        ),
    )

    def fail_get_azure_ad_token(litellm_params):
        raise AssertionError("should not resolve an AD token when an api_key is configured")

    monkeypatch.setattr(realtime_main, "get_azure_ad_token", fail_get_azure_ad_token)

    await realtime_main._arealtime(
        model="azure/gpt-realtime-whisper",
        websocket=MagicMock(),
        api_key="test-key",
        api_version="2024-10-01-preview",
        litellm_logging_obj=MagicMock(),
    )

    assert mock_async_realtime.call_args.kwargs["azure_ad_token"] is None


@pytest.mark.asyncio
async def test_realtime_health_check_uses_bearer_token_when_no_api_key(monkeypatch):
    """
    An Entra ID-only realtime deployment must also pass its realtime health check.

    Regression test for https://github.com/BerriAI/litellm/issues/34654
    """
    from litellm.realtime_api import main as realtime_main

    connect_calls = []

    monkeypatch.setattr(
        realtime_main,
        "get_azure_ad_token",
        lambda litellm_params: "my-entra-token",
    )

    def fake_connect(url, **kwargs):
        connect_calls.append(kwargs)
        return _DummyAsyncContextManager(MagicMock())

    monkeypatch.setattr("websockets.connect", fake_connect)

    assert (
        await realtime_main._realtime_health_check(
            model="gpt-realtime-whisper",
            custom_llm_provider="azure",
            api_key=None,
            api_base="https://my-endpoint.openai.azure.com",
            api_version="2024-10-01-preview",
            model_params={"tenant_id": "my-tenant"},
        )
        is True
    )
    assert connect_calls[0]["additional_headers"] == {"Authorization": "Bearer my-entra-token"}


@pytest.mark.asyncio
async def test_arealtime_forwards_deployment_azure_ad_token(monkeypatch):
    """
    The router binds a deployment's `azure_ad_token` to `_arealtime`'s named parameter rather than
    **kwargs, so it must still reach the handler.

    Regression test for https://github.com/BerriAI/litellm/issues/34654
    """
    from litellm.realtime_api import main as realtime_main

    mock_async_realtime = AsyncMock()
    monkeypatch.setattr(realtime_main, "azure_realtime", MagicMock(async_realtime=mock_async_realtime))
    monkeypatch.setattr(
        realtime_main,
        "get_llm_provider",
        lambda model, api_base=None, api_key=None: (
            "gpt-realtime-whisper",
            "azure",
            None,
            "https://my-endpoint.openai.azure.com",
        ),
    )
    monkeypatch.delenv("AZURE_API_KEY", raising=False)
    monkeypatch.setattr(realtime_main.litellm, "api_key", None)

    await realtime_main._arealtime(
        model="azure/gpt-realtime-whisper",
        websocket=MagicMock(),
        api_version="2024-10-01-preview",
        azure_ad_token="deployment-entra-token",
        litellm_logging_obj=MagicMock(),
    )

    assert mock_async_realtime.call_args.kwargs["azure_ad_token"] == "deployment-entra-token"
