""" Unit tests for ResponsePollingHandler Tests core functionality including: 1. Polling ID generation and detection 2. Initial state creation (queued status) 3. State updates with batched output 4. Status transitions (queued -> in_progress -> completed) 5. Response completion with reasoning, tools, tool_choice 6. Error handling and cancellation 7. Cache key generation These tests ensure the polling handler correctly manages response state following the OpenAI Response API format. """ import json import os import sys from datetime import datetime, timezone from typing import Any, Dict, Optional from unittest.mock import AsyncMock, Mock, patch import pytest sys.path.insert(0, os.path.abspath("../..")) from litellm.proxy.response_polling.polling_handler import ResponsePollingHandler class TestResponsePollingHandler: """Test cases for ResponsePollingHandler""" # ==================== Polling ID Tests ==================== def test_generate_polling_id_has_correct_prefix(self): """Test that generated polling IDs have the correct prefix""" polling_id = ResponsePollingHandler.generate_polling_id() assert polling_id.startswith("litellm_poll_") assert len(polling_id) > len("litellm_poll_") # Has UUID after prefix def test_generate_polling_id_is_unique(self): """Test that each generated polling ID is unique""" ids = [ResponsePollingHandler.generate_polling_id() for _ in range(100)] assert len(ids) == len(set(ids)) # All unique def test_is_polling_id_returns_true_for_polling_ids(self): """Test that is_polling_id correctly identifies polling IDs""" polling_id = ResponsePollingHandler.generate_polling_id() assert ResponsePollingHandler.is_polling_id(polling_id) is True def test_is_polling_id_returns_false_for_provider_ids(self): """Test that is_polling_id returns False for provider response IDs""" # OpenAI format assert ResponsePollingHandler.is_polling_id("resp_abc123") is False # Anthropic format assert ResponsePollingHandler.is_polling_id("msg_01XFDUDYJgAACzvnptvVoYEL") is False # Generic UUID assert ResponsePollingHandler.is_polling_id("550e8400-e29b-41d4-a716-446655440000") is False def test_get_cache_key_format(self): """Test that cache keys have the correct format""" polling_id = "litellm_poll_abc123" cache_key = ResponsePollingHandler.get_cache_key(polling_id) assert cache_key == "litellm:polling:response:litellm_poll_abc123" # ==================== Initial State Tests ==================== @pytest.mark.asyncio async def test_create_initial_state_returns_queued_status(self): """Test that create_initial_state returns response with queued status""" mock_redis = AsyncMock() handler = ResponsePollingHandler(redis_cache=mock_redis, ttl=3600) polling_id = "litellm_poll_test123" request_data = { "model": "gpt-4o", "input": "Hello", "metadata": {"test": "value"} } response = await handler.create_initial_state( polling_id=polling_id, request_data=request_data, ) assert response.id == polling_id assert response.object == "response" assert response.status == "queued" assert response.output == [] assert response.usage is None assert response.metadata == {"test": "value"} @pytest.mark.asyncio async def test_create_initial_state_stores_in_redis(self): """Test that create_initial_state stores state in Redis with correct TTL""" mock_redis = AsyncMock() handler = ResponsePollingHandler(redis_cache=mock_redis, ttl=7200) polling_id = "litellm_poll_test123" request_data = {"model": "gpt-4o", "input": "Hello"} await handler.create_initial_state( polling_id=polling_id, request_data=request_data, ) # Verify Redis was called with correct parameters mock_redis.async_set_cache.assert_called_once() call_args = mock_redis.async_set_cache.call_args assert call_args.kwargs["key"] == "litellm:polling:response:litellm_poll_test123" assert call_args.kwargs["ttl"] == 7200 # Verify the stored value is valid JSON stored_value = call_args.kwargs["value"] parsed = json.loads(stored_value) assert parsed["id"] == polling_id assert parsed["status"] == "queued" @pytest.mark.asyncio async def test_create_initial_state_sets_created_at_timestamp(self): """Test that create_initial_state sets a valid created_at timestamp""" mock_redis = AsyncMock() handler = ResponsePollingHandler(redis_cache=mock_redis) before_time = int(datetime.now(timezone.utc).timestamp()) response = await handler.create_initial_state( polling_id="litellm_poll_test", request_data={}, ) after_time = int(datetime.now(timezone.utc).timestamp()) assert before_time <= response.created_at <= after_time # ==================== State Update Tests ==================== @pytest.mark.asyncio async def test_update_state_changes_status_to_in_progress(self): """Test that update_state can change status to in_progress""" mock_redis = AsyncMock() mock_redis.async_get_cache.return_value = json.dumps({ "id": "litellm_poll_test", "object": "response", "status": "queued", "output": [], "created_at": 1234567890 }) handler = ResponsePollingHandler(redis_cache=mock_redis, ttl=3600) await handler.update_state( polling_id="litellm_poll_test", status="in_progress", ) # Verify the update was saved mock_redis.async_set_cache.assert_called_once() call_args = mock_redis.async_set_cache.call_args stored = json.loads(call_args.kwargs["value"]) assert stored["status"] == "in_progress" @pytest.mark.asyncio async def test_update_state_replaces_full_output_list(self): """Test that update_state replaces the full output list""" mock_redis = AsyncMock() mock_redis.async_get_cache.return_value = json.dumps({ "id": "litellm_poll_test", "object": "response", "status": "in_progress", "output": [{"id": "old_item", "type": "message"}], "created_at": 1234567890 }) handler = ResponsePollingHandler(redis_cache=mock_redis, ttl=3600) new_output = [ {"id": "item_1", "type": "message", "content": [{"type": "text", "text": "Hello"}]}, {"id": "item_2", "type": "message", "content": [{"type": "text", "text": "World"}]}, ] await handler.update_state( polling_id="litellm_poll_test", output=new_output, ) call_args = mock_redis.async_set_cache.call_args stored = json.loads(call_args.kwargs["value"]) assert len(stored["output"]) == 2 assert stored["output"][0]["id"] == "item_1" assert stored["output"][1]["id"] == "item_2" @pytest.mark.asyncio async def test_update_state_with_usage(self): """Test that update_state correctly stores usage data""" mock_redis = AsyncMock() mock_redis.async_get_cache.return_value = json.dumps({ "id": "litellm_poll_test", "object": "response", "status": "in_progress", "output": [], "created_at": 1234567890 }) handler = ResponsePollingHandler(redis_cache=mock_redis) usage_data = { "input_tokens": 10, "output_tokens": 50, "total_tokens": 60 } await handler.update_state( polling_id="litellm_poll_test", status="completed", usage=usage_data, ) call_args = mock_redis.async_set_cache.call_args stored = json.loads(call_args.kwargs["value"]) assert stored["status"] == "completed" assert stored["usage"] == usage_data @pytest.mark.asyncio async def test_update_state_with_reasoning_tools_tool_choice(self): """Test that update_state stores reasoning, tools, and tool_choice from response.completed""" mock_redis = AsyncMock() mock_redis.async_get_cache.return_value = json.dumps({ "id": "litellm_poll_test", "object": "response", "status": "in_progress", "output": [], "created_at": 1234567890 }) handler = ResponsePollingHandler(redis_cache=mock_redis) reasoning_data = {"effort": "medium", "summary": "Step by step analysis"} tool_choice_data = {"type": "function", "function": {"name": "get_weather"}} tools_data = [{"type": "function", "function": {"name": "get_weather", "parameters": {}}}] await handler.update_state( polling_id="litellm_poll_test", status="completed", reasoning=reasoning_data, tool_choice=tool_choice_data, tools=tools_data, ) call_args = mock_redis.async_set_cache.call_args stored = json.loads(call_args.kwargs["value"]) assert stored["reasoning"] == reasoning_data assert stored["tool_choice"] == tool_choice_data assert stored["tools"] == tools_data @pytest.mark.asyncio async def test_update_state_with_all_responses_api_fields(self): """Test that update_state stores all ResponsesAPIResponse fields from response.completed""" mock_redis = AsyncMock() mock_redis.async_get_cache.return_value = json.dumps({ "id": "litellm_poll_test", "object": "response", "status": "in_progress", "output": [], "created_at": 1234567890 }) handler = ResponsePollingHandler(redis_cache=mock_redis) # All ResponsesAPIResponse fields that can be updated await handler.update_state( polling_id="litellm_poll_test", status="completed", usage={"input_tokens": 10, "output_tokens": 50, "total_tokens": 60}, reasoning={"effort": "medium"}, tool_choice={"type": "auto"}, tools=[{"type": "function", "function": {"name": "test"}}], model="gpt-4o", instructions="You are a helpful assistant", temperature=0.7, top_p=0.9, max_output_tokens=1000, previous_response_id="resp_prev_123", text={"format": {"type": "text"}}, truncation="auto", parallel_tool_calls=True, user="user_123", store=True, incomplete_details={"reason": "max_output_tokens"}, ) call_args = mock_redis.async_set_cache.call_args stored = json.loads(call_args.kwargs["value"]) # Verify all fields are stored correctly assert stored["status"] == "completed" assert stored["usage"] == {"input_tokens": 10, "output_tokens": 50, "total_tokens": 60} assert stored["reasoning"] == {"effort": "medium"} assert stored["tool_choice"] == {"type": "auto"} assert stored["tools"] == [{"type": "function", "function": {"name": "test"}}] assert stored["model"] == "gpt-4o" assert stored["instructions"] == "You are a helpful assistant" assert stored["temperature"] == 0.7 assert stored["top_p"] == 0.9 assert stored["max_output_tokens"] == 1000 assert stored["previous_response_id"] == "resp_prev_123" assert stored["text"] == {"format": {"type": "text"}} assert stored["truncation"] == "auto" assert stored["parallel_tool_calls"] is True assert stored["user"] == "user_123" assert stored["store"] is True assert stored["incomplete_details"] == {"reason": "max_output_tokens"} @pytest.mark.asyncio async def test_update_state_preserves_existing_fields(self): """Test that update_state preserves fields not being updated""" mock_redis = AsyncMock() mock_redis.async_get_cache.return_value = json.dumps({ "id": "litellm_poll_test", "object": "response", "status": "in_progress", "output": [{"id": "item_1", "type": "message"}], "created_at": 1234567890, "model": "gpt-4o", "temperature": 0.5, }) handler = ResponsePollingHandler(redis_cache=mock_redis) # Only update status await handler.update_state( polling_id="litellm_poll_test", status="completed", ) call_args = mock_redis.async_set_cache.call_args stored = json.loads(call_args.kwargs["value"]) # Verify existing fields are preserved assert stored["status"] == "completed" assert stored["model"] == "gpt-4o" assert stored["temperature"] == 0.5 assert stored["output"] == [{"id": "item_1", "type": "message"}] @pytest.mark.asyncio async def test_update_state_with_error_sets_failed_status(self): """Test that providing an error automatically sets status to failed""" mock_redis = AsyncMock() mock_redis.async_get_cache.return_value = json.dumps({ "id": "litellm_poll_test", "object": "response", "status": "in_progress", "output": [], "created_at": 1234567890 }) handler = ResponsePollingHandler(redis_cache=mock_redis) error_data = { "type": "internal_error", "message": "Something went wrong", "code": "server_error" } await handler.update_state( polling_id="litellm_poll_test", error=error_data, ) call_args = mock_redis.async_set_cache.call_args stored = json.loads(call_args.kwargs["value"]) assert stored["status"] == "failed" assert stored["error"] == error_data @pytest.mark.asyncio async def test_update_state_with_incomplete_details(self): """Test that update_state stores incomplete_details""" mock_redis = AsyncMock() mock_redis.async_get_cache.return_value = json.dumps({ "id": "litellm_poll_test", "object": "response", "status": "in_progress", "output": [], "created_at": 1234567890 }) handler = ResponsePollingHandler(redis_cache=mock_redis) incomplete_details = { "reason": "max_output_tokens" } await handler.update_state( polling_id="litellm_poll_test", status="incomplete", incomplete_details=incomplete_details, ) call_args = mock_redis.async_set_cache.call_args stored = json.loads(call_args.kwargs["value"]) assert stored["status"] == "incomplete" assert stored["incomplete_details"] == incomplete_details @pytest.mark.asyncio async def test_update_state_does_nothing_without_redis(self): """Test that update_state gracefully handles no Redis cache""" handler = ResponsePollingHandler(redis_cache=None) # Should not raise an exception await handler.update_state( polling_id="litellm_poll_test", status="in_progress", ) @pytest.mark.asyncio async def test_update_state_handles_missing_cached_state(self): """Test that update_state handles case when cached state doesn't exist""" mock_redis = AsyncMock() mock_redis.async_get_cache.return_value = None # Cache miss handler = ResponsePollingHandler(redis_cache=mock_redis) # Should not raise an exception await handler.update_state( polling_id="litellm_poll_test", status="in_progress", ) # Should not try to set cache if nothing was found mock_redis.async_set_cache.assert_not_called() # ==================== Get State Tests ==================== @pytest.mark.asyncio async def test_get_state_returns_cached_state(self): """Test that get_state returns the cached state""" mock_redis = AsyncMock() cached_state = { "id": "litellm_poll_test", "object": "response", "status": "in_progress", "output": [{"id": "item_1", "type": "message"}], "created_at": 1234567890, "usage": {"input_tokens": 10, "output_tokens": 20} } mock_redis.async_get_cache.return_value = json.dumps(cached_state) handler = ResponsePollingHandler(redis_cache=mock_redis) result = await handler.get_state("litellm_poll_test") assert result == cached_state @pytest.mark.asyncio async def test_get_state_returns_none_for_missing_state(self): """Test that get_state returns None when state doesn't exist""" mock_redis = AsyncMock() mock_redis.async_get_cache.return_value = None handler = ResponsePollingHandler(redis_cache=mock_redis) result = await handler.get_state("litellm_poll_nonexistent") assert result is None @pytest.mark.asyncio async def test_get_state_returns_none_without_redis(self): """Test that get_state returns None when Redis is not configured""" handler = ResponsePollingHandler(redis_cache=None) result = await handler.get_state("litellm_poll_test") assert result is None # ==================== Cancel Polling Tests ==================== @pytest.mark.asyncio async def test_cancel_polling_updates_status_to_cancelled(self): """Test that cancel_polling sets status to cancelled""" mock_redis = AsyncMock() mock_redis.async_get_cache.return_value = json.dumps({ "id": "litellm_poll_test", "object": "response", "status": "in_progress", "output": [], "created_at": 1234567890 }) handler = ResponsePollingHandler(redis_cache=mock_redis) result = await handler.cancel_polling("litellm_poll_test") assert result is True call_args = mock_redis.async_set_cache.call_args stored = json.loads(call_args.kwargs["value"]) assert stored["status"] == "cancelled" # ==================== Delete Polling Tests ==================== @pytest.mark.asyncio async def test_delete_polling_removes_from_cache(self): """Test that delete_polling removes the entry from Redis""" mock_redis = AsyncMock() mock_async_client = AsyncMock() mock_redis.redis_async_client = True # hasattr check # init_async_client is a sync method that returns an async client mock_redis.init_async_client = Mock(return_value=mock_async_client) # Mock async_delete_cache to actually call init_async_client and delete async def mock_async_delete_cache(key): client = mock_redis.init_async_client() await client.delete(key) mock_redis.async_delete_cache = mock_async_delete_cache handler = ResponsePollingHandler(redis_cache=mock_redis) result = await handler.delete_polling("litellm_poll_test") assert result is True mock_async_client.delete.assert_called_once_with( "litellm:polling:response:litellm_poll_test" ) @pytest.mark.asyncio async def test_delete_polling_returns_false_without_redis(self): """Test that delete_polling returns False when Redis is not configured""" handler = ResponsePollingHandler(redis_cache=None) result = await handler.delete_polling("litellm_poll_test") assert result is False # ==================== TTL Tests ==================== def test_default_ttl_is_one_hour(self): """Test that default TTL is 3600 seconds (1 hour)""" handler = ResponsePollingHandler(redis_cache=None) assert handler.ttl == 3600 def test_custom_ttl_is_respected(self): """Test that custom TTL is stored correctly""" handler = ResponsePollingHandler(redis_cache=None, ttl=7200) assert handler.ttl == 7200 @pytest.mark.asyncio async def test_update_state_uses_configured_ttl(self): """Test that update_state uses the configured TTL""" mock_redis = AsyncMock() mock_redis.async_get_cache.return_value = json.dumps({ "id": "litellm_poll_test", "object": "response", "status": "queued", "output": [], "created_at": 1234567890 }) handler = ResponsePollingHandler(redis_cache=mock_redis, ttl=1800) await handler.update_state( polling_id="litellm_poll_test", status="in_progress", ) call_args = mock_redis.async_set_cache.call_args assert call_args.kwargs["ttl"] == 1800 class TestStreamingEventProcessing: """ Test cases for streaming event processing logic. These tests verify the expected behavior when processing different OpenAI streaming event types. """ def test_accumulated_text_structure(self): """Test the structure used for accumulating text deltas""" accumulated_text = {} # Simulate accumulating deltas for (item_id, content_index) key = ("item_123", 0) accumulated_text[key] = "" accumulated_text[key] += "Hello " accumulated_text[key] += "World" assert accumulated_text[key] == "Hello World" assert ("item_123", 0) in accumulated_text assert ("item_123", 1) not in accumulated_text def test_output_items_tracking_structure(self): """Test the structure used for tracking output items by ID""" output_items = {} # Simulate adding output items item1 = {"id": "item_1", "type": "message", "content": []} item2 = {"id": "item_2", "type": "function_call", "name": "get_weather"} output_items[item1["id"]] = item1 output_items[item2["id"]] = item2 assert len(output_items) == 2 assert output_items["item_1"]["type"] == "message" assert output_items["item_2"]["type"] == "function_call" def test_150ms_batch_interval_constant(self): """Test that the batch interval is 150ms""" UPDATE_INTERVAL = 0.150 # 150ms assert UPDATE_INTERVAL == 0.150 assert UPDATE_INTERVAL * 1000 == 150 # 150 milliseconds @pytest.mark.xdist_group("heavy_imports") class TestBackgroundStreamingModule: """Test cases for background_streaming module imports and structure""" def test_background_streaming_task_can_be_imported(self): """Test that background_streaming_task can be imported from the module""" from litellm.proxy.response_polling.background_streaming import ( background_streaming_task, ) assert background_streaming_task is not None assert callable(background_streaming_task) def test_module_exports_from_init(self): """Test that the module exports are available from __init__""" from litellm.proxy.response_polling import ( ResponsePollingHandler, background_streaming_task, ) assert ResponsePollingHandler is not None assert background_streaming_task is not None def test_background_streaming_task_is_async(self): """Test that background_streaming_task is an async function""" import inspect from litellm.proxy.response_polling.background_streaming import ( background_streaming_task, ) assert inspect.iscoroutinefunction(background_streaming_task) class TestProviderResolutionForPolling: """ Test cases for provider resolution logic used to determine if polling_via_cache should be enabled for a given model. This tests the logic in endpoints.py that resolves model names to their providers using the router's deployment configuration. """ def test_provider_from_model_string_with_slash(self): """Test extracting provider from 'provider/model' format""" model = "openai/gpt-4o" # Direct extraction when model has slash if "/" in model: provider = model.split("/")[0] else: provider = None assert provider == "openai" def test_provider_from_model_string_without_slash(self): """Test that model without slash doesn't extract provider directly""" model = "gpt-5" # No slash means we can't extract provider directly if "/" in model: provider = model.split("/")[0] else: provider = None assert provider is None def test_provider_resolution_from_router_single_deployment(self): """Test resolving provider from router with single deployment""" # Simulate router's model_name_to_deployment_indices model_name_to_deployment_indices = { "gpt-5": [0], # Single deployment at index 0 } model_list = [ { "model_name": "gpt-5", "litellm_params": { "model": "openai/gpt-5", "api_key": "sk-test", } } ] model = "gpt-5" polling_via_cache_enabled = ["openai"] should_use_polling = False # Simulate the resolution logic indices = model_name_to_deployment_indices.get(model, []) for idx in indices: deployment_dict = model_list[idx] litellm_params = deployment_dict.get("litellm_params", {}) dep_provider = litellm_params.get("custom_llm_provider") if not dep_provider: dep_model = litellm_params.get("model", "") if "/" in dep_model: dep_provider = dep_model.split("/")[0] if dep_provider and dep_provider in polling_via_cache_enabled: should_use_polling = True break assert should_use_polling is True def test_provider_resolution_from_router_multiple_deployments_match(self): """Test resolving provider when multiple deployments exist and one matches""" model_name_to_deployment_indices = { "gpt-4o": [0, 1], # Two deployments } model_list = [ { "model_name": "gpt-4o", "litellm_params": { "model": "openai/gpt-4o", } }, { "model_name": "gpt-4o", "litellm_params": { "model": "azure/gpt-4o-deployment", } } ] model = "gpt-4o" polling_via_cache_enabled = ["openai"] # Only openai in list should_use_polling = False indices = model_name_to_deployment_indices.get(model, []) for idx in indices: deployment_dict = model_list[idx] litellm_params = deployment_dict.get("litellm_params", {}) dep_provider = litellm_params.get("custom_llm_provider") if not dep_provider: dep_model = litellm_params.get("model", "") if "/" in dep_model: dep_provider = dep_model.split("/")[0] if dep_provider and dep_provider in polling_via_cache_enabled: should_use_polling = True break # Should be True because first deployment is openai assert should_use_polling is True def test_provider_resolution_from_router_no_match(self): """Test that polling is disabled when no deployment provider matches""" model_name_to_deployment_indices = { "claude-3": [0], } model_list = [ { "model_name": "claude-3", "litellm_params": { "model": "anthropic/claude-3-sonnet", } } ] model = "claude-3" polling_via_cache_enabled = ["openai", "bedrock"] # anthropic not in list should_use_polling = False indices = model_name_to_deployment_indices.get(model, []) for idx in indices: deployment_dict = model_list[idx] litellm_params = deployment_dict.get("litellm_params", {}) dep_provider = litellm_params.get("custom_llm_provider") if not dep_provider: dep_model = litellm_params.get("model", "") if "/" in dep_model: dep_provider = dep_model.split("/")[0] if dep_provider and dep_provider in polling_via_cache_enabled: should_use_polling = True break assert should_use_polling is False def test_provider_resolution_with_custom_llm_provider(self): """Test that custom_llm_provider takes precedence over model string""" model_name_to_deployment_indices = { "my-model": [0], } model_list = [ { "model_name": "my-model", "litellm_params": { "model": "some-custom-model", "custom_llm_provider": "openai", # Explicit provider } } ] model = "my-model" polling_via_cache_enabled = ["openai"] should_use_polling = False indices = model_name_to_deployment_indices.get(model, []) for idx in indices: deployment_dict = model_list[idx] litellm_params = deployment_dict.get("litellm_params", {}) # custom_llm_provider should be checked first dep_provider = litellm_params.get("custom_llm_provider") if not dep_provider: dep_model = litellm_params.get("model", "") if "/" in dep_model: dep_provider = dep_model.split("/")[0] if dep_provider and dep_provider in polling_via_cache_enabled: should_use_polling = True break assert should_use_polling is True def test_provider_resolution_model_not_in_router(self): """Test that unknown model doesn't enable polling""" model_name_to_deployment_indices = { "gpt-5": [0], } model_list = [ { "model_name": "gpt-5", "litellm_params": {"model": "openai/gpt-5"} } ] model = "unknown-model" # Not in router polling_via_cache_enabled = ["openai"] should_use_polling = False indices = model_name_to_deployment_indices.get(model, []) # Empty list for idx in indices: # This loop won't execute pass assert should_use_polling is False assert len(indices) == 0 class TestPollingConditionChecks: """ Test cases for the conditions that determine whether polling should be enabled. Tests the should_use_polling_for_request function. """ def test_polling_enabled_when_all_conditions_met(self): """Test polling is enabled when background=true, polling_via_cache="all", and redis is available""" from litellm.proxy.response_polling.polling_handler import should_use_polling_for_request result = should_use_polling_for_request( background_mode=True, polling_via_cache_enabled="all", redis_cache=Mock(), model="gpt-4o", llm_router=None, ) assert result is True def test_polling_disabled_when_background_false(self): """Test polling is disabled when background=false""" from litellm.proxy.response_polling.polling_handler import should_use_polling_for_request result = should_use_polling_for_request( background_mode=False, polling_via_cache_enabled="all", redis_cache=Mock(), model="gpt-4o", llm_router=None, ) assert result is False def test_polling_disabled_when_config_false(self): """Test polling is disabled when polling_via_cache is False""" from litellm.proxy.response_polling.polling_handler import should_use_polling_for_request result = should_use_polling_for_request( background_mode=True, polling_via_cache_enabled=False, redis_cache=Mock(), model="gpt-4o", llm_router=None, ) assert result is False def test_polling_disabled_when_redis_not_configured(self): """Test polling is disabled when Redis is not configured""" from litellm.proxy.response_polling.polling_handler import should_use_polling_for_request result = should_use_polling_for_request( background_mode=True, polling_via_cache_enabled="all", redis_cache=None, model="gpt-4o", llm_router=None, ) assert result is False def test_polling_enabled_with_provider_list_match(self): """Test polling is enabled when provider list matches""" from litellm.proxy.response_polling.polling_handler import should_use_polling_for_request result = should_use_polling_for_request( background_mode=True, polling_via_cache_enabled=["openai", "anthropic"], redis_cache=Mock(), model="openai/gpt-4o", llm_router=None, ) assert result is True def test_polling_disabled_with_provider_list_no_match(self): """Test polling is disabled when provider not in list""" from litellm.proxy.response_polling.polling_handler import should_use_polling_for_request result = should_use_polling_for_request( background_mode=True, polling_via_cache_enabled=["openai"], redis_cache=Mock(), model="anthropic/claude-3", llm_router=None, ) assert result is False def test_polling_with_router_lookup(self): """Test polling uses router to resolve model name to provider""" from litellm.proxy.response_polling.polling_handler import should_use_polling_for_request # Create mock router mock_router = Mock() mock_router.model_name_to_deployment_indices = {"gpt-5": [0]} mock_router.model_list = [ { "model_name": "gpt-5", "litellm_params": {"model": "openai/gpt-5"} } ] result = should_use_polling_for_request( background_mode=True, polling_via_cache_enabled=["openai"], redis_cache=Mock(), model="gpt-5", # No slash, needs router lookup llm_router=mock_router, ) assert result is True def test_polling_with_router_lookup_no_match(self): """Test polling returns False when router lookup finds non-matching provider""" from litellm.proxy.response_polling.polling_handler import should_use_polling_for_request mock_router = Mock() mock_router.model_name_to_deployment_indices = {"claude-3": [0]} mock_router.model_list = [ { "model_name": "claude-3", "litellm_params": {"model": "anthropic/claude-3-sonnet"} } ] result = should_use_polling_for_request( background_mode=True, polling_via_cache_enabled=["openai"], redis_cache=Mock(), model="claude-3", llm_router=mock_router, ) assert result is False # ==================== Native Background Mode Tests ==================== def test_polling_disabled_when_model_in_native_background_mode(self): """Test that polling is disabled when model is in native_background_mode list""" from litellm.proxy.response_polling.polling_handler import should_use_polling_for_request result = should_use_polling_for_request( background_mode=True, polling_via_cache_enabled="all", redis_cache=Mock(), model="o4-mini-deep-research", llm_router=None, native_background_mode=["o4-mini-deep-research", "o3-deep-research"], ) assert result is False def test_polling_disabled_for_native_background_mode_with_provider_list(self): """Test that native_background_mode takes precedence even when provider matches""" from litellm.proxy.response_polling.polling_handler import should_use_polling_for_request result = should_use_polling_for_request( background_mode=True, polling_via_cache_enabled=["openai"], redis_cache=Mock(), model="openai/o4-mini-deep-research", llm_router=None, native_background_mode=["openai/o4-mini-deep-research"], ) assert result is False def test_polling_enabled_when_model_not_in_native_background_mode(self): """Test that polling is enabled when model is not in native_background_mode list""" from litellm.proxy.response_polling.polling_handler import should_use_polling_for_request result = should_use_polling_for_request( background_mode=True, polling_via_cache_enabled="all", redis_cache=Mock(), model="gpt-4o", llm_router=None, native_background_mode=["o4-mini-deep-research"], ) assert result is True def test_polling_enabled_when_native_background_mode_is_none(self): """Test that polling works normally when native_background_mode is None""" from litellm.proxy.response_polling.polling_handler import should_use_polling_for_request result = should_use_polling_for_request( background_mode=True, polling_via_cache_enabled="all", redis_cache=Mock(), model="gpt-4o", llm_router=None, native_background_mode=None, ) assert result is True def test_polling_enabled_when_native_background_mode_is_empty_list(self): """Test that polling works normally when native_background_mode is empty list""" from litellm.proxy.response_polling.polling_handler import should_use_polling_for_request result = should_use_polling_for_request( background_mode=True, polling_via_cache_enabled="all", redis_cache=Mock(), model="gpt-4o", llm_router=None, native_background_mode=[], ) assert result is True def test_native_background_mode_exact_match_required(self): """Test that native_background_mode uses exact model name matching""" from litellm.proxy.response_polling.polling_handler import should_use_polling_for_request # "o4-mini" should not match "o4-mini-deep-research" result = should_use_polling_for_request( background_mode=True, polling_via_cache_enabled="all", redis_cache=Mock(), model="o4-mini", llm_router=None, native_background_mode=["o4-mini-deep-research"], ) assert result is True def test_native_background_mode_with_provider_prefix_in_request(self): """Test native_background_mode matching when request model has provider prefix""" from litellm.proxy.response_polling.polling_handler import should_use_polling_for_request # Model in native_background_mode without provider prefix # Request comes in with provider prefix - should not match result = should_use_polling_for_request( background_mode=True, polling_via_cache_enabled=["openai"], redis_cache=Mock(), model="openai/o4-mini-deep-research", llm_router=None, native_background_mode=["o4-mini-deep-research"], # Without prefix ) # Should return True because "openai/o4-mini-deep-research" != "o4-mini-deep-research" assert result is True def test_native_background_mode_with_router_lookup(self): """Test that native_background_mode works with router-resolved models""" from litellm.proxy.response_polling.polling_handler import should_use_polling_for_request mock_router = Mock() mock_router.model_name_to_deployment_indices = {"deep-research": [0]} mock_router.model_list = [ { "model_name": "deep-research", "litellm_params": {"model": "openai/o4-mini-deep-research"} } ] # Model alias "deep-research" is in native_background_mode result = should_use_polling_for_request( background_mode=True, polling_via_cache_enabled=["openai"], redis_cache=Mock(), model="deep-research", llm_router=mock_router, native_background_mode=["deep-research"], ) assert result is False class TestStreamingEventParsing: """ Test cases for parsing OpenAI streaming events in the background task. Tests the event handling logic in background_streaming.py. """ def test_parse_response_output_item_added_event(self): """Test parsing response.output_item.added event""" event = { "type": "response.output_item.added", "item": { "id": "item_123", "type": "message", "role": "assistant", "content": [] } } output_items = {} event_type = event.get("type", "") if event_type == "response.output_item.added": item = event.get("item", {}) item_id = item.get("id") if item_id: output_items[item_id] = item assert "item_123" in output_items assert output_items["item_123"]["type"] == "message" def test_parse_response_output_text_delta_event(self): """Test parsing response.output_text.delta event and accumulating text""" output_items = { "item_123": { "id": "item_123", "type": "message", "content": [{"type": "text", "text": ""}] } } accumulated_text = {} # Simulate receiving multiple delta events delta_events = [ {"type": "response.output_text.delta", "item_id": "item_123", "content_index": 0, "delta": "Hello "}, {"type": "response.output_text.delta", "item_id": "item_123", "content_index": 0, "delta": "World!"}, ] for event in delta_events: event_type = event.get("type", "") if event_type == "response.output_text.delta": item_id = event.get("item_id") content_index = event.get("content_index", 0) delta = event.get("delta", "") if item_id and item_id in output_items: key = (item_id, content_index) if key not in accumulated_text: accumulated_text[key] = "" accumulated_text[key] += delta # Update content if "content" in output_items[item_id]: content_list = output_items[item_id]["content"] if content_index < len(content_list): if isinstance(content_list[content_index], dict): content_list[content_index]["text"] = accumulated_text[key] assert accumulated_text[("item_123", 0)] == "Hello World!" assert output_items["item_123"]["content"][0]["text"] == "Hello World!" def test_parse_response_completed_event(self): """Test parsing response.completed event extracts all fields""" event = { "type": "response.completed", "response": { "id": "resp_123", "status": "completed", "usage": {"input_tokens": 10, "output_tokens": 50}, "reasoning": {"effort": "medium"}, "tool_choice": {"type": "auto"}, "tools": [{"type": "function", "function": {"name": "test"}}], "model": "gpt-4o", "output": [{"id": "item_1", "type": "message"}] } } event_type = event.get("type", "") usage_data = None reasoning_data = None tool_choice_data = None tools_data = None model_data = None if event_type == "response.completed": response_data = event.get("response", {}) usage_data = response_data.get("usage") reasoning_data = response_data.get("reasoning") tool_choice_data = response_data.get("tool_choice") tools_data = response_data.get("tools") model_data = response_data.get("model") assert usage_data == {"input_tokens": 10, "output_tokens": 50} assert reasoning_data == {"effort": "medium"} assert tool_choice_data == {"type": "auto"} assert tools_data == [{"type": "function", "function": {"name": "test"}}] assert model_data == "gpt-4o" def test_parse_done_marker(self): """Test that [DONE] marker is detected correctly""" chunks = [ "data: {\"type\": \"response.in_progress\"}", "data: {\"type\": \"response.completed\"}", "data: [DONE]", ] done_received = False for chunk in chunks: if chunk.startswith("data: "): chunk_data = chunk[6:].strip() if chunk_data == "[DONE]": done_received = True break assert done_received is True def test_parse_sse_format(self): """Test parsing Server-Sent Events format""" raw_chunk = b"data: {\"type\": \"response.output_item.added\", \"item\": {\"id\": \"123\"}}" # Decode bytes to string if isinstance(raw_chunk, bytes): chunk = raw_chunk.decode('utf-8') else: chunk = raw_chunk # Extract JSON from SSE format if isinstance(chunk, str) and chunk.startswith("data: "): chunk_data = chunk[6:].strip() import json event = json.loads(chunk_data) assert event["type"] == "response.output_item.added" assert event["item"]["id"] == "123" def test_content_part_added_event(self): """Test parsing response.content_part.added event""" output_items = { "item_123": { "id": "item_123", "type": "message", } } event = { "type": "response.content_part.added", "item_id": "item_123", "part": {"type": "text", "text": ""} } event_type = event.get("type", "") if event_type == "response.content_part.added": item_id = event.get("item_id") content_part = event.get("part", {}) if item_id and item_id in output_items: if "content" not in output_items[item_id]: output_items[item_id]["content"] = [] output_items[item_id]["content"].append(content_part) assert "content" in output_items["item_123"] assert len(output_items["item_123"]["content"]) == 1 assert output_items["item_123"]["content"][0]["type"] == "text" class TestEdgeCases: """Test edge cases and error scenarios""" def test_empty_model_string(self): """Test handling of empty model string""" model = "" polling_via_cache_enabled = ["openai"] should_use_polling = False if "/" in model: provider = model.split("/")[0] if provider in polling_via_cache_enabled: should_use_polling = True assert should_use_polling is False def test_model_with_multiple_slashes(self): """Test handling model with multiple slashes (e.g., bedrock ARN)""" model = "bedrock/arn:aws:bedrock:us-east-1:123456:model/my-model" polling_via_cache_enabled = ["bedrock"] # Only split on first slash if "/" in model: provider = model.split("/")[0] else: provider = None assert provider == "bedrock" assert provider in polling_via_cache_enabled def test_polling_id_detection_edge_cases(self): """Test polling ID detection with edge cases""" # Empty string assert ResponsePollingHandler.is_polling_id("") is False # Just prefix without UUID assert ResponsePollingHandler.is_polling_id("litellm_poll_") is True # Similar but different prefix assert ResponsePollingHandler.is_polling_id("litellm_polling_abc") is False # Case sensitivity assert ResponsePollingHandler.is_polling_id("LITELLM_POLL_abc") is False @pytest.mark.asyncio async def test_create_initial_state_with_empty_metadata(self): """Test create_initial_state handles missing metadata gracefully""" mock_redis = AsyncMock() handler = ResponsePollingHandler(redis_cache=mock_redis) response = await handler.create_initial_state( polling_id="litellm_poll_test", request_data={"model": "gpt-4o"}, # No metadata field ) assert response.metadata == {} @pytest.mark.asyncio async def test_update_state_with_none_output_clears_output(self): """Test that output=[] explicitly sets empty output""" mock_redis = AsyncMock() mock_redis.async_get_cache.return_value = json.dumps({ "id": "litellm_poll_test", "object": "response", "status": "in_progress", "output": [{"id": "item_1"}], # Has existing output "created_at": 1234567890 }) handler = ResponsePollingHandler(redis_cache=mock_redis) await handler.update_state( polling_id="litellm_poll_test", output=[], # Explicitly set empty ) call_args = mock_redis.async_set_cache.call_args stored = json.loads(call_args.kwargs["value"]) assert stored["output"] == []