diff --git a/tests/test_litellm/litellm_core_utils/test_realtime_streaming.py b/tests/test_litellm/litellm_core_utils/test_realtime_streaming.py index 870b0051f78..10bfd473b41 100644 --- a/tests/test_litellm/litellm_core_utils/test_realtime_streaming.py +++ b/tests/test_litellm/litellm_core_utils/test_realtime_streaming.py @@ -632,6 +632,62 @@ def test_client_session_update_marks_transcription_session(): assert streaming._is_transcription_session is True +def test_detect_transcription_session_from_backend_transcription_session_events(): + """Backend transcription_session.created/updated events flag the session.""" + streaming = RealTimeStreaming(MagicMock(), MagicMock(), MagicMock()) + assert streaming._is_transcription_session is False + streaming._detect_transcription_session_from_backend( + {"type": "transcription_session.created"} + ) + assert streaming._is_transcription_session is True + + streaming2 = RealTimeStreaming(MagicMock(), MagicMock(), MagicMock()) + streaming2._detect_transcription_session_from_backend( + {"type": "transcription_session.updated"} + ) + assert streaming2._is_transcription_session is True + + +def test_detect_transcription_session_from_backend_session_created_with_type(): + """Backend session.created with type=transcription flags the session.""" + streaming = RealTimeStreaming(MagicMock(), MagicMock(), MagicMock()) + streaming._detect_transcription_session_from_backend( + {"type": "session.created", "session": {"type": "transcription"}} + ) + assert streaming._is_transcription_session is True + + +def test_detect_transcription_session_from_backend_ignores_non_transcription(): + """Backend session.created without type=transcription does not flag the session.""" + streaming = RealTimeStreaming(MagicMock(), MagicMock(), MagicMock()) + streaming._detect_transcription_session_from_backend( + {"type": "session.created", "session": {"model": "gpt-4o-realtime-preview"}} + ) + assert streaming._is_transcription_session is False + + +def test_capture_transcription_usage_deduplicates_when_already_stored(): + """ + When the event is already in messages (logged via store_message), it must not + be appended a second time by _capture_transcription_usage. + """ + import litellm + + streaming = RealTimeStreaming(MagicMock(), MagicMock(), MagicMock()) + # Add the event type to the default logged list so _should_store_message returns True. + streaming.logged_real_time_event_types = [ + "conversation.item.input_audio_transcription.completed" + ] + event = { + "type": "conversation.item.input_audio_transcription.completed", + "usage": {"type": "duration", "seconds": 5.0}, + } + streaming.store_message(json.dumps(event)) + initial_count = len(streaming.messages) + streaming._capture_transcription_usage(event) + assert len(streaming.messages) == initial_count # no duplicate + + @pytest.mark.asyncio async def test_client_ack_caches_setup_to_prevent_duplicate_session_update_setup(): websocket = MagicMock() diff --git a/tests/test_litellm/llms/openai/realtime/test_transcription_sessions.py b/tests/test_litellm/llms/openai/realtime/test_transcription_sessions.py index 08ef9485677..62fc3a8d0aa 100644 --- a/tests/test_litellm/llms/openai/realtime/test_transcription_sessions.py +++ b/tests/test_litellm/llms/openai/realtime/test_transcription_sessions.py @@ -55,7 +55,10 @@ def test_azure_transcription_session_url_uses_deployment_and_api_version(): ) -def test_request_resolves_model_from_top_level_hint(): +def test_request_resolves_model_returns_none_when_both_absent(): + req = RealtimeTranscriptionSessionRequest(input_audio_format="pcm16") + assert req.resolved_model() is None + req = RealtimeTranscriptionSessionRequest( model="openai/gpt-realtime-whisper", input_audio_transcription={"model": "gpt-realtime-whisper"}, diff --git a/tests/test_litellm/proxy/realtime_endpoints/test_realtime_webrtc_endpoints.py b/tests/test_litellm/proxy/realtime_endpoints/test_realtime_webrtc_endpoints.py index ec541903dd0..362307252ee 100644 --- a/tests/test_litellm/proxy/realtime_endpoints/test_realtime_webrtc_endpoints.py +++ b/tests/test_litellm/proxy/realtime_endpoints/test_realtime_webrtc_endpoints.py @@ -499,3 +499,113 @@ async def test_transcription_sessions_encrypts_client_secret( ) finally: proxy_app.dependency_overrides.pop(user_api_key_auth, None) + + +def test_session_type_coerced_for_unknown_value(): + """An unrecognized session_type in the token falls back to 'realtime'.""" + payload = _encode_realtime_token_payload( + ephemeral_key="epk", + model_id="gpt-4o", + user_id=None, + team_id=None, + expires_at=None, + session_type="INJECTED_TYPE", + ) + # Force-deserialize and check the coercion that happens in proxy_realtime_calls. + decoded = json.loads(payload) + session_type = decoded.get("session_type") or "realtime" + if session_type not in ("realtime", "transcription"): + session_type = "realtime" + assert session_type == "realtime" + + +@pytest.mark.asyncio +async def test_transcription_sessions_returns_upstream_error_verbatim( + proxy_app, + mock_add_litellm_data, + mock_pre_call_hook, +): + """Non-200 upstream response is forwarded unchanged (no encryption attempted).""" + mock_resp = MagicMock(spec=httpx.Response) + mock_resp.status_code = 400 + mock_resp.content = b'{"error":"bad_request"}' + mock_resp.headers = {} + mock_resp.json.return_value = {"error": "bad_request"} + mock_resp.text = '{"error":"bad_request"}' + + async def _mock_route(*args, **kwargs): + async def _inner(): + return mock_resp + + return _inner() + + proxy_app.dependency_overrides[user_api_key_auth] = lambda: UserAPIKeyAuth( + user_id="test-user", team_id="test-team" + ) + try: + client = TestClient(proxy_app) + with ( + patch( + "litellm.proxy.proxy_server.route_request", + side_effect=_mock_route, + ), + patch( + "litellm.proxy.proxy_server.add_litellm_data_to_request", + side_effect=mock_add_litellm_data, + ), + patch("litellm.proxy.proxy_server.proxy_logging_obj") as mock_logging, + ): + mock_logging.pre_call_hook = AsyncMock(side_effect=mock_pre_call_hook) + mock_logging.post_call_failure_hook = AsyncMock() + + response = client.post( + "/v1/realtime/transcription_sessions", + headers={"Authorization": "Bearer sk-test-master-key"}, + json={"input_audio_transcription": {"model": "gpt-realtime-whisper"}}, + ) + assert response.status_code == 400 + assert response.content == b'{"error":"bad_request"}' + finally: + proxy_app.dependency_overrides.pop(user_api_key_auth, None) + + +@pytest.mark.asyncio +async def test_transcription_sessions_wraps_route_exception( + proxy_app, + mock_add_litellm_data, + mock_pre_call_hook, +): + """A route exception is wrapped in a ProxyException with a human-readable message.""" + from fastapi import HTTPException + + async def _raise_http(*args, **kwargs): + raise HTTPException(status_code=403, detail="Model not allowed") + + proxy_app.dependency_overrides[user_api_key_auth] = lambda: UserAPIKeyAuth( + user_id="test-user" + ) + try: + client = TestClient(proxy_app, raise_server_exceptions=False) + with ( + patch( + "litellm.proxy.proxy_server.route_request", + side_effect=_raise_http, + ), + patch( + "litellm.proxy.proxy_server.add_litellm_data_to_request", + side_effect=mock_add_litellm_data, + ), + patch("litellm.proxy.proxy_server.proxy_logging_obj") as mock_logging, + ): + mock_logging.pre_call_hook = AsyncMock(side_effect=mock_pre_call_hook) + mock_logging.post_call_failure_hook = AsyncMock() + + response = client.post( + "/v1/realtime/transcription_sessions", + headers={"Authorization": "Bearer sk-test-master-key"}, + json={"input_audio_transcription": {"model": "gpt-realtime-whisper"}}, + ) + assert response.status_code == 403 + assert "Model not allowed" in response.text + finally: + proxy_app.dependency_overrides.pop(user_api_key_auth, None) diff --git a/tests/test_litellm/test_cost_calculator.py b/tests/test_litellm/test_cost_calculator.py index a9cdaf27197..1590a3f5935 100644 --- a/tests/test_litellm/test_cost_calculator.py +++ b/tests/test_litellm/test_cost_calculator.py @@ -421,6 +421,8 @@ def test_realtime_logging_object_allows_null_transcript_in_conversation_item_add assert logging_result.usage.total_tokens == 18 assert logging_result.results[0]["item"]["content"][0]["transcript"] is None assert logging_result.results[0]["item"]["content"][0]["transcript"] is None + + def test_realtime_transcription_duration_cost(monkeypatch): """ gpt-realtime-whisper transcription sessions are billed by input audio duration @@ -544,7 +546,26 @@ def test_realtime_transcription_token_billed_fallback(monkeypatch): assert abs(cost - expected) < 1e-12 -def test_custom_pricing_with_router_model_id(): +def test_transcription_usage_cost_returns_zero_for_unknown_type(): + """An unrecognized usage type yields 0 (safe fallback, no exception).""" + from litellm.cost_calculator import _transcription_usage_cost + + assert _transcription_usage_cost({"type": "future_billing_type"}, {}) == 0.0 + assert _transcription_usage_cost({}, {}) == 0.0 + + +def test_get_transcription_model_falls_back_to_session_model(monkeypatch): + """session.model is used when transcription-specific model fields are absent.""" + monkeypatch.setenv("LITELLM_LOCAL_MODEL_COST_MAP", "True") + monkeypatch.setattr(litellm, "model_cost", litellm.get_model_cost_map(url="")) + + from litellm.cost_calculator import _get_transcription_model_name_from_results + + results: OpenAIRealtimeStreamList = [ + {"type": "session.created", "session": {"model": "gpt-realtime-whisper"}}, + ] + assert _get_transcription_model_name_from_results(results) == "gpt-realtime-whisper" + from litellm import Router router = Router(