From cbb161f10b817f0968a5dfc44566b85b27b0e5f2 Mon Sep 17 00:00:00 2001 From: Krrish Dholakia Date: Sat, 23 Aug 2025 11:00:18 -0700 Subject: [PATCH] test: handle internal server errors --- .../base_responses_api.py | 224 +++++++++--------- 1 file changed, 114 insertions(+), 110 deletions(-) diff --git a/tests/llm_responses_api_testing/base_responses_api.py b/tests/llm_responses_api_testing/base_responses_api.py index 72f6fb8c9cf..5cb8295b1af 100644 --- a/tests/llm_responses_api_testing/base_responses_api.py +++ b/tests/llm_responses_api_testing/base_responses_api.py @@ -101,11 +101,11 @@ def validate_responses_api_response(response, final_chunk: bool = False): return True # Return True if validation passes - class BaseResponsesAPITest(ABC): """ Abstract base test class that enforces a common test across all test classes. """ + @abstractmethod def get_base_completion_call_args(self) -> dict: """Must return the base completion call args""" @@ -115,32 +115,32 @@ class BaseResponsesAPITest(ABC): """Must return the base completion reasoning call args""" return None - @pytest.mark.parametrize("sync_mode", [True, False]) @pytest.mark.asyncio async def test_basic_openai_responses_api(self, sync_mode): litellm._turn_on_debug() litellm.set_verbose = True base_completion_call_args = self.get_base_completion_call_args() - try: + try: if sync_mode: response = litellm.responses( - input="Basic ping", max_output_tokens=20, - **base_completion_call_args + input="Basic ping", + max_output_tokens=20, + **base_completion_call_args, ) else: response = await litellm.aresponses( - input="Basic ping", max_output_tokens=20, - **base_completion_call_args + input="Basic ping", + max_output_tokens=20, + **base_completion_call_args, ) - except litellm.InternalServerError: + except litellm.InternalServerError: pytest.skip("Skipping test due to litellm.InternalServerError") print("litellm response=", json.dumps(response, indent=4, default=str)) # Use the helper function to validate the response validate_responses_api_response(response, final_chunk=True) - @pytest.mark.parametrize("sync_mode", [True, False]) @pytest.mark.asyncio @pytest.mark.flaky(retries=3, delay=2) @@ -151,9 +151,7 @@ class BaseResponsesAPITest(ABC): response_completed_event = None if sync_mode: response = litellm.responses( - input="Basic ping", - stream=True, - **base_completion_call_args + input="Basic ping", stream=True, **base_completion_call_args ) for event in response: print("litellm response=", json.dumps(event, indent=4, default=str)) @@ -163,9 +161,7 @@ class BaseResponsesAPITest(ABC): response_completed_event = event else: response = await litellm.aresponses( - input="Basic ping", - stream=True, - **base_completion_call_args + input="Basic ping", stream=True, **base_completion_call_args ) async for event in response: print("litellm response=", json.dumps(event, indent=4, default=str)) @@ -188,15 +184,29 @@ class BaseResponsesAPITest(ABC): assert response_completed_event.response.usage is not None # basic test assert the usage seems reasonable - print("response_completed_event.response.usage=", response_completed_event.response.usage) - assert response_completed_event.response.usage.input_tokens > 0 and response_completed_event.response.usage.input_tokens < 100 - assert response_completed_event.response.usage.output_tokens > 0 and response_completed_event.response.usage.output_tokens < 2000 - assert response_completed_event.response.usage.total_tokens > 0 and response_completed_event.response.usage.total_tokens < 2000 + print( + "response_completed_event.response.usage=", + response_completed_event.response.usage, + ) + assert ( + response_completed_event.response.usage.input_tokens > 0 + and response_completed_event.response.usage.input_tokens < 100 + ) + assert ( + response_completed_event.response.usage.output_tokens > 0 + and response_completed_event.response.usage.output_tokens < 2000 + ) + assert ( + response_completed_event.response.usage.total_tokens > 0 + and response_completed_event.response.usage.total_tokens < 2000 + ) # total tokens should be the sum of input and output tokens - assert response_completed_event.response.usage.total_tokens == response_completed_event.response.usage.input_tokens + response_completed_event.response.usage.output_tokens - - + assert ( + response_completed_event.response.usage.total_tokens + == response_completed_event.response.usage.input_tokens + + response_completed_event.response.usage.output_tokens + ) @pytest.mark.parametrize("sync_mode", [False, True]) @pytest.mark.asyncio @@ -206,48 +216,44 @@ class BaseResponsesAPITest(ABC): base_completion_call_args = self.get_base_completion_call_args() if sync_mode: response = litellm.responses( - input="Basic ping", max_output_tokens=20, - **base_completion_call_args + input="Basic ping", max_output_tokens=20, **base_completion_call_args ) # delete the response if isinstance(response, ResponsesAPIResponse): litellm.delete_responses( - response_id=response.id, - **base_completion_call_args + response_id=response.id, **base_completion_call_args ) else: raise ValueError("response is not a ResponsesAPIResponse") else: response = await litellm.aresponses( - input="Basic ping", max_output_tokens=20, - **base_completion_call_args + input="Basic ping", max_output_tokens=20, **base_completion_call_args ) # async delete the response if isinstance(response, ResponsesAPIResponse): await litellm.adelete_responses( - response_id=response.id, - **base_completion_call_args + response_id=response.id, **base_completion_call_args ) else: raise ValueError("response is not a ResponsesAPIResponse") - @pytest.mark.parametrize("sync_mode", [True, False]) @pytest.mark.flaky(retries=3, delay=2) @pytest.mark.asyncio async def test_basic_openai_responses_streaming_delete_endpoint(self, sync_mode): - #litellm._turn_on_debug() - #litellm.set_verbose = True + # litellm._turn_on_debug() + # litellm.set_verbose = True base_completion_call_args = self.get_base_completion_call_args() response_id = None if sync_mode: response_id = None response = litellm.responses( - input="Basic ping", max_output_tokens=20, + input="Basic ping", + max_output_tokens=20, stream=True, - **base_completion_call_args + **base_completion_call_args, ) for event in response: print("litellm response=", json.dumps(event, indent=4, default=str)) @@ -260,14 +266,14 @@ class BaseResponsesAPITest(ABC): # delete the response assert response_id is not None litellm.delete_responses( - response_id=response_id, - **base_completion_call_args + response_id=response_id, **base_completion_call_args ) else: response = await litellm.aresponses( - input="Basic ping", max_output_tokens=20, + input="Basic ping", + max_output_tokens=20, stream=True, - **base_completion_call_args + **base_completion_call_args, ) async for event in response: print("litellm response=", json.dumps(event, indent=4, default=str)) @@ -280,8 +286,7 @@ class BaseResponsesAPITest(ABC): # delete the response assert response_id is not None await litellm.adelete_responses( - response_id=response_id, - **base_completion_call_args + response_id=response_id, **base_completion_call_args ) @pytest.mark.parametrize("sync_mode", [False, True]) @@ -293,15 +298,13 @@ class BaseResponsesAPITest(ABC): base_completion_call_args = self.get_base_completion_call_args() if sync_mode: response = litellm.responses( - input="Basic ping", max_output_tokens=20, - **base_completion_call_args + input="Basic ping", max_output_tokens=20, **base_completion_call_args ) # get the response if isinstance(response, ResponsesAPIResponse): result = litellm.get_responses( - response_id=response.id, - **base_completion_call_args + response_id=response.id, **base_completion_call_args ) assert result is not None assert result.id == response.id @@ -310,14 +313,12 @@ class BaseResponsesAPITest(ABC): raise ValueError("response is not a ResponsesAPIResponse") else: response = await litellm.aresponses( - input="Basic ping", max_output_tokens=20, - **base_completion_call_args + input="Basic ping", max_output_tokens=20, **base_completion_call_args ) # async get the response if isinstance(response, ResponsesAPIResponse): result = await litellm.aget_responses( - response_id=response.id, - **base_completion_call_args + response_id=response.id, **base_completion_call_args ) assert result is not None assert result.id == response.id @@ -351,58 +352,60 @@ class BaseResponsesAPITest(ABC): json.dumps(list_items_response, indent=4, default=str), ) - @pytest.mark.asyncio async def test_multiturn_responses_api(self): litellm._turn_on_debug() litellm.set_verbose = True - base_completion_call_args = self.get_base_completion_call_args() - response_1 = await litellm.aresponses( - input="Basic ping", max_output_tokens=20, **base_completion_call_args - ) + try: + base_completion_call_args = self.get_base_completion_call_args() + response_1 = await litellm.aresponses( + input="Basic ping", max_output_tokens=20, **base_completion_call_args + ) - # follow up with a second request - response_1_id = response_1.id - response_2 = await litellm.aresponses( - input="Basic ping", - max_output_tokens=20, - previous_response_id=response_1_id, - **base_completion_call_args - ) + # follow up with a second request + response_1_id = response_1.id + response_2 = await litellm.aresponses( + input="Basic ping", + max_output_tokens=20, + previous_response_id=response_1_id, + **base_completion_call_args, + ) + + # assert the response is not None + assert response_1 is not None + assert response_2 is not None + except litellm.InternalServerError: + pytest.skip("Skipping test due to litellm.InternalServerError") - # assert the response is not None - assert response_1 is not None - assert response_2 is not None - @pytest.mark.asyncio async def test_responses_api_with_tool_calls(self): """Test that calls the Responses API with tool calls including function call and output""" litellm._turn_on_debug() litellm.set_verbose = True base_completion_call_args = self.get_base_completion_call_args() - + # Define the input with message, function call, and function call output input_data: ResponseInputParam = [ { "type": "message", "role": "user", - "content": "How is the weather in São Paulo today ?" + "content": "How is the weather in São Paulo today ?", }, { "type": "function_call", - "arguments": "{\"location\": \"São Paulo, Brazil\"}", + "arguments": '{"location": "São Paulo, Brazil"}', "call_id": "fc_1fe70e2a-a596-45ef-b72c-9b8567c460e5", "name": "get_weather", "id": "fc_1fe70e2a-a596-45ef-b72c-9b8567c460e5", - "status": "completed" + "status": "completed", }, { "type": "function_call_output", "call_id": "fc_1fe70e2a-a596-45ef-b72c-9b8567c460e5", - "output": "Rainy" - } + "output": "Rainy", + }, ] - + # Define the tools tools = [ { @@ -414,71 +417,67 @@ class BaseResponsesAPITest(ABC): "properties": { "location": { "type": "string", - "description": "City and country e.g. Bogotá, Colombia" + "description": "City and country e.g. Bogotá, Colombia", } }, "required": ["location"], - "additionalProperties": False - } + "additionalProperties": False, + }, } ] - + try: # Make the responses API call response = await litellm.aresponses( - input=input_data, - store=False, - tools=tools, - **base_completion_call_args + input=input_data, store=False, tools=tools, **base_completion_call_args ) except litellm.InternalServerError: pytest.skip("Skipping test due to litellm.InternalServerError") - + print("litellm response=", json.dumps(response, indent=4, default=str)) - + # Validate the response structure validate_responses_api_response(response, final_chunk=True) - + # Additional assertions specific to tool calls assert response is not None assert "output" in response assert len(response["output"]) > 0 - + @pytest.mark.asyncio async def test_responses_api_multi_turn_with_reasoning_and_structured_output(self): """ Test multi-turn conversation with reasoning, structured output, and tool calls. - + This test validates: - First call: Model uses reasoning to process a question and makes a tool call - - Tool call handling: Function call output is properly processed + - Tool call handling: Function call output is properly processed - Second call: Model produces structured output incorporating tool results - Structured output: Response conforms to defined Pydantic model schema """ from pydantic import BaseModel - + litellm._turn_on_debug() litellm.set_verbose = True base_completion_call_args = self.get_base_completion_reasoning_call_args() if base_completion_call_args is None: pytest.skip("Skipping test due to no base completion reasoning call args") - + # Define tools for the conversation tools = [{"type": "function", "name": "get_today"}] - + # Define structured output schema class Output(BaseModel): today: str number_of_r: str - + # Initial conversation input input_messages = [ { - "role": "user", + "role": "user", "content": "How many r in strrawberrry? While you're thinking, you should call tool get_today. Then you output the today and number of r", } ] - # First call - should trigger reasoning and tool call response = await litellm.aresponses( @@ -486,49 +485,54 @@ class BaseResponsesAPITest(ABC): tools=tools, reasoning={"effort": "low", "summary": "detailed"}, text_format=Output, - **base_completion_call_args + **base_completion_call_args, ) print("First call output:") print(json.dumps(response.output, indent=4, default=str)) - + # Validate first response structure validate_responses_api_response(response, final_chunk=True) assert response.output is not None assert len(response.output) > 0 - + # Extend input with first response output input_messages.extend(response.output) - + # Process any tool calls and add function outputs function_outputs = [] for item in response.output: - if hasattr(item, 'type') and item.type in ["function_call", "custom_tool_call"]: - if hasattr(item, 'name') and item.name == "get_today": - function_outputs.append({ - "type": "function_call_output", - "call_id": item.call_id, - "output": "2025-01-15" - }) - + if hasattr(item, "type") and item.type in [ + "function_call", + "custom_tool_call", + ]: + if hasattr(item, "name") and item.name == "get_today": + function_outputs.append( + { + "type": "function_call_output", + "call_id": item.call_id, + "output": "2025-01-15", + } + ) + # Add function outputs to conversation input_messages.extend(function_outputs) - + print("Second call input:") print(json.dumps(input_messages, indent=4, default=str)) - + # Second call - should produce structured output final_response = await litellm.aresponses( input=input_messages, tools=tools, reasoning={"effort": "low", "summary": "detailed"}, text_format=Output, - **base_completion_call_args + **base_completion_call_args, ) - + print("Second call output:") print(json.dumps(final_response.output, indent=4, default=str)) - + # Validate final response structure validate_responses_api_response(final_response, final_chunk=True) assert final_response.output is not None