scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-10-09 03:18:44 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 109

test: delete 34 legacy tests already covered by e2e (#45340)

Browse source 
Co-authored-by: yuneng <yuneng@berri.ai>
This commit is contained in:
devin-ai-integration[bot] 2026-10-08 09:43:56 -07:00 • committed by GitHub
parent 35b0992533
commit 92654b69a9
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
24 changed files with 0 additions and 1884 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

45
.circleci/config.yml
View file

@ -1095,50 +1095,6 @@ jobs:
- google_generate_content_endpoint_coverage.xml
- google_generate_content_endpoint_coverage
llm_responses_api_testing:
docker:
- *python312_image
working_directory: ~/project
resource_class: large
environment:
REQUEST_TIMEOUT: "180"
steps:
- checkout
- skip_if_unrelated_changes
- setup_google_dns
- install_uv
- install_rust
- restore_cache:
keys:
- v1-uv-cache-{{ checksum "uv.lock" }}
- run:
name: Install Dependencies
command: |
uv sync --frozen --all-groups --all-extras --python 3.12
- save_cache:
paths:
- ~/.cache/uv
key: v1-uv-cache-{{ checksum "uv.lock" }}
# Run pytest and generate JUnit XML report
- run:
name: Run tests
command: |
mkdir -p test-results
TEST_FILES=$(circleci tests glob "tests/llm_responses_api_testing/**/test_*.py")
echo "$TEST_FILES" | circleci tests run \
--verbose \
--command="tr ' ' '\\n' | awk '/\\.py/ {print; next} {sub(/\\.[A-Z][^.]*$/, \"\"); gsub(/\\./, \"/\"); print \$0 \".py\"}' | xargs uv run --no-sync python -m pytest --tb=short \
-v \
--junitxml=test-results/junit.xml \
--durations=5 \
-n 8 \
--reruns 1 --only-rerun Timeout"
no_output_timeout: 15m
# Store test results
- store_test_results:
path: test-results
ocr_testing:
docker:
- *python312_image
@ -3516,7 +3472,6 @@ workflows:
- realtime_translation_testing
- guardrails_testing
- google_generate_content_endpoint_testing
- llm_responses_api_testing
- ocr_testing
- search_testing
- batches_testing

107
tests/litellm_utils_tests/test_bedrock_token_counter.py
View file

@ -1,107 +0,0 @@
"""
Bedrock Token Counter Tests.
Tests for the Bedrock token counter implementation using the base test suite.
Note: Not all Bedrock models support token counting. The CountTokens API
is only available for specific models. If the model doesn't support token
counting, the test will be skipped.
"""
import os
from typing import Any, Dict, List
import pytest
from litellm.llms.base_llm.base_utils import BaseTokenCounter
from litellm.llms.bedrock.count_tokens.bedrock_token_counter import BedrockTokenCounter
from tests.litellm_utils_tests.base_token_counter_test import BaseTokenCounterTest
class TestBedrockTokenCounter(BaseTokenCounterTest):
"""Test suite for Bedrock token counter.
Note: Bedrock CountTokens API support varies by model. Some models
(like older Claude versions) may not support token counting.
Use amazon.nova-* models for reliable token counting support.
"""
def get_token_counter(self) -> BaseTokenCounter:
return BedrockTokenCounter()
def get_test_model(self) -> str:
# Use Amazon Nova model which supports token counting
# Alternatively, use environment variable to override
return os.getenv("BEDROCK_TEST_MODEL", "amazon.nova-lite-v1:0")
def get_test_messages(self) -> List[Dict[str, Any]]:
return [{"role": "user", "content": "Hello, how are you today?"}]
def get_deployment_config(self) -> Dict[str, Any]:
# Bedrock uses AWS credentials from environment
# Check for AWS credentials
aws_access_key = os.getenv("AWS_ACCESS_KEY_ID")
aws_secret_key = os.getenv("AWS_SECRET_ACCESS_KEY")
aws_region = os.getenv("AWS_REGION_NAME", "us-east-1")
if not aws_access_key or not aws_secret_key:
pytest.skip(
"AWS credentials not set (AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY)"
)
return {
"litellm_params": {
"aws_access_key_id": aws_access_key,
"aws_secret_access_key": aws_secret_key,
"aws_region_name": aws_region,
}
}
def get_custom_llm_provider(self) -> str:
return "bedrock"
@pytest.mark.asyncio
async def test_count_tokens_basic(self):
"""
Test basic token counting functionality.
Override to handle models that don't support token counting.
"""
from litellm.types.utils import TokenCountResponse
token_counter = self.get_token_counter()
model = self.get_test_model()
messages = self.get_test_messages()
deployment = self.get_deployment_config()
result = await token_counter.count_tokens(
model_to_use=model,
messages=messages,
contents=None,
deployment=deployment,
request_model=model,
)
print(f"Token count result: {result}")
assert result is not None, "Token counter should return a result"
assert isinstance(
result, TokenCountResponse
), "Result should be TokenCountResponse"
# Check if the model doesn't support token counting
if result.error and "doesn't support counting tokens" in str(
result.error_message
):
pytest.skip(
f"Model {model} doesn't support token counting: {result.error_message}"
)
assert (
result.total_tokens > 0
), f"Token count should be > 0, got {result.total_tokens}"
assert result.tokenizer_type is not None, "tokenizer_type should be set"
assert (
result.error is not True
), f"Token counting should not error: {result.error_message}"

550
tests/llm_responses_api_testing/base_responses_api.py
View file

@ -1,550 +0,0 @@
import httpx
import json
import pytest
from typing import Any, Dict, List, Optional
from unittest.mock import MagicMock, Mock, patch
from litellm._uuid import uuid
import time
import base64
import litellm
from abc import ABC, abstractmethod
from litellm.integrations.custom_logger import CustomLogger
from litellm.types.utils import StandardLoggingPayload
from litellm.types.llms.openai import (
ResponseCompletedEvent,
ResponsesAPIResponse,
ResponseAPIUsage,
IncompleteDetails,
)
from openai.types.responses.response_create_params import (
ResponseInputParam,
)
from litellm.llms.custom_httpx.http_handler import AsyncHTTPHandler
import openai
def validate_responses_api_response(response, final_chunk: bool = False):
"""
Validate that a response from litellm.responses() or litellm.aresponses()
conforms to the expected ResponsesAPIResponse structure.
Args:
response: The response object to validate
Raises:
AssertionError: If the response doesn't match the expected structure
"""
# Validate response structure
print("response=", json.dumps(response, indent=4, default=str))
assert isinstance(
response, ResponsesAPIResponse
), "Response should be an instance of ResponsesAPIResponse"
# Required fields
assert "id" in response and isinstance(
response["id"], str
), "Response should have a string 'id' field"
assert "created_at" in response and isinstance(
response["created_at"], int
), "Response should have an integer 'created_at' field"
if response.get("status") == "completed":
assert "output" in response and isinstance(
response["output"], list
), "Response should have a list 'output' field"
# Optional fields with their expected types
optional_fields = {
"error": (dict, type(None)), # error can be dict or None
"incomplete_details": (IncompleteDetails, type(None)),
"instructions": (str, type(None)),
"metadata": dict,
"model": str,
"object": str,
"parallel_tool_calls": (bool, type(None)),
"temperature": (int, float, type(None)),
"tool_choice": (dict, str, type(None)),
"tools": (list, type(None)),
"top_p": (int, float, type(None)),
"max_output_tokens": (int, type(None)),
"previous_response_id": (str, type(None)),
"reasoning": (dict, type(None)),
"status": str,
"text": dict,
"truncation": (str, type(None)),
"usage": ResponseAPIUsage,
"user": (str, type(None)),
"store": (bool, type(None)),
}
if final_chunk is False:
optional_fields["usage"] = type(None)
for field, expected_type in optional_fields.items():
if field in response:
assert isinstance(
response[field], expected_type
), f"Field '{field}' should be of type {expected_type}, but got {type(response[field])}"
# Check if output has at least one item
if final_chunk is True and response.get("status") == "completed":
assert (
len(response["output"]) > 0
), "Response 'output' field should have at least one item"
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"""
pass
def get_advanced_model_for_shell_tool(self) -> Optional[str]:
"""If specified, overrides the model used by test_responses_api_shell_tool_streaming_sees_shell_output (e.g. openai/gpt-5.2 for shell support)."""
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:
if sync_mode:
response = litellm.responses(
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,
)
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)
async def test_basic_openai_responses_api_streaming(self, sync_mode):
litellm.turn_on_debug()
# Enable cost calculation for streaming usage
litellm.include_cost_in_streaming_usage = True
base_completion_call_args = self.get_base_completion_call_args()
collected_content_string = ""
response_completed_event = None
if sync_mode:
response = litellm.responses(
input="Basic ping", stream=True, **base_completion_call_args
)
for event in response:
print("litellm response=", json.dumps(event, indent=4, default=str))
if event.type == "response.output_text.delta":
collected_content_string += event.delta
elif event.type == "response.completed":
response_completed_event = event
else:
response = await litellm.aresponses(
input="Basic ping", stream=True, **base_completion_call_args
)
async for event in response:
print("litellm response=", json.dumps(event, indent=4, default=str))
if event.type == "response.output_text.delta":
collected_content_string += event.delta
elif event.type == "response.completed":
response_completed_event = event
# assert the response completed event is not None
assert response_completed_event is not None
# assert the response completed event has a response
assert response_completed_event.response is not None
# For async agent APIs (like Manus), the response may be in 'running' state
# without content yet - this is valid behavior
response_status = response_completed_event.response.status
if response_status in ["running", "pending"]:
# Running/pending state is acceptable - task started successfully
print(
f"Response is in '{response_status}' state - async agent API behavior"
)
assert response_completed_event.response.id is not None
else:
# For completed responses, validate content and usage
# assert the delta chunks content had len(collected_content_string) > 0
# this content is typically rendered on chat ui's
assert len(collected_content_string) > 0
# assert the response completed event includes the usage
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
)
# 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 the response completed event includes cost when include_cost_in_streaming_usage is True
assert hasattr(
response_completed_event.response.usage, "cost"
), "Cost should be included in streaming responses API usage object"
assert (
response_completed_event.response.usage.cost > 0
), "Cost should be greater than 0"
print(
f"Cost found in streaming response: {response_completed_event.response.usage.cost}"
)
# Reset the setting
litellm.include_cost_in_streaming_usage = False
@pytest.mark.parametrize("sync_mode", [False, True])
@pytest.mark.asyncio
async def test_basic_openai_responses_delete_endpoint(self, sync_mode):
litellm.turn_on_debug()
litellm.set_verbose = True
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
)
# delete the response
if isinstance(response, ResponsesAPIResponse):
litellm.delete_responses(
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
)
# async delete the response
if isinstance(response, ResponsesAPIResponse):
await litellm.adelete_responses(
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
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,
stream=True,
**base_completion_call_args,
)
for event in response:
print("litellm response=", json.dumps(event, indent=4, default=str))
if "response" in event:
response_obj = event.get("response")
if response_obj is not None:
response_id = response_obj.get("id")
print("got response_id=", response_id)
# delete the response
assert response_id is not None
litellm.delete_responses(
response_id=response_id, **base_completion_call_args
)
else:
response = await litellm.aresponses(
input="Basic ping",
max_output_tokens=20,
stream=True,
**base_completion_call_args,
)
async for event in response:
print("litellm response=", json.dumps(event, indent=4, default=str))
if "response" in event:
response_obj = event.get("response")
if response_obj is not None:
response_id = response_obj.get("id")
print("got response_id=", response_id)
# delete the response
assert response_id is not None
await litellm.adelete_responses(
response_id=response_id, **base_completion_call_args
)
@pytest.mark.parametrize("sync_mode", [False, True])
@pytest.mark.flaky(retries=3, delay=2)
@pytest.mark.asyncio
async def test_basic_openai_responses_get_endpoint(self, sync_mode):
litellm.turn_on_debug()
litellm.set_verbose = True
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
)
# get the response
if isinstance(response, ResponsesAPIResponse):
result = litellm.get_responses(
response_id=response.id, **base_completion_call_args
)
assert result is not None
assert result.id == response.id
assert result.output_text == response.output_text
else:
raise ValueError("response is not a ResponsesAPIResponse")
else:
response = await litellm.aresponses(
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
)
assert result is not None
assert result.id == response.id
assert result.output_text == response.output_text
else:
raise ValueError("response is not a ResponsesAPIResponse")
@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 ?",
},
{
"type": "function_call",
"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",
},
{
"type": "function_call_output",
"call_id": "fc_1fe70e2a-a596-45ef-b72c-9b8567c460e5",
"output": "Rainy",
},
]
# Define the tools
tools = [
{
"type": "function",
"name": "get_weather",
"description": "Get current temperature for a given location.",
"parameters": {
"type": "object",
"properties": {
"location": {
"type": "string",
"description": "City and country e.g. Bogotá, Colombia",
}
},
"required": ["location"],
"additionalProperties": False,
},
}
]
try:
# Make the responses API call
response = await litellm.aresponses(
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
# For async agent APIs (like Manus), the response may be in 'running' state
# without output yet - this is valid behavior
if response.get("status") in ["running", "pending"]:
print(
f"Response is in '{response.get('status')}' state - async agent API behavior"
)
assert response.get("id") is not None
else:
assert len(response["output"]) > 0
@pytest.mark.parametrize("sync_mode", [False, True])
@pytest.mark.flaky(retries=3, delay=2)
@pytest.mark.asyncio
async def test_basic_openai_responses_cancel_endpoint(self, sync_mode):
try:
litellm.turn_on_debug()
litellm.set_verbose = True
base_completion_call_args = self.get_base_completion_call_args()
if sync_mode:
response = litellm.responses(
input="Basic ping",
max_output_tokens=20,
background=True,
**base_completion_call_args,
)
# cancel the response
if isinstance(response, ResponsesAPIResponse):
cancel_result = litellm.cancel_responses(
response_id=response.id, **base_completion_call_args
)
assert cancel_result is not None
assert hasattr(cancel_result, "id")
# The actual response structure depends on the provider implementation
assert isinstance(cancel_result, ResponsesAPIResponse)
else:
raise ValueError("response is not a ResponsesAPIResponse")
else:
response = await litellm.aresponses(
input="Basic ping",
max_output_tokens=20,
background=True,
**base_completion_call_args,
)
# async cancel the response
if isinstance(response, ResponsesAPIResponse):
cancel_result = await litellm.acancel_responses(
response_id=response.id, **base_completion_call_args
)
assert cancel_result is not None
assert hasattr(cancel_result, "id")
# The actual response structure depends on the provider implementation
assert isinstance(cancel_result, ResponsesAPIResponse)
else:
raise ValueError("response is not a ResponsesAPIResponse")
except Exception as e:
if "Cannot cancel a completed response" in str(e):
pass
else:
raise e
@pytest.mark.asyncio
async def test_responses_api_context_management_server_side_compaction(self):
"""
E2E test for server-side compaction (context_management) on OpenAI Responses API.
Passes context_management with compact_threshold; validates that the request is
accepted and returns a valid response. Compaction may not run for short inputs.
"""
base_completion_call_args = self.get_base_completion_call_args()
model = base_completion_call_args.get("model") or ""
# Azure does not support compaction context_management (only clear_tool_results)
if "azure/" in str(model):
pytest.skip("context_management compaction is not supported on Azure")
if "openai/" not in str(model):
pytest.skip(
"context_management server-side compaction e2e is only run for OpenAI"
)
context_management = [{"type": "compaction", "compact_threshold": 200000}]
try:
response = await litellm.aresponses(
input="Short ping to verify context_management is accepted.",
max_output_tokens=20,
context_management=context_management,
**base_completion_call_args,
)
except litellm.InternalServerError:
pytest.skip("Skipping test due to litellm.InternalServerError")
validate_responses_api_response(response, final_chunk=True)
assert response.get("id") is not None
assert response.get("status") is not None
@pytest.mark.asyncio
async def test_responses_api_shell_tool(self):
"""
E2E test for Shell tool on OpenAI Responses API.
Passes tools=[{"type": "shell", "environment": {"type": "container_auto"}}];
validates that the request is accepted and returns a valid response.
Only runs for OpenAI; offline coverage for the Azure route lives in
tests/unit/responses/test_responses_api_request_body.py.
"""
base_completion_call_args = self.get_base_completion_call_args()
model = (
self.get_advanced_model_for_shell_tool()
or base_completion_call_args.get("model")
or ""
)
if "openai/" not in str(model):
pytest.skip(
"Shell tool e2e is OpenAI-only; no Azure deployment supports the shell tool yet, re-enable once one exists"
)
tools = [{"type": "shell", "environment": {"type": "container_auto"}}]
input_msg = "List files in /mnt/data and show python --version."
try:
response = await litellm.aresponses(
**{**base_completion_call_args, "model": model},
input=input_msg,
max_output_tokens=256,
tools=tools,
tool_choice="auto",
timeout=90,
)
except litellm.Timeout:
pytest.skip("Provider did not answer the shell tool request within 90s")
except litellm.InternalServerError:
pytest.skip("Skipping test due to litellm.InternalServerError")
except litellm.BadRequestError as e:
if "shell" in str(e).lower() and "not supported" in str(e).lower():
pytest.skip(
"Shell tool is not supported for this model (e.g. gpt-5.5); use a model that supports shell"
)
raise
validate_responses_api_response(response, final_chunk=True)
assert response.get("id") is not None
assert response.get("status") is not None

112
tests/llm_responses_api_testing/conftest.py
View file

@ -1,112 +0,0 @@
# conftest.py
import asyncio
import importlib
import pytest
import litellm # noqa: E402
from tests._vcr_conftest_common import ( # noqa: E402,F401
VerboseReporterState,
_pin_multipart_boundary,
apply_vcr_auto_marker_to_items,
emit_cassette_cache_session_banner,
emit_vcr_classification_summary,
emit_vcr_diagnostic_log,
install_live_call_probe,
record_vcr_outcome,
register_persister_if_enabled,
reset_vcr_diag_dir,
vcr_config_dict,
)
_verbose_state = VerboseReporterState()
@pytest.fixture(scope="module")
def vcr_config():
return vcr_config_dict()
def pytest_recording_configure(config, vcr):
register_persister_if_enabled(vcr)
@pytest.hookimpl(hookwrapper=True)
def pytest_runtest_makereport(item, call):
outcome = yield
rep = outcome.get_result()
setattr(item, f"rep_{rep.when}", rep)
@pytest.fixture(autouse=True)
def _vcr_outcome_gate(request, vcr):
install_live_call_probe(request, vcr)
yield
record_vcr_outcome(request, vcr)
def pytest_configure(config):
_verbose_state.remember_pluginmanager(config)
reset_vcr_diag_dir()
def pytest_runtest_logreport(report):
_verbose_state.maybe_emit_verdict(report)
@pytest.fixture(scope="session")
def event_loop():
try:
loop = asyncio.get_running_loop()
except RuntimeError:
loop = asyncio.new_event_loop()
yield loop
loop.close()
@pytest.fixture(scope="function", autouse=True)
def setup_and_teardown():
"""
This fixture reloads litellm before every function. To speed up testing by removing callbacks being chained.
"""
importlib.reload(litellm)
try:
if hasattr(litellm, "proxy") and hasattr(litellm.proxy, "proxy_server"):
importlib.reload(litellm.proxy.proxy_server)
except Exception as e:
print(f"Error reloading litellm.proxy.proxy_server: {e}")
loop = asyncio.get_event_loop_policy().new_event_loop()
asyncio.set_event_loop(loop)
print(litellm)
yield
# Teardown code (executes after the yield point)
loop.close() # Close the loop created earlier
asyncio.set_event_loop(None) # Remove the reference to the loop
def pytest_collection_modifyitems(config, items):
apply_vcr_auto_marker_to_items(items)
custom_logger_tests = [
item for item in items if "custom_logger" in item.parent.name
]
other_tests = [item for item in items if "custom_logger" not in item.parent.name]
custom_logger_tests.sort(key=lambda x: x.name)
other_tests.sort(key=lambda x: x.name)
items[:] = custom_logger_tests + other_tests
def pytest_terminal_summary(terminalreporter, exitstatus, config):
emit_cassette_cache_session_banner(terminalreporter)
emit_vcr_classification_summary(terminalreporter)
emit_vcr_diagnostic_log(terminalreporter)

90
tests/llm_responses_api_testing/test_anthropic_responses_api.py
View file

@ -1,90 +0,0 @@
import litellm
from base_responses_api import BaseResponsesAPITest
from openai.types.responses.function_tool import FunctionTool
class TestAnthropicResponsesAPITest(BaseResponsesAPITest):
test_basic_openai_responses_delete_endpoint = None
test_basic_openai_responses_streaming_delete_endpoint = None
test_basic_openai_responses_get_endpoint = None
test_basic_openai_responses_cancel_endpoint = None
def get_base_completion_call_args(self):
# litellm.turn_on_debug()
return {
"model": "anthropic/claude-sonnet-4-5",
}
def test_multiturn_tool_calls():
# Test streaming response with tools for Anthropic
litellm.turn_on_debug()
shell_tool = dict(
FunctionTool(
type="function",
name="shell",
description="Runs a shell command, and returns its output.",
parameters={
"type": "object",
"properties": {
"command": {"type": "array", "items": {"type": "string"}},
"workdir": {
"type": "string",
"description": "The working directory for the command.",
},
},
"required": ["command"],
},
strict=True,
)
)
# Step 1: Initial request with the tool
response = litellm.responses(
input=[
{
"role": "user",
"content": [
{"type": "input_text", "text": "make a hello world html file"}
],
"type": "message",
}
],
model="anthropic/claude-haiku-4-5-20251001",
instructions="You are a helpful coding assistant.",
tools=[shell_tool],
)
print("response=", response)
# Step 2: Send the results of the tool call back to the model
# Get the response ID and tool call ID from the response
response_id = response.id
tool_call_id = None
for item in response.output:
if hasattr(item, "type") and item.type == "function_call":
tool_call_id = getattr(item, "call_id", None)
if tool_call_id:
break
# Validate that we got a tool call with a valid call_id
if not tool_call_id:
raise AssertionError(
f"Expected a function_call with a valid call_id in response.output, but got: {response.output}"
)
# Use await with asyncio.run for the async function
follow_up_response = litellm.responses(
model="anthropic/claude-haiku-4-5-20251001",
previous_response_id=response_id,
input=[
{
"type": "function_call_output",
"call_id": tool_call_id,
"output": '{"output":"<html>\\n<head>\\n <title>Hello Page</title>\\n</head>\\n<body>\\n <h1>Hi</h1>\\n <p>Welcome to this simple webpage!</p>\\n</body>\\n</html> > index.html\\n","metadata":{"exit_code":0,"duration_seconds":0}}',
}
],
tools=[shell_tool],
)
print("follow_up_response=", follow_up_response)

34
tests/llm_responses_api_testing/test_azure_responses_api.py
View file

@ -1,34 +0,0 @@
import os
import pytest
import litellm
from base_responses_api import BaseResponsesAPITest
class TestAzureResponsesAPITest(BaseResponsesAPITest):
test_responses_api_with_tool_calls = None
def get_base_completion_call_args(self):
return {
"model": "azure/gpt-4.1-mini",
"truncation": "auto",
"api_base": os.getenv("AZURE_AI_API_BASE"),
"api_key": os.getenv("AZURE_AI_API_KEY"),
"api_version": "2025-03-01-preview",
}
@pytest.mark.asyncio
async def test_azure_responses_api_preview_api_version():
"""
Ensure new azure preview api version is working
"""
litellm.turn_on_debug()
response = await litellm.aresponses(
model="azure/gpt-5-mini",
truncation="auto",
api_version="preview",
api_base=os.getenv("AZURE_AI_API_BASE"),
api_key=os.getenv("AZURE_AI_API_KEY"),
input="Hello, can you tell me a short joke?",
)

12
tests/llm_responses_api_testing/test_google_ai_studio_responses_api.py
View file

@ -1,12 +0,0 @@
from base_responses_api import BaseResponsesAPITest
class TestGoogleAIStudioResponsesAPITest(BaseResponsesAPITest):
test_basic_openai_responses_delete_endpoint = None
test_basic_openai_responses_streaming_delete_endpoint = None
test_basic_openai_responses_get_endpoint = None
test_basic_openai_responses_cancel_endpoint = None
def get_base_completion_call_args(self):
# litellm.turn_on_debug()
return {"model": "gemini/gemini-2.5-flash-lite"}

64
tests/llm_responses_api_testing/test_openai_responses_api.py
View file

@ -1,64 +0,0 @@
import json
import pytest
from base_responses_api import BaseResponsesAPITest
import litellm
class TestOpenAIResponsesAPITest(BaseResponsesAPITest):
test_responses_api_with_tool_calls = None
def get_base_completion_call_args(self):
return {
"model": "openai/gpt-5.5",
}
def get_advanced_model_for_shell_tool(self):
return "openai/gpt-5.2"
@pytest.mark.asyncio
@pytest.mark.parametrize("sync_mode", [True, False])
async def test_openai_compact_responses_api(sync_mode):
"""
Test the compact_responses API for OpenAI.
This test verifies that the compact_responses endpoint works correctly
for compressing conversation history.
"""
litellm.turn_on_debug()
litellm.set_verbose = True
input_messages = [
{"role": "user", "content": "Hello, how are you?"},
{"role": "assistant", "content": "I'm doing well, thank you for asking!"},
{"role": "user", "content": "What is the weather like today?"},
]
try:
if sync_mode:
response = litellm.compact_responses(
model="openai/gpt-5.5",
input=input_messages,
instructions="Be helpful and concise",
)
else:
response = await litellm.acompact_responses(
model="openai/gpt-5.5",
input=input_messages,
instructions="Be helpful and concise",
)
except litellm.InternalServerError:
pytest.skip("Skipping test due to InternalServerError")
except litellm.BadRequestError as e:
# compact_responses may not be available for all models/accounts
pytest.skip(f"Skipping test due to BadRequestError: {e}")
print("compact_responses response=", json.dumps(response, indent=4, default=str))
# Validate response structure
assert response is not None
assert "id" in response, "Response should have an 'id' field"
assert "output" in response, "Response should have an 'output' field"
assert isinstance(response["output"], list), "Output should be a list"

1
tests/llm_translation/Readme.md
View file

@ -26,7 +26,6 @@ provider APIs. The reusable conftest plumbing lives in
`tests/_vcr_conftest_common.py` and is wired into:
- `tests/llm_translation/`
- `tests/llm_responses_api_testing/`
- `tests/audio_tests/`
- `tests/batches_tests/`
- `tests/guardrails_tests/`

19
tests/llm_translation/base_llm_unit_tests.py
View file

@ -1564,25 +1564,6 @@ class BaseAnthropicChatTest(ABC):
print(response)
def test_completion_with_thinking_basic(self):
litellm.turn_on_debug()
base_completion_call_args = self.get_base_completion_call_args_with_thinking()
messages = [{"role": "user", "content": "Generate 5 question + answer pairs"}]
response = self.completion_function(
**base_completion_call_args,
messages=messages,
)
print(f"response: {response}")
assert response.choices[0].message.reasoning_content is not None
assert isinstance(response.choices[0].message.reasoning_content, str)
assert response.choices[0].message.thinking_blocks is not None
assert isinstance(response.choices[0].message.thinking_blocks, list)
assert len(response.choices[0].message.thinking_blocks) > 0
assert response.choices[0].message.thinking_blocks[0]["signature"] is not None
def test_anthropic_thinking_output_stream(self):
# litellm.set_verbose = True
try:

138
tests/llm_translation/realtime/test_openai_realtime.py
View file

@ -1,138 +0,0 @@
import os
import pytest
from websockets.exceptions import ConnectionClosedError, ConnectionClosedOK
import litellm
@pytest.mark.asyncio
@pytest.mark.skipif(
os.environ.get("OPENAI_API_KEY", None) is None,
reason="No OpenAI API key provided",
)
async def test_openai_realtime_direct_call_no_intent():
"""
End-to-end test calling the actual OpenAI realtime endpoint via LiteLLM SDK
without intent parameter. This should succeed without "Invalid intent" error.
Uses real websocket connection to OpenAI.
"""
import asyncio
import json
class RealTimeWebSocketClient:
def __init__(self):
self.messages_sent = []
self.messages_received = []
self.received_session_created = False
self.connection_successful = False
self._receive_called = False
self.close_code = None
self.close_reason = None
async def accept(self):
pass
async def send_text(self, message):
self.messages_sent.append(message)
try:
if isinstance(message, bytes):
message_str = message.decode("utf-8")
else:
message_str = message
msg_data = json.loads(message_str)
msg_type = msg_data.get("type", "unknown")
if msg_type == "error":
error_info = msg_data.get("error", {})
error_code = error_info.get("code", "unknown")
error_message = error_info.get("message", "unknown")
# Don't fail on error, just record it - some errors are expected
self.messages_received.append(msg_data)
return
if msg_type == "session.created" and not self.received_session_created:
self.messages_received.append(msg_data)
self.received_session_created = True
self.connection_successful = True
except (json.JSONDecodeError, UnicodeDecodeError):
# Non-JSON messages are acceptable
pass
async def receive_text(self):
if not self._receive_called:
self._receive_called = True
max_wait = 60.0
check_interval = 0.1
waited = 0.0
while waited < max_wait:
if self.connection_successful:
break
await asyncio.sleep(check_interval)
waited += check_interval
if not self.connection_successful:
await asyncio.sleep(3.0)
raise ConnectionClosedOK(None, None)
async def close(self, code=1000, reason=""):
self.close_code = code
self.close_reason = reason
@property
def headers(self):
return {}
websocket_client = RealTimeWebSocketClient()
caught_exception = None
try:
await litellm._arealtime(
# OpenAI shut down the gpt-4o-realtime-preview family (incl. the
# undated alias) on 2026-05-07; gpt-realtime is the GA successor.
model="openai/gpt-realtime",
websocket=websocket_client,
api_key=os.environ.get("OPENAI_API_KEY"),
timeout=60,
)
except (ConnectionClosedOK, ConnectionClosedError):
pass
except Exception as e:
caught_exception = e
if "invalid_intent" in str(e).lower():
pytest.fail(f"Still getting invalid intent error: {e}")
# Other exceptions are recorded but don't fail immediately
# Build detailed error message for debugging
error_details = []
error_details.append(f"messages_sent count: {len(websocket_client.messages_sent)}")
error_details.append(
f"messages_received count: {len(websocket_client.messages_received)}"
)
error_details.append(f"close_code: {websocket_client.close_code}")
error_details.append(f"close_reason: {websocket_client.close_reason}")
if caught_exception:
error_details.append(
f"exception: {type(caught_exception).__name__}: {caught_exception}"
)
assert (
websocket_client.connection_successful
), f"Failed to establish connection. Debug info: {'; '.join(error_details)}"
assert (
websocket_client.received_session_created
), "Did not receive session.created response"
assert len(websocket_client.messages_received) > 0, "No messages received"
session_message = websocket_client.messages_received[0]
assert (
session_message["type"] == "session.created"
), f"Expected session.created, got {session_message.get('type')}"
assert (
"session" in session_message
), "session.created response missing session object"
assert "id" in session_message["session"], "Session object missing id field"
assert "model" in session_message["session"], "Session object missing model field"

28
tests/llm_translation/test_anthropic_completion.py
View file

@ -494,34 +494,6 @@ def test_anthropic_citations_api_streaming():
assert has_citations
@pytest.mark.parametrize(
"model",
[
"anthropic/claude-sonnet-4-5-20250929",
"bedrock/us.anthropic.claude-sonnet-4-5-20250929-v1:0",
],
)
def test_anthropic_thinking_output(model):
litellm.turn_on_debug()
resp = completion(
model=model,
messages=[{"role": "user", "content": "What is the capital of France?"}],
thinking={"type": "enabled", "budget_tokens": 1024},
)
print(resp)
assert resp.choices[0].message.reasoning_content is not None
assert isinstance(resp.choices[0].message.reasoning_content, str)
assert resp.choices[0].message.thinking_blocks is not None
assert isinstance(resp.choices[0].message.thinking_blocks, list)
assert len(resp.choices[0].message.thinking_blocks) > 0
assert resp.choices[0].message.thinking_blocks[0]["type"] == "thinking"
assert resp.choices[0].message.thinking_blocks[0]["signature"] is not None
@pytest.mark.parametrize(
"model",
[

79
tests/llm_translation/test_bedrock_completion.py
View file

@ -279,85 +279,6 @@ def test_bedrock_system_prompt(system, model):
pytest.fail(f"Error occurred: {e}")
def test_bedrock_claude_3_tool_calling():
try:
litellm.set_verbose = True
litellm.turn_on_debug()
tools = [
{
"type": "function",
"function": {
"name": "get_current_weather",
"description": "Get the current weather in a given location",
"parameters": {
"type": "object",
"properties": {
"location": {
"type": "string",
"description": "The city and state, e.g. San Francisco, CA",
},
"unit": {
"type": "string",
"enum": ["celsius", "fahrenheit"],
},
},
"required": ["location"],
},
},
}
]
messages = [
{
"role": "user",
"content": "What's the weather like in Boston today in fahrenheit?",
}
]
response: ModelResponse = completion(
model="bedrock/us.anthropic.claude-sonnet-4-5-20250929-v1:0",
messages=messages,
tools=tools,
tool_choice="auto",
) # type: ignore
print(f"response: {response}")
# Add any assertions here to check the response
assert isinstance(response.choices[0].message.tool_calls[0].function.name, str)
assert isinstance(
response.choices[0].message.tool_calls[0].function.arguments, str
)
messages.append(
response.choices[0].message.model_dump()
) # Add assistant tool invokes
tool_result = (
'{"location": "Boston", "temperature": "72", "unit": "fahrenheit"}'
)
# Add user submitted tool results in the OpenAI format
messages.append(
{
"tool_call_id": response.choices[0].message.tool_calls[0].id,
"role": "tool",
"name": response.choices[0].message.tool_calls[0].function.name,
"content": tool_result,
}
)
# In the second response, Claude should deduce answer from tool results
second_response = completion(
model="bedrock/us.anthropic.claude-sonnet-4-5-20250929-v1:0",
messages=messages,
tools=tools,
tool_choice="auto",
)
print(f"second response: {second_response}")
assert isinstance(second_response.choices[0].message.content, str)
except RateLimitError:
pass
except Exception as e:
pytest.fail(f"Error occurred: {e}")
def test_completion_bedrock_mistral_completion_auth():
print("calling bedrock mistral completion params auth")

53
tests/local_testing/test_caching.py
View file

@ -2277,56 +2277,3 @@ def test_caching_thinking_args_hit(): # test in memory cache
except Exception as e:
print(f"error occurred: {traceback.format_exc()}")
pytest.fail(f"Error occurred: {e}")
@pytest.mark.asyncio
async def test_cache_key_in_hidden_params_acompletion():
"""
Test that cache_key is present in _hidden_params on cache hits for acompletion.
Validates fix for missing x-litellm-cache-key header on proxy cache hits.
"""
litellm.cache = Cache(
type="redis",
host=os.environ["REDIS_HOST"],
port=os.environ["REDIS_PORT"],
password=os.environ["REDIS_PASSWORD"],
)
unique_content = f"test cache key hidden params {uuid.uuid4()}"
messages = [{"role": "user", "content": unique_content}]
# First call - cache miss
response1 = await litellm.acompletion(
model="gpt-3.5-turbo",
messages=messages,
mock_response="test response",
caching=True,
)
print(f"Response 1 _hidden_params: {response1._hidden_params}")
assert response1._hidden_params.get("cache_hit") is not True
await asyncio.sleep(0.5)
# Second call - cache hit
response2 = await litellm.acompletion(
model="gpt-3.5-turbo",
messages=messages,
mock_response="test response",
caching=True,
)
print(f"Response 2 _hidden_params: {response2._hidden_params}")
# Verify cache hit occurred
assert response2._hidden_params.get("cache_hit") is True
# Verify cache_key is present in _hidden_params
assert "cache_key" in response2._hidden_params
assert response2._hidden_params["cache_key"] is not None
# Verify both responses have same ID (cache hit)
assert response1.id == response2.id
litellm.cache = None

41
tests/local_testing/test_completion.py
View file

@ -1413,28 +1413,6 @@ def test_replicate_custom_prompt_dict():
# test_completion_together_ai_mixtral()
def test_completion_together_ai_llama():
litellm.set_verbose = True
model_name = "together_ai/meta-llama/Llama-3.3-70B-Instruct-Turbo"
try:
messages = [
{"role": "user", "content": "What llm are you?"},
]
response = completion(model=model_name, messages=messages, max_tokens=5)
# Add any assertions here to check the response
print(response)
cost = completion_cost(completion_response=response)
assert cost > 0.0
print(
"Cost for completion call together-computer/llama-2-70b: ",
f"${float(cost):.10f}",
)
except litellm.Timeout as e:
pass
except Exception as e:
pytest.fail(f"Error occurred: {e}")
# test_completion_together_ai_yi_chat()
@ -1903,25 +1881,6 @@ def test_langfuse_completion(monkeypatch):
)
def test_deepseek_reasoning_content_completion():
try:
litellm.set_verbose = True
litellm.turn_on_debug()
resp = litellm.completion(
timeout=5,
model="deepseek/deepseek-reasoner",
messages=[{"role": "user", "content": "Tell me a joke."}],
)
assert resp.choices[0].message.reasoning_content is not None
except litellm.Timeout:
pytest.skip("Model is timing out")
def test_qwen_text_completion():
# litellm.turn_on_debug()
resp = litellm.completion(

48
tests/local_testing/test_custom_logger.py
View file

@ -365,54 +365,6 @@ async def test_async_custom_handler_embedding_optional_param():
# asyncio.run(test_async_custom_handler_embedding_optional_param())
@pytest.mark.asyncio
@pytest.mark.flaky(retries=3, delay=1)
async def test_cost_tracking_with_caching():
"""
Important Test - This tests if that cost is 0 for cached responses
"""
from litellm import Cache
litellm.set_verbose = True
litellm.cache = Cache(
type="redis",
host=os.environ["REDIS_HOST"],
port=os.environ["REDIS_PORT"],
password=os.environ["REDIS_PASSWORD"],
)
customHandler_optional_params = MyCustomHandler()
litellm.callbacks = [customHandler_optional_params]
messages = [
{
"role": "user",
"content": f"write a one sentence poem about: {time.time()}",
}
]
response1 = await litellm.acompletion(
model="gpt-3.5-turbo",
messages=messages,
max_tokens=40,
temperature=0.2,
caching=True,
mock_response="Hey, i'm doing well!",
)
await asyncio.sleep(3) # success callback is async
response_cost = customHandler_optional_params.response_cost
assert response_cost > 0
response2 = await litellm.acompletion(
model="gpt-3.5-turbo",
messages=messages,
max_tokens=40,
temperature=0.2,
caching=True,
)
await asyncio.sleep(1) # success callback is async
response_cost_2 = customHandler_optional_params.response_cost
assert response_cost_2 == 0
@pytest.mark.flaky(retries=3, delay=3)
def test_redis_cache_completion_stream():
# Important Test - This tests if we can add to streaming cache, when custom callbacks are set

57
tests/local_testing/test_lunary.py
View file

@ -1,57 +0,0 @@
import io
import litellm
litellm.failure_callback = ["lunary"]
litellm.success_callback = ["lunary"]
litellm.set_verbose = True
def test_lunary_with_tools():
import litellm
messages = [
{
"role": "user",
"content": "What's the weather like in San Francisco, Tokyo, and Paris?",
}
]
tools = [
{
"type": "function",
"function": {
"name": "get_current_weather",
"description": "Get the current weather in a given location",
"parameters": {
"type": "object",
"properties": {
"location": {
"type": "string",
"description": "The city and state, e.g. San Francisco, CA",
},
"unit": {"type": "string", "enum": ["celsius", "fahrenheit"]},
},
"required": ["location"],
},
},
}
]
response = litellm.completion(
model="gpt-6-luna",
messages=messages,
tools=tools,
tool_choice="auto", # auto is default, but we'll be explicit
)
response_message = response.choices[0].message
assert response.choices[0].message.tool_calls
assert all(call.function.name == "get_current_weather" for call in response.choices[0].message.tool_calls)
print("\nLLM Response:\n", response.choices[0].message)

13
tests/local_testing/test_text_completion.py
View file

@ -2729,19 +2729,6 @@ def test_completion_openai_engine() -> None:
# test_completion_openai_engine()
def test_completion_chatgpt_prompt():
try:
print("\n gpt3.5 test\n")
response = text_completion(model="openai/gpt-3.5-turbo", prompt="What's the weather in SF?")
print(response)
response_str = response["choices"][0]["text"]
print("\n", response.choices)
print("\n", response.choices[0])
# print(response.choices[0].text)
except Exception as e:
pytest.fail(f"Error occurred: {e}")
# test_completion_chatgpt_prompt()

51
tests/pass_through_tests/test_gemini_with_spend.test.js
View file

@ -24,57 +24,6 @@ global.fetch = async function patchedFetch(url, options) {
jest.retryTimes(3);
describe('Gemini AI Tests', () => {
test('should successfully generate non-streaming content with tags', async () => {
const genAI = new GoogleGenerativeAI(masterKey);
const requestOptions = {
baseUrl: 'http://127.0.0.1:4000/gemini',
customHeaders: {
"tags": "gemini-js-sdk,pass-through-endpoint"
}
};
const model = genAI.getGenerativeModel({
model: 'gemini-3.1-flash-lite'
}, requestOptions);
const prompt = 'Say "hello test" and nothing else';
const result = await model.generateContent(prompt);
expect(result).toBeDefined();
// Use the captured callId
const callId = lastCallId;
console.log("Captured Call ID:", callId);
// Poll for spend data with retries (DB writes can be slow in CI)
let spendData = null;
for (let attempt = 0; attempt < 6; attempt++) {
await new Promise(resolve => setTimeout(resolve, 10000));
const spendResponse = await fetch(
`http://127.0.0.1:4000/spend/logs?request_id=${callId}`,
{ headers: { 'Authorization': `Bearer ${masterKey}` } }
);
spendData = await spendResponse.json();
console.log(`spendData (attempt ${attempt + 1}):`, spendData);
if (spendData && spendData.length > 0 && spendData[0] && spendData[0].request_id) break;
}
if (!spendData || !spendData.length || !spendData[0] || !spendData[0].request_id) {
console.warn('Spend data not available after polling - skipping spend assertions (DB write may be slow in CI)');
return;
}
expect(spendData).toBeDefined();
expect(spendData[0].request_id).toBe(callId);
expect(spendData[0].call_type).toBe('pass_through_endpoint');
expect(spendData[0].request_tags).toEqual(['gemini-js-sdk', 'pass-through-endpoint']);
expect(spendData[0].metadata).toHaveProperty('user_api_key');
expect(spendData[0].model).toContain('gemini');
expect(spendData[0].custom_llm_provider).toBe('gemini');
expect(spendData[0].spend).toBeGreaterThan(0);
}, 90000);
test('should successfully generate streaming content with tags', async () => {
const genAI = new GoogleGenerativeAI(masterKey);

78
tests/pass_through_unit_tests/test_anthropic_messages_passthrough.py
View file

@ -1,13 +1,9 @@
import json
import os
from datetime import datetime
from typing import Dict, Any
import asyncio
import unittest.mock
from unittest.mock import MagicMock
import litellm
import pytest
from dotenv import load_dotenv
from litellm.llms.anthropic.pass_through.messages.handler import (
anthropic_messages,
@ -16,47 +12,12 @@ from litellm.llms.anthropic.pass_through.messages.handler import (
from typing import Optional
from litellm.types.utils import StandardLoggingPayload
from litellm.integrations.custom_logger import CustomLogger
from litellm.router import Router
import importlib
from base_anthropic_unified_messages_test import BaseAnthropicMessagesTest
# Load environment variables
load_dotenv()
@pytest.fixture(scope="session")
def event_loop():
"""Create an instance of the default event loop for each test session."""
loop = asyncio.get_event_loop_policy().new_event_loop()
yield loop
loop.close()
@pytest.fixture(scope="function", autouse=True)
def setup_and_teardown(event_loop): # Add event_loop as a dependency
curr_dir = os.getcwd()
import litellm
from litellm import Router
importlib.reload(litellm)
# Set the event loop from the fixture
asyncio.set_event_loop(event_loop)
print(litellm)
yield
# Clean up any pending tasks
pending = asyncio.all_tasks(event_loop)
for task in pending:
task.cancel()
# Run the event loop until all tasks are cancelled
if pending:
event_loop.run_until_complete(asyncio.gather(*pending, return_exceptions=True))
def _validate_anthropic_response(response: Dict[str, Any]):
assert "id" in response
assert "content" in response
@ -117,42 +78,3 @@ class TestAnthropicOpenAIAPI(BaseAnthropicMessagesTest):
This is the model name that is expected to be in the logging payload
"""
return "gpt-4.1-mini"
@pytest.mark.asyncio
async def test_anthropic_messages_litellm_router_non_streaming():
"""
Test the anthropic_messages with non-streaming request
"""
litellm.turn_on_debug()
router = Router(
model_list=[
{
"model_name": "claude-special-alias",
"litellm_params": {
"model": "claude-haiku-4-5-20251001",
"api_key": os.getenv("ANTHROPIC_API_KEY"),
},
}
]
)
# Set up test parameters
messages = [{"role": "user", "content": "Hello, can you tell me a short joke?"}]
# Call the handler
response = await router.aanthropic_messages(
messages=messages,
model="claude-special-alias",
max_tokens=100,
)
# Verify response
assert "id" in response
assert "content" in response
assert "model" in response
assert response["role"] == "assistant"
print(f"Non-streaming response: {json.dumps(response, indent=2)}")
return response

10
tests/test_end_users.py
View file

@ -109,16 +109,6 @@ async def new_budget(session, i, budget_id=None):
print()
@pytest.mark.asyncio
async def test_end_user_new():
"""
Make 20 parallel calls to /user/new. Assert all worked.
"""
async with aiohttp.ClientSession() as session:
tasks = [new_end_user(session, i, str(uuid.uuid4())) for i in range(1, 11)]
await asyncio.gather(*tasks)
@pytest.mark.asyncio
async def test_enduser_tpm_limits_non_master_key():
"""

68
tests/test_keys.py
View file

@ -183,33 +183,6 @@ async def chat_completion(session, key, model="gpt-4"):
pass
async def chat_completion_streaming(session, key, model="gpt-4"):
client = AsyncOpenAI(api_key=key, base_url="http://0.0.0.0:4000")
messages = [
{"role": "system", "content": "You are a helpful assistant"},
{"role": "user", "content": "Hello!"},
]
prompt_tokens = litellm.token_counter(model="gpt-35-turbo", messages=messages)
data = {
"model": model,
"messages": messages,
"stream": True,
}
response = await client.chat.completions.create(**data)
content = ""
async for chunk in response:
content += chunk.choices[0].delta.content or ""
print(f"content: {content}")
completion_tokens = litellm.token_counter(
model="gpt-35-turbo", text=content, count_response_tokens=True
)
return prompt_tokens, completion_tokens
async def delete_key(session, get_key, auth_key=os.environ["LITELLM_MASTER_KEY"]):
"""
Delete key
@ -367,47 +340,6 @@ async def get_spend_logs(session, request_id):
return await response.json()
@pytest.mark.asyncio
@pytest.mark.flaky(retries=6, delay=2)
@pytest.mark.skip(
reason="Temporarily skipping due to model change. Will be updated soon."
)
async def test_aaaaakey_info_spend_values_streaming():
"""
Test to ensure spend is correctly calculated.
- create key
- make completion call
- assert cost is expected value
"""
async with aiohttp.ClientSession() as session:
## streaming - azure
key_gen = await generate_key(session=session, i=0)
new_key = key_gen["key"]
prompt_tokens, completion_tokens = await chat_completion_streaming(
session=session, key=new_key
)
print(f"prompt_tokens: {prompt_tokens}, completion_tokens: {completion_tokens}")
prompt_cost, completion_cost = litellm.cost_per_token(
model="azure/gpt-4o",
prompt_tokens=prompt_tokens,
completion_tokens=completion_tokens,
)
response_cost = prompt_cost + completion_cost
await asyncio.sleep(8) # allow db log to be updated
print(f"new_key: {new_key}")
key_info = await get_key_info(
session=session, get_key=new_key, call_key=new_key
)
print(
f"response_cost: {response_cost}; key_info spend: {key_info['info']['spend']}"
)
rounded_response_cost = round(response_cost, 8)
rounded_key_info_spend = round(key_info["info"]["spend"], 8)
assert (
rounded_response_cost == rounded_key_info_spend
), f"Expected={rounded_response_cost}, Got={rounded_key_info_spend}"
@pytest.mark.skip(reason="Frequent check on ci/cd leads to read timeout issue.")
@pytest.mark.asyncio
async def test_key_with_budgets():

174
tests/test_new_vector_store_endpoints.py
View file

@ -13,82 +13,6 @@ import litellm
from litellm.proxy._types import UserAPIKeyAuth
@pytest.mark.asyncio
async def test_vector_store_retrieve_basic():
"""Test basic vector store retrieve functionality."""
mock_response = {
"id": "vs_test123",
"object": "vector_store",
"created_at": 1699061776,
"name": "Test Vector Store",
"file_counts": {
"in_progress": 0,
"completed": 5,
"failed": 0,
"cancelled": 0,
"total": 5,
},
"status": "completed",
"usage_bytes": 12345,
}
with patch(
"litellm.vector_stores.main.aretrieve",
new=AsyncMock(return_value=mock_response),
) as mock_retrieve:
router = litellm.Router(model_list=[])
result = await router.avector_store_retrieve(
vector_store_id="vs_test123",
custom_llm_provider="openai",
)
assert result["id"] == "vs_test123"
assert result["object"] == "vector_store"
assert result["status"] == "completed"
mock_retrieve.assert_called_once()
@pytest.mark.asyncio
async def test_vector_store_list_basic():
"""Test basic vector store list functionality."""
mock_response = {
"object": "list",
"data": [
{
"id": "vs_test1",
"object": "vector_store",
"created_at": 1699061776,
"name": "Store 1",
},
{
"id": "vs_test2",
"object": "vector_store",
"created_at": 1699061777,
"name": "Store 2",
},
],
"first_id": "vs_test1",
"last_id": "vs_test2",
"has_more": False,
}
with patch(
"litellm.vector_stores.main.alist",
new=AsyncMock(return_value=mock_response),
) as mock_list:
router = litellm.Router(model_list=[])
result = await router.avector_store_list(
limit=20,
order="desc",
custom_llm_provider="openai",
)
assert result["object"] == "list"
assert len(result["data"]) == 2
assert result["data"][0]["id"] == "vs_test1"
mock_list.assert_called_once()
@pytest.mark.asyncio
async def test_vector_store_update_basic():
"""Test basic vector store update functionality."""
@ -119,76 +43,6 @@ async def test_vector_store_update_basic():
mock_update.assert_called_once()
@pytest.mark.asyncio
async def test_vector_store_delete_basic():
"""Test basic vector store delete functionality."""
mock_response = {
"id": "vs_test123",
"object": "vector_store.deleted",
"deleted": True,
}
with patch(
"litellm.vector_stores.main.adelete",
new=AsyncMock(return_value=mock_response),
) as mock_delete:
router = litellm.Router(model_list=[])
result = await router.avector_store_delete(
vector_store_id="vs_test123",
custom_llm_provider="openai",
)
assert result["id"] == "vs_test123"
assert result["deleted"] is True
assert result["object"] == "vector_store.deleted"
mock_delete.assert_called_once()
@pytest.mark.asyncio
async def test_async_vector_store_retrieve():
"""Test async vector store retrieve."""
mock_response = {
"id": "vs_async123",
"object": "vector_store",
"name": "Async Test Store",
}
with patch(
"litellm.vector_stores.main.aretrieve",
new=AsyncMock(return_value=mock_response),
) as mock_aretrieve:
router = litellm.Router(model_list=[])
result = await router.avector_store_retrieve(
vector_store_id="vs_async123",
custom_llm_provider="openai",
)
assert result["id"] == "vs_async123"
mock_aretrieve.assert_called_once()
@pytest.mark.asyncio
async def test_async_vector_store_list():
"""Test async vector store list."""
mock_response = {
"object": "list",
"data": [{"id": "vs_1"}, {"id": "vs_2"}],
}
with patch(
"litellm.vector_stores.main.alist",
new=AsyncMock(return_value=mock_response),
) as mock_alist:
router = litellm.Router(model_list=[])
result = await router.avector_store_list(
limit=10,
custom_llm_provider="openai",
)
assert len(result["data"]) == 2
mock_alist.assert_called_once()
@pytest.mark.asyncio
async def test_async_vector_store_update():
"""Test async vector store update."""
@ -212,28 +66,6 @@ async def test_async_vector_store_update():
mock_aupdate.assert_called_once()
@pytest.mark.asyncio
async def test_async_vector_store_delete():
"""Test async vector store delete."""
mock_response = {
"id": "vs_async123",
"deleted": True,
}
with patch(
"litellm.vector_stores.main.adelete",
new=AsyncMock(return_value=mock_response),
) as mock_adelete:
router = litellm.Router(model_list=[])
result = await router.avector_store_delete(
vector_store_id="vs_async123",
custom_llm_provider="openai",
)
assert result["deleted"] is True
mock_adelete.assert_called_once()
@pytest.mark.asyncio
async def test_vector_store_list_with_pagination():
"""Test vector store list with pagination parameters."""
@ -335,18 +167,12 @@ if __name__ == "__main__":
# Test basic sync operations
print("✓ Testing basic sync operations...")
asyncio.run(test_vector_store_retrieve_basic())
asyncio.run(test_vector_store_list_basic())
asyncio.run(test_vector_store_update_basic())
asyncio.run(test_vector_store_delete_basic())
print("✓ Basic sync operations successful")
# Test async operations
print("✓ Testing async operations...")
asyncio.run(test_async_vector_store_retrieve())
asyncio.run(test_async_vector_store_list())
asyncio.run(test_async_vector_store_update())
asyncio.run(test_async_vector_store_delete())
print("✓ Async operations successful")
print("\n✅ All smoke tests passed!")

12
tests/test_team_members.py
View file

@ -99,18 +99,6 @@ def verify_member_in_team(team_info: Dict, user_email: str) -> bool:
)
def test_team_creation(api_client):
"""Test team creation"""
team_alias = f"Test Team {uuid.uuid4().hex[:6]}"
team_response, team_id = api_client.create_team(team_alias)
# Verify team was created
team_info = api_client.get_team_info(team_id)
assert team_info["team_id"] == team_id
assert team_info["team_info"]["team_alias"] == team_alias
assert "o3-mini" in team_info["team_info"]["models"]
def test_add_single_member(api_client, new_team):
"""Test adding a single member to a new team"""
# Get initial team info