Improve test coverage: detection from backend, error paths, unknown usage type, resolved_model None

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()

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

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)

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(