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

test: migrate huggingface, json and openrouter translation tests

Browse source
This commit is contained in:
yuneng 2026-10-10 05:20:20 +00:00
parent 2334ae25d0
commit c2350d7dcc
13 changed files with 789 additions and 657 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

10
tests/integration/proxy_config.yaml
View file

@ -349,6 +349,16 @@ model_list:
model: openrouter/typesafe/jev-1.13
api_base: http://127.0.0.1:8191
api_key: synthetic-openrouter-key
- model_name: openrouter/anthropic/claude-sonnet-4
litellm_params:
model: openrouter/anthropic/claude-sonnet-4
api_base: http://127.0.0.1:8191
api_key: synthetic-openrouter-key
- model_name: openrouter/google/gemini-3.1-flash-image
litellm_params:
model: openrouter/google/gemini-3.1-flash-image
api_base: http://127.0.0.1:8191
api_key: synthetic-openrouter-key
- model_name: cloudflare/@cf/cloudflare/clef
litellm_params:
model: cloudflare/@cf/cloudflare/clef

131
tests/integration/translation/chat_completions/bases/openrouter.py Normal file
View file

@ -0,0 +1,131 @@
from pathlib import Path
from typing import Final
from integration._support.client import JSON_OBJECT
from integration.translation.case import TranslationTestCase
CLAUDE_SONNET_4_RESPONSE: Final = JSON_OBJECT.validate_json(
Path(__file__).with_name("openrouter_claude_reasoning_response.json").read_bytes()
)
CLAUDE_SONNET_4_CHOICE: Final = JSON_OBJECT.validate_python(CLAUDE_SONNET_4_RESPONSE["choices"][0])
CLAUDE_SONNET_4_MESSAGE: Final = JSON_OBJECT.validate_python(CLAUDE_SONNET_4_CHOICE["message"])
CLAUDE_SONNET_4_USAGE: Final = JSON_OBJECT.validate_python(CLAUDE_SONNET_4_RESPONSE["usage"])
GEMINI_3_1_FLASH_IMAGE_RESPONSE: Final = JSON_OBJECT.validate_json(
Path(__file__).with_name("openrouter_gemini_image_response.json").read_bytes()
)
GEMINI_3_1_FLASH_IMAGE_CHOICE: Final = JSON_OBJECT.validate_python(GEMINI_3_1_FLASH_IMAGE_RESPONSE["choices"][0])
GEMINI_3_1_FLASH_IMAGE_MESSAGE: Final = JSON_OBJECT.validate_python(GEMINI_3_1_FLASH_IMAGE_CHOICE["message"])
GEMINI_3_1_FLASH_IMAGE_USAGE: Final = JSON_OBJECT.validate_python(GEMINI_3_1_FLASH_IMAGE_RESPONSE["usage"])
CLAUDE_SONNET_4_TEST_CASE: Final = TranslationTestCase(
scenario="reasoning_content",
litellm_endpoint="/v1/chat/completions",
litellm_request={
"model": "openrouter/anthropic/claude-sonnet-4",
"messages": [{"role": "user", "content": "Hello world"}],
"reasoning": {"effort": "high"},
"cache": {"no-cache": True},
},
expected_provider_endpoint="/api/v1/chat/completions",
expected_provider_headers={
"authorization": "Bearer synthetic-openrouter-key",
"content-type": "application/json",
},
expected_provider_request={
"model": "anthropic/claude-sonnet-4",
"messages": [{"role": "user", "content": "Hello world"}],
"reasoning": {"effort": "high"},
"usage": {"include": True},
},
mock_provider_response=CLAUDE_SONNET_4_RESPONSE,
expected_litellm_response={
"id": CLAUDE_SONNET_4_RESPONSE["id"],
"object": "chat.completion",
"created": CLAUDE_SONNET_4_RESPONSE["created"],
"model": "openrouter/anthropic/claude-sonnet-4",
"choices": [
{
"index": 0,
"message": {
"role": "assistant",
"content": CLAUDE_SONNET_4_MESSAGE["content"],
"reasoning_content": CLAUDE_SONNET_4_MESSAGE["reasoning"],
},
"finish_reason": "stop",
}
],
"usage": {
"prompt_tokens": CLAUDE_SONNET_4_USAGE["prompt_tokens"],
"completion_tokens": CLAUDE_SONNET_4_USAGE["completion_tokens"],
"total_tokens": CLAUDE_SONNET_4_USAGE["total_tokens"],
"completion_tokens_details": {
"reasoning_tokens": JSON_OBJECT.validate_python(CLAUDE_SONNET_4_USAGE["completion_tokens_details"])[
"reasoning_tokens"
]
},
"prompt_tokens_details": {
"cached_tokens": JSON_OBJECT.validate_python(CLAUDE_SONNET_4_USAGE["prompt_tokens_details"])[
"cached_tokens"
]
},
},
},
)
GEMINI_3_1_FLASH_IMAGE_TEST_CASE: Final = TranslationTestCase(
scenario="image_generation",
litellm_endpoint="/v1/chat/completions",
litellm_request={
"model": "openrouter/google/gemini-3.1-flash-image",
"messages": [
{
"role": "user",
"content": "Generate a perfectly flat solid red square with no details, texture, gradients, shadows, or text.",
}
],
"modalities": ["image", "text"],
"cache": {"no-cache": True},
},
expected_provider_endpoint="/api/v1/chat/completions",
expected_provider_headers={
"authorization": "Bearer synthetic-openrouter-key",
"content-type": "application/json",
},
expected_provider_request={
"model": "google/gemini-3.1-flash-image",
"messages": [
{
"role": "user",
"content": "Generate a perfectly flat solid red square with no details, texture, gradients, shadows, or text.",
}
],
"modalities": ["image", "text"],
"usage": {"include": True},
},
mock_provider_response=GEMINI_3_1_FLASH_IMAGE_RESPONSE,
expected_litellm_response={
"id": GEMINI_3_1_FLASH_IMAGE_RESPONSE["id"],
"object": "chat.completion",
"created": GEMINI_3_1_FLASH_IMAGE_RESPONSE["created"],
"model": "openrouter/google/gemini-3.1-flash-image",
"choices": [
{
"index": 0,
"message": {
"role": "assistant",
"content": None,
"images": GEMINI_3_1_FLASH_IMAGE_MESSAGE["images"],
},
"finish_reason": "stop",
}
],
"usage": {
"prompt_tokens": GEMINI_3_1_FLASH_IMAGE_USAGE["prompt_tokens"],
"completion_tokens": GEMINI_3_1_FLASH_IMAGE_USAGE["completion_tokens"],
"total_tokens": GEMINI_3_1_FLASH_IMAGE_USAGE["total_tokens"],
"completion_tokens_details": {"reasoning_tokens": 0},
"prompt_tokens_details": {"cached_tokens": 0},
},
},
)

55
tests/integration/translation/chat_completions/bases/openrouter_claude_reasoning_response.json Normal file
View file

@ -0,0 +1,55 @@
{
"id": "gen-1791609229-WpnW70pHKiPTu3s3QugG",
"object": "chat.completion",
"created": 1791609229,
"model": "anthropic/claude-sonnet-4",
"provider": "Amazon Bedrock",
"system_fingerprint": null,
"service_tier": null,
"choices": [
{
"index": 0,
"logprobs": null,
"finish_reason": "stop",
"native_finish_reason": "end_turn",
"message": {
"role": "assistant",
"content": "Hello! Nice to meet you. How are you doing today? Is there anything I can help you with?",
"refusal": null,
"reasoning": "The user has just said \"Hello world\" which is a classic greeting, often used as the first program people write when learning a new programming language. They're simply greeting me, so I should respond in a friendly and welcoming way.",
"reasoning_details": [
{
"type": "reasoning.text",
"text": "The user has just said \"Hello world\" which is a classic greeting, often used as the first program people write when learning a new programming language. They're simply greeting me, so I should respond in a friendly and welcoming way.",
"format": "anthropic-claude-v1",
"index": 0,
"signature": "EpQDCkgIEhABGAIqQAVfsNK+CpI8VVV9QHuwaxVcbVQ3X9PVHVuriphXxf7o5ScqJuyRrLmclG5X0UtR8uXuWNftoIN4Jr8LrDtgGaISDGtmH9HlePM65EAvDhoMI/ErOdLdZ2YQXOPPIjDnD/vKsmhqYYOFNk3+Y26sZUUydL8jVnJrKkKyGNKRWYoE5dWsxHO/a/UPDV2ku3Qq+QEqOXXoMSMeTs28hekzdrRQ6QSOjAAbyKzRxUJKjw75+pFmv/qXqWVV6LqkPHtvAEGdlQ5Qc3ZH2ddLmTHAYKcjH6UXKCHFwFUwjH/EoDMmo8xVrHwBRphCMkE3+k7QbitfhSEDtNvGR/LJdCQ2A1pMFLxbbsYcBUsKsr9RujUrQ8et94x0n/f0gLyQOWLRbJ6wGZfSMMzdIEC5xfV1H+qtb3rY4zfyN2FKMI83G/zs6lq5RvlTY9n+1C/OaLqLlb1UIimEh8m8fsJgtCnO5dkHqC/l4Vz0ob+TdnpjCMBBBsItZ3XVPP1Xi5h8T0RknwqCzRvUyMcqNmwYAQ==.eyJlbmRwb2ludF9zbHVnIjoiYW50aHJvcGljL2NsYXVkZS00LXNvbm5ldC0yMDI1MDUyMnxhbWF6b24tYmVkcm9jayJ9"
}
]
}
}
],
"usage": {
"prompt_tokens": 38,
"completion_tokens": 81,
"total_tokens": 119,
"cost": 0.001329,
"is_byok": false,
"prompt_tokens_details": {
"cached_tokens": 0,
"cache_write_tokens": 0,
"audio_tokens": 0,
"video_tokens": 0
},
"cost_details": {
"upstream_inference_cost": 0.001329,
"upstream_inference_prompt_cost": 0.000114,
"upstream_inference_completions_cost": 0.001215
},
"completion_tokens_details": {
"reasoning_tokens": 59,
"image_tokens": 0,
"audio_tokens": 0
}
}
}

62
tests/integration/translation/chat_completions/bases/openrouter_gemini_image_response.json Normal file
View file

File diff suppressed because one or more lines are too long

20
tests/integration/translation/chat_completions/basic/test_chat_completions_basic_openrouter.py Normal file
View file

@ -0,0 +1,20 @@
import pytest
from integration._support.client import Gateway
from integration._support.provider import SharedProvider
from integration.translation.case import TranslationTestCase
from integration.translation.chat_completions.bases.openrouter import (
CLAUDE_SONNET_4_TEST_CASE,
GEMINI_3_1_FLASH_IMAGE_TEST_CASE,
)
from integration.translation.runner import assert_translation
@pytest.mark.parametrize(
"case",
[CLAUDE_SONNET_4_TEST_CASE, GEMINI_3_1_FLASH_IMAGE_TEST_CASE],
ids=lambda case: case.id,
)
def test_chat_completions_basic_openrouter(
case: TranslationTestCase, gateway: Gateway, provider: SharedProvider
) -> None:
assert_translation(case, gateway, provider)

414
tests/llm_translation/test_huggingface_chat_completion.py
View file

@ -1,414 +0,0 @@
"""
Test HuggingFace LLM
"""
import json
from unittest.mock import AsyncMock, MagicMock, patch
from base_llm_unit_tests import BaseLLMChatTest
import pytest
import litellm
from litellm.types.utils import ModelResponse, ModelResponseStream
MOCK_COMPLETION_RESPONSE = {
"id": "9115d3daeab10608",
"object": "chat.completion",
"created": 11111,
"model": "meta-llama/Meta-Llama-3-8B-Instruct",
"prompt": [],
"choices": [
{
"finish_reason": "stop",
"seed": 3629048360264764400,
"logprobs": None,
"index": 0,
"message": {
"role": "assistant",
"content": "This is a test response from the mocked HuggingFace API.",
"tool_calls": [],
},
}
],
"usage": {"prompt_tokens": 10, "completion_tokens": 20, "total_tokens": 30},
}
MOCK_STREAMING_CHUNKS = [
{
"id": "id1",
"object": "chat.completion.chunk",
"created": 1111,
"choices": [
{
"index": 0,
"text": "Deep",
"logprobs": None,
"finish_reason": None,
"seed": None,
"delta": {
"token_id": 34564,
"role": "assistant",
"content": "Deep",
"tool_calls": None,
},
}
],
"model": "meta-llama/Meta-Llama-3-8B-Instruct-Turbo",
"usage": None,
},
{
"id": "id2",
"object": "chat.completion.chunk",
"created": 1111,
"choices": [
{
"index": 0,
"text": " learning",
"logprobs": None,
"finish_reason": None,
"seed": None,
"delta": {
"token_id": 6975,
"role": "assistant",
"content": " learning",
"tool_calls": None,
},
}
],
"model": "meta-llama/Meta-Llama-3-8B-Instruct-Turbo",
"usage": None,
},
{
"id": "id3",
"object": "chat.completion.chunk",
"created": 1111,
"choices": [
{
"index": 0,
"text": " is",
"logprobs": None,
"finish_reason": None,
"seed": None,
"delta": {
"token_id": 374,
"role": "assistant",
"content": " is",
"tool_calls": None,
},
}
],
"model": "meta-llama/Meta-Llama-3-8B-Instruct-Turbo",
"usage": None,
},
{
"id": "sid4",
"object": "chat.completion.chunk",
"created": 1111,
"choices": [
{
"index": 0,
"text": " response",
"logprobs": None,
"finish_reason": "length",
"seed": 2853637492034609700,
"delta": {
"token_id": 323,
"role": "assistant",
"content": " response",
"tool_calls": None,
},
}
],
"model": "meta-llama/Meta-Llama-3-8B-Instruct-Turbo",
"usage": {"prompt_tokens": 26, "completion_tokens": 20, "total_tokens": 46},
},
]
PROVIDER_MAPPING_RESPONSE = {
"fireworks-ai": {
"status": "live",
"providerId": "accounts/fireworks/models/llama-v3-8b-instruct",
"task": "conversational",
},
"together": {
"status": "live",
"providerId": "meta-llama/Meta-Llama-3-8B-Instruct-Turbo",
"task": "conversational",
},
"hf-inference": {
"status": "live",
"providerId": "meta-llama/Meta-Llama-3-8B-Instruct",
"task": "conversational",
},
}
@pytest.fixture
def mock_provider_mapping():
with patch("litellm.llms.huggingface.chat.transformation.fetch_inference_provider_mapping") as mock:
mock.return_value = PROVIDER_MAPPING_RESPONSE
yield mock
@pytest.fixture(autouse=True)
def clear_lru_cache():
from litellm.llms.huggingface.common_utils import fetch_inference_provider_mapping
fetch_inference_provider_mapping.cache_clear()
yield
fetch_inference_provider_mapping.cache_clear()
@pytest.fixture
def mock_http_handler():
"""Fixture to mock the HTTP handler"""
with patch("litellm.llms.custom_httpx.http_handler.HTTPHandler.post") as mock:
print(f"Creating mock HTTP handler: {mock}") # noqa: T201
mock_response = MagicMock()
mock_response.raise_for_status.return_value = None
mock_response.status_code = 200
def mock_side_effect(*args, **kwargs):
if kwargs.get("stream", True):
mock_response.iter_lines.return_value = iter(
[
f"data: {json.dumps(chunk)}".encode("utf-8")
for chunk in MOCK_STREAMING_CHUNKS
]
+ [b"data: [DONE]"]
)
else:
mock_response.json.return_value = MOCK_COMPLETION_RESPONSE
return mock_response
mock.side_effect = mock_side_effect
yield mock
@pytest.fixture
def mock_http_async_handler():
"""Fixture to mock the async HTTP handler"""
with patch(
"litellm.llms.custom_httpx.http_handler.AsyncHTTPHandler.post",
new_callable=AsyncMock,
) as mock:
print(f"Creating mock async HTTP handler: {mock}") # noqa: T201
mock_response = MagicMock()
mock_response.raise_for_status.return_value = None
mock_response.status_code = 200
mock_response.headers = {"content-type": "application/json"}
mock_response.json.return_value = MOCK_COMPLETION_RESPONSE
mock_response.text = json.dumps(MOCK_COMPLETION_RESPONSE)
async def mock_side_effect(*args, **kwargs):
if kwargs.get("stream", True):
async def mock_aiter():
for chunk in MOCK_STREAMING_CHUNKS:
yield f"data: {json.dumps(chunk)}".encode("utf-8")
yield b"data: [DONE]"
mock_response.aiter_lines = mock_aiter
return mock_response
mock.side_effect = mock_side_effect
yield mock
class TestHuggingFace(BaseLLMChatTest):
@pytest.fixture(autouse=True)
def setup(self, mock_provider_mapping, mock_http_handler, mock_http_async_handler):
self.mock_provider_mapping = mock_provider_mapping
self.mock_http = mock_http_handler
self.mock_http_async = mock_http_async_handler
self.model = "huggingface/together/meta-llama/Meta-Llama-3-8B-Instruct"
litellm.set_verbose = False
def get_base_completion_call_args(self) -> dict:
"""Implementation of abstract method from BaseLLMChatTest"""
return {"model": self.model}
def test_completion_non_streaming(self):
messages = [{"role": "user", "content": "This is a dummy message"}]
response = litellm.completion(model=self.model, messages=messages, stream=False)
assert isinstance(response, ModelResponse)
assert (
response.choices[0].message.content
== "This is a test response from the mocked HuggingFace API."
)
assert response.usage is not None
assert response.model == self.model.split("/", 2)[2]
def test_completion_streaming(self):
messages = [{"role": "user", "content": "This is a dummy message"}]
response = litellm.completion(model=self.model, messages=messages, stream=True)
chunks = list(response)
assert len(chunks) > 0
assert self.mock_http.called
call_args = self.mock_http.call_args
assert call_args is not None
kwargs = call_args[1]
data = json.loads(kwargs["data"])
assert data["stream"] is True
assert data["messages"] == messages
assert isinstance(chunks, list)
assert isinstance(chunks[0], ModelResponseStream)
assert isinstance(chunks[0].id, str)
assert chunks[0].model == self.model.split("/", 1)[1]
@pytest.mark.asyncio
async def test_async_completion_streaming(self):
"""Test async streaming completion"""
messages = [{"role": "user", "content": "This is a dummy message"}]
response = await litellm.acompletion(
model=self.model, messages=messages, stream=True
)
chunks = []
async for chunk in response:
chunks.append(chunk)
assert self.mock_http_async.called
assert len(chunks) > 0
assert isinstance(chunks[0], ModelResponseStream)
assert isinstance(chunks[0].id, str)
assert chunks[0].model == self.model.split("/", 1)[1]
@pytest.mark.asyncio
async def test_async_completion_non_streaming(self):
"""Test async non-streaming completion"""
messages = [{"role": "user", "content": "This is a dummy message"}]
response = await litellm.acompletion(
model=self.model, messages=messages, stream=False
)
assert self.mock_http_async.called
assert isinstance(response, ModelResponse)
assert (
response.choices[0].message.content
== "This is a test response from the mocked HuggingFace API."
)
assert response.usage is not None
assert response.model == self.model.split("/", 2)[2]
def test_tool_call_no_arguments(self, tool_call_no_arguments):
mock_tool_response = {
**MOCK_COMPLETION_RESPONSE,
"choices": [
{
"finish_reason": "tool_calls",
"index": 0,
"message": tool_call_no_arguments,
}
],
}
with patch.object(
self.mock_http,
"side_effect",
lambda *args, **kwargs: MagicMock(
status_code=200,
json=lambda: mock_tool_response,
raise_for_status=lambda: None,
),
):
messages = [{"role": "user", "content": "Get the FAQ"}]
tools = [
{
"type": "function",
"function": {
"name": "Get-FAQ",
"description": "Get FAQ information",
"parameters": {
"type": "object",
"properties": {},
"required": [],
},
},
}
]
response = litellm.completion(
model=self.model, messages=messages, tools=tools, tool_choice="auto"
)
assert response.choices[0].message.tool_calls is not None
assert len(response.choices[0].message.tool_calls) == 1
assert (
response.choices[0].message.tool_calls[0].function.name
== tool_call_no_arguments["tool_calls"][0]["function"]["name"]
)
assert (
response.choices[0].message.tool_calls[0].function.arguments
== tool_call_no_arguments["tool_calls"][0]["function"]["arguments"]
)
def test_completion_with_api_base(self):
messages = [{"role": "user", "content": "This is a test message"}]
api_base = "https://abcd123.us-east-1.aws.endpoints.huggingface.cloud"
response = litellm.completion(
model="huggingface/tgi", messages=messages, api_base=api_base, stream=False
)
assert isinstance(response, ModelResponse)
assert (
response.choices[0].message.content
== "This is a test response from the mocked HuggingFace API."
)
assert self.mock_http.called
call_args = self.mock_http.call_args
assert call_args is not None
called_url = call_args[1]["url"]
assert called_url == f"{api_base}/v1/chat/completions"
@pytest.mark.asyncio
async def test_async_completion_with_api_base(self):
messages = [{"role": "user", "content": "This is a test message"}]
api_base = "https://abcd123.us-east-1.aws.endpoints.huggingface.cloud"
response = await litellm.acompletion(
model="huggingface/tgi", messages=messages, api_base=api_base, stream=False
)
assert isinstance(response, ModelResponse)
assert (
response.choices[0].message.content
== "This is a test response from the mocked HuggingFace API."
)
assert self.mock_http_async.called
call_args = self.mock_http_async.call_args
assert call_args is not None
called_url = call_args[1]["url"]
assert called_url == f"{api_base}/v1/chat/completions"
def test_completion_streaming_with_api_base(self):
"""Test streaming completion with api_base parameter"""
messages = [{"role": "user", "content": "This is a test message"}]
api_base = "https://abcd123.us-east-1.aws.endpoints.huggingface.cloud"
response = litellm.completion(
model="huggingface/tgi", messages=messages, api_base=api_base, stream=True
)
chunks = list(response)
assert len(chunks) > 0
assert isinstance(chunks[0], ModelResponseStream)
# Check that the correct URL was called
assert self.mock_http.called
call_args = self.mock_http.call_args
assert call_args is not None
called_url = call_args[1]["url"]
assert called_url == f"{api_base}/v1/chat/completions"

150
tests/llm_translation/test_json_providers.py
View file

@ -1,150 +0,0 @@
"""
Tests for JSON-based provider configuration system.
"""
import os
import pytest
import litellm
class TestPublicAIIntegration:
"""Integration tests for PublicAI provider"""
def test_publicai_completion_basic(self):
"""Test basic completion call to PublicAI"""
# Skip test if API key not set in environment
if not os.environ.get("PUBLICAI_API_KEY"):
if pytest:
pytest.skip("PUBLICAI_API_KEY not set")
return
try:
response = litellm.completion(
model="publicai/swiss-ai/apertus-8b-instruct",
messages=[
{
"role": "user",
"content": "Say 'test successful' and nothing else",
}
],
max_tokens=10,
)
# Verify response structure
assert response is not None
assert hasattr(response, "choices")
assert len(response.choices) > 0
assert hasattr(response.choices[0], "message")
assert hasattr(response.choices[0].message, "content")
assert response.choices[0].message.content is not None
# Check that we got a response
content = response.choices[0].message.content.lower()
assert len(content) > 0
print(
f"✓ PublicAI completion successful: {response.choices[0].message.content}"
)
except Exception as e:
if pytest:
pytest.fail(f"PublicAI completion failed: {str(e)}")
else:
raise
def test_publicai_completion_with_streaming(self):
"""Test streaming completion with PublicAI"""
# Skip test if API key not set in environment
if not os.environ.get("PUBLICAI_API_KEY"):
if pytest:
pytest.skip("PUBLICAI_API_KEY not set")
return
try:
response = litellm.completion(
model="publicai/swiss-ai/apertus-8b-instruct",
messages=[{"role": "user", "content": "Count to 3"}],
max_tokens=20,
stream=True,
)
# Collect chunks
chunks = []
for chunk in response:
assert chunk is not None
if hasattr(chunk.choices[0], "delta") and hasattr(
chunk.choices[0].delta, "content"
):
if chunk.choices[0].delta.content:
chunks.append(chunk.choices[0].delta.content)
# Verify we got chunks
assert len(chunks) > 0
full_response = "".join(chunks)
assert len(full_response) > 0
print(f"✓ PublicAI streaming successful: {full_response}")
except Exception as e:
if pytest:
pytest.fail(f"PublicAI streaming failed: {str(e)}")
else:
raise
def test_publicai_parameter_mapping(self):
"""Test that max_completion_tokens is mapped to max_tokens"""
# Skip test if API key not set in environment
if not os.environ.get("PUBLICAI_API_KEY"):
if pytest:
pytest.skip("PUBLICAI_API_KEY not set")
return
try:
# Use max_completion_tokens (OpenAI's newer parameter)
response = litellm.completion(
model="publicai/swiss-ai/apertus-8b-instruct",
messages=[{"role": "user", "content": "Hi"}],
max_completion_tokens=5, # This should be mapped to max_tokens
)
assert response is not None
assert len(response.choices) > 0
print("✓ Parameter mapping successful")
except Exception as e:
if pytest:
pytest.fail(f"Parameter mapping test failed: {str(e)}")
else:
raise
def test_publicai_content_list_conversion(self):
"""Test that content list format is converted to string"""
# Skip test if API key not set in environment
if not os.environ.get("PUBLICAI_API_KEY"):
if pytest:
pytest.skip("PUBLICAI_API_KEY not set")
return
try:
# Send message with content as list (should be converted to string)
response = litellm.completion(
model="publicai/swiss-ai/apertus-8b-instruct",
messages=[
{"role": "user", "content": [{"type": "text", "text": "Say hello"}]}
],
max_tokens=10,
)
assert response is not None
assert len(response.choices) > 0
print("✓ Content list conversion successful")
except Exception as e:
if pytest:
pytest.fail(f"Content list conversion test failed: {str(e)}")
else:
raise

42
tests/llm_translation/test_openrouter.py
View file

@ -1,42 +0,0 @@
import pytest
import litellm
def test_completion_openrouter_reasoning_content():
litellm.turn_on_debug()
resp = litellm.completion(
model="openrouter/anthropic/claude-sonnet-4",
messages=[{"role": "user", "content": "Hello world"}],
reasoning={"effort": "high"},
)
print(resp)
assert resp.choices[0].message.reasoning_content is not None
def test_completion_openrouter_image_generation():
litellm.turn_on_debug()
resp = litellm.completion(
model="openrouter/google/gemini-2.5-flash-image",
messages=[{"role": "user", "content": "Generate an image of a cat"}],
modalities=["image", "text"],
)
print(resp)
assert (
resp.choices[0].message.images[0]["image_url"]["url"].startswith("data:image/")
)
def test_openrouter_embedding():
"""Test OpenRouter embeddings support."""
litellm.turn_on_debug()
resp = litellm.embedding(
model="openrouter/openai/text-embedding-3-small",
input=["Hello world", "How are you?"],
)
print(resp)
assert resp is not None
assert len(resp.data) == 2
assert resp.data[0]["embedding"] is not None
assert isinstance(resp.data[0]["embedding"], list)
assert len(resp.data[0]["embedding"]) > 0

308
tests/unit/llms/huggingface/chat/test_huggingface_chat_transformation.py
View file

@ -1,11 +1,82 @@
import json
from collections.abc import Iterator
from typing import Final
from unittest.mock import MagicMock, patch
import httpx
import pytest
import respx
import litellm
from litellm.llms.custom_httpx.http_handler import AsyncHTTPHandler, HTTPHandler
from litellm.llms.huggingface.common_utils import _fetch_inference_provider_mapping
HUGGINGFACE_MODEL: Final = "meta-llama/Meta-Llama-3-8B-Instruct"
HUGGINGFACE_ROUTER_MODEL: Final = "meta-llama/Meta-Llama-3-8B-Instruct-Turbo"
HUGGINGFACE_COMPLETION_RESPONSE: Final = {
"id": "hf-completion",
"object": "chat.completion",
"created": 11111,
"model": HUGGINGFACE_ROUTER_MODEL,
"choices": [
{
"index": 0,
"message": {
"role": "assistant",
"content": "Hugging Face response",
"tool_calls": [],
},
"finish_reason": "stop",
}
],
"usage": {"prompt_tokens": 10, "completion_tokens": 20, "total_tokens": 30},
}
HUGGINGFACE_STREAMING_CHUNKS: Final = (
{
"id": "hf-stream-1",
"object": "chat.completion.chunk",
"created": 11111,
"model": HUGGINGFACE_ROUTER_MODEL,
"choices": [{"index": 0, "delta": {"role": "assistant", "content": "Hugging "}, "finish_reason": None}],
},
{
"id": "hf-stream-2",
"object": "chat.completion.chunk",
"created": 11111,
"model": HUGGINGFACE_ROUTER_MODEL,
"choices": [{"index": 0, "delta": {"content": "Face"}, "finish_reason": None}],
},
{
"id": "hf-stream-3",
"object": "chat.completion.chunk",
"created": 11111,
"model": HUGGINGFACE_ROUTER_MODEL,
"choices": [{"index": 0, "delta": {}, "finish_reason": "stop"}],
},
)
HUGGINGFACE_STREAMING_RESPONSE: Final = (
b"".join(b"data: " + json.dumps(chunk).encode() + b"\n\n" for chunk in HUGGINGFACE_STREAMING_CHUNKS)
+ b"data: [DONE]\n\n"
)
class RespxAsyncTransport(httpx.AsyncBaseTransport):
def __init__(self, router: respx.MockRouter) -> None:
self.router: Final = router
async def handle_async_request(self, request: httpx.Request) -> httpx.Response:
return await self.router.async_handler(request)
def _add_provider_mapping_route(router: respx.MockRouter) -> None:
route: Final = router.get(f"https://huggingface.co/api/models/{HUGGINGFACE_MODEL}")
route.return_value = httpx.Response(200, json={"inferenceProviderMapping": PROVIDER_MAPPING_RESPONSE})
def _add_tokenizer_route(router: respx.MockRouter) -> None:
route: Final = router.head("https://huggingface.co/Xenova/llama-3-tokenizer/resolve/main/tokenizer.json")
route.return_value = httpx.Response(200)
@pytest.fixture(autouse=True)
def isolate_huggingface_environment(monkeypatch: pytest.MonkeyPatch) -> Iterator[None]:
@ -215,3 +286,240 @@ def test_validate_environment():
assert headers["Authorization"] == "Bearer test_api_key"
assert headers["content-type"] == "application/json"
@pytest.mark.respx(assert_all_called=True)
def test_completion_non_streaming_uses_resolved_provider_model(respx_mock: respx.MockRouter) -> None:
_add_provider_mapping_route(respx_mock)
route: Final = respx_mock.post("https://router.huggingface.co/together/v1/chat/completions")
route.return_value = httpx.Response(200, json=HUGGINGFACE_COMPLETION_RESPONSE)
messages: Final = [{"role": "user", "content": "This is a dummy message"}]
response: Final = litellm.completion(
model="huggingface/together/meta-llama/Meta-Llama-3-8B-Instruct",
messages=messages,
api_key="test-api-key",
client=HTTPHandler(),
)
request_body: Final = route.calls[0].request.read()
assert json.loads(request_body) == {
"model": HUGGINGFACE_ROUTER_MODEL,
"messages": messages,
}
assert route.calls[0].request.headers["Authorization"] == "Bearer test-api-key"
assert response.choices[0].message.content == "Hugging Face response"
assert response.usage is not None
assert response.usage.total_tokens == 30
assert response.model == HUGGINGFACE_ROUTER_MODEL
@pytest.mark.respx(assert_all_called=True)
def test_completion_streaming_uses_resolved_provider_model(respx_mock: respx.MockRouter) -> None:
_add_provider_mapping_route(respx_mock)
route: Final = respx_mock.post("https://router.huggingface.co/together/v1/chat/completions")
route.return_value = httpx.Response(
200,
headers={"content-type": "text/event-stream"},
content=HUGGINGFACE_STREAMING_RESPONSE,
)
messages: Final = [{"role": "user", "content": "This is a dummy message"}]
chunks: Final = tuple(
litellm.completion(
model="huggingface/together/meta-llama/Meta-Llama-3-8B-Instruct",
messages=messages,
stream=True,
api_key="test-api-key",
client=HTTPHandler(),
)
)
request_body: Final = json.loads(route.calls[0].request.content)
assert request_body == {"model": HUGGINGFACE_ROUTER_MODEL, "messages": messages, "stream": True}
assert tuple(chunk.choices[0].delta.content for chunk in chunks if chunk.choices[0].delta.content) == (
"Hugging ",
"Face",
)
@pytest.mark.asyncio
@pytest.mark.respx(assert_all_called=True)
async def test_async_completion_non_streaming_uses_resolved_provider_model(
respx_mock: respx.MockRouter,
) -> None:
_add_provider_mapping_route(respx_mock)
route: Final = respx_mock.post("https://router.huggingface.co/together/v1/chat/completions")
route.return_value = httpx.Response(200, json=HUGGINGFACE_COMPLETION_RESPONSE)
client: Final = AsyncHTTPHandler(transport=RespxAsyncTransport(respx_mock))
messages: Final = [{"role": "user", "content": "This is a dummy message"}]
response: Final = await litellm.acompletion(
model="huggingface/together/meta-llama/Meta-Llama-3-8B-Instruct",
messages=messages,
api_key="test-api-key",
client=client,
)
await client.client.aclose()
request_body: Final = json.loads(route.calls[0].request.content)
assert request_body == {"model": HUGGINGFACE_ROUTER_MODEL, "messages": messages}
assert response.choices[0].message.content == "Hugging Face response"
assert response.usage is not None
assert response.usage.total_tokens == 30
assert response.model == HUGGINGFACE_ROUTER_MODEL
@pytest.mark.asyncio
@pytest.mark.respx(assert_all_called=True)
async def test_async_completion_streaming_uses_resolved_provider_model(
respx_mock: respx.MockRouter,
) -> None:
_add_provider_mapping_route(respx_mock)
_add_tokenizer_route(respx_mock)
route: Final = respx_mock.post("https://router.huggingface.co/together/v1/chat/completions")
route.return_value = httpx.Response(
200,
headers={"content-type": "text/event-stream"},
content=HUGGINGFACE_STREAMING_RESPONSE,
)
client: Final = AsyncHTTPHandler(transport=RespxAsyncTransport(respx_mock))
messages: Final = [{"role": "user", "content": "This is a dummy message"}]
stream: Final = await litellm.acompletion(
model="huggingface/together/meta-llama/Meta-Llama-3-8B-Instruct",
messages=messages,
stream=True,
api_key="test-api-key",
client=client,
)
chunks: Final = tuple([chunk async for chunk in stream])
await client.client.aclose()
request_body: Final = json.loads(route.calls[0].request.content)
assert request_body == {"model": HUGGINGFACE_ROUTER_MODEL, "messages": messages, "stream": True}
assert tuple(chunk.choices[0].delta.content for chunk in chunks if chunk.choices[0].delta.content) == (
"Hugging ",
"Face",
)
@pytest.mark.respx(assert_all_called=True)
def test_completion_with_api_base_uses_endpoint_url(respx_mock: respx.MockRouter) -> None:
route: Final = respx_mock.post("https://custom.huggingface.test/v1/chat/completions")
route.return_value = httpx.Response(200, json=HUGGINGFACE_COMPLETION_RESPONSE)
messages: Final = [{"role": "user", "content": "This is a test message"}]
response: Final = litellm.completion(
model="huggingface/tgi",
messages=messages,
api_base="https://custom.huggingface.test/v1",
api_key="test-api-key",
client=HTTPHandler(),
)
assert route.called
assert response.choices[0].message.content == "Hugging Face response"
@pytest.mark.asyncio
@pytest.mark.respx(assert_all_called=True)
async def test_async_completion_with_api_base_uses_endpoint_url(
respx_mock: respx.MockRouter,
) -> None:
route: Final = respx_mock.post("https://custom.huggingface.test/v1/chat/completions")
route.return_value = httpx.Response(200, json=HUGGINGFACE_COMPLETION_RESPONSE)
client: Final = AsyncHTTPHandler(transport=RespxAsyncTransport(respx_mock))
messages: Final = [{"role": "user", "content": "This is a test message"}]
response: Final = await litellm.acompletion(
model="huggingface/tgi",
messages=messages,
api_base="https://custom.huggingface.test/v1",
api_key="test-api-key",
client=client,
)
await client.client.aclose()
assert route.called
assert response.choices[0].message.content == "Hugging Face response"
@pytest.mark.respx(assert_all_called=True)
def test_completion_streaming_with_api_base_uses_endpoint_url(respx_mock: respx.MockRouter) -> None:
route: Final = respx_mock.post("https://custom.huggingface.test/v1/chat/completions")
route.return_value = httpx.Response(
200,
headers={"content-type": "text/event-stream"},
content=HUGGINGFACE_STREAMING_RESPONSE,
)
messages: Final = [{"role": "user", "content": "This is a test message"}]
chunks: Final = tuple(
litellm.completion(
model="huggingface/tgi",
messages=messages,
api_base="https://custom.huggingface.test/v1",
stream=True,
api_key="test-api-key",
client=HTTPHandler(),
)
)
request_body: Final = json.loads(route.calls[0].request.content)
assert request_body["stream"] is True
assert request_body["messages"] == messages
assert tuple(chunk.choices[0].delta.content for chunk in chunks if chunk.choices[0].delta.content) == (
"Hugging ",
"Face",
)
@pytest.mark.respx(assert_all_called=True)
def test_tool_call_without_arguments_is_preserved(respx_mock: respx.MockRouter) -> None:
_add_provider_mapping_route(respx_mock)
tool_response: Final = {
**HUGGINGFACE_COMPLETION_RESPONSE,
"choices": [
{
"index": 0,
"message": {
"role": "assistant",
"content": None,
"tool_calls": [
{
"id": "call_faq",
"type": "function",
"function": {"name": "Get-FAQ", "arguments": ""},
}
],
},
"finish_reason": "tool_calls",
}
],
}
route: Final = respx_mock.post("https://router.huggingface.co/together/v1/chat/completions")
route.return_value = httpx.Response(200, json=tool_response)
messages: Final = [{"role": "user", "content": "Get the FAQ"}]
tools: Final = [
{
"type": "function",
"function": {"name": "Get-FAQ", "description": "Get FAQ information", "parameters": {"type": "object"}},
}
]
response: Final = litellm.completion(
model="huggingface/together/meta-llama/Meta-Llama-3-8B-Instruct",
messages=messages,
tools=tools,
tool_choice="auto",
api_key="test-api-key",
client=HTTPHandler(),
)
request_body: Final = json.loads(route.calls[0].request.content)
assert request_body["model"] == HUGGINGFACE_ROUTER_MODEL
assert request_body["tools"] == tools
assert response.choices[0].message.tool_calls is not None
assert response.choices[0].message.tool_calls[0].function.name == "Get-FAQ"
assert response.choices[0].message.tool_calls[0].function.arguments == ""

43
tests/unit/llms/openrouter/test_openrouter_embedding_transformation.py
View file

@ -2,6 +2,15 @@
Unit tests for OpenRouter embedding transformation logic.
"""
import json
from typing import Final
import httpx
import pytest
import respx
import litellm
from litellm.llms.custom_httpx.http_handler import HTTPHandler
from litellm.llms.openrouter.embedding.transformation import (
OpenrouterEmbeddingConfig,
)
@ -131,3 +140,37 @@ def test_openrouter_embedding_map_params():
assert result["timeout"] == 30
# Unsupported params should not be included
assert "unsupported" not in result
@pytest.mark.respx(assert_all_called=True)
def test_openrouter_embedding_request_and_response(respx_mock: respx.MockRouter) -> None:
route: Final = respx_mock.post("https://openrouter.ai/api/v1/embeddings")
route.return_value = httpx.Response(
200,
json={
"object": "list",
"data": [
{"object": "embedding", "index": 0, "embedding": [0.1, 0.2]},
{"object": "embedding", "index": 1, "embedding": [0.3, 0.4]},
],
"model": "openai/text-embedding-3-small",
"usage": {"prompt_tokens": 4, "total_tokens": 4},
},
)
inputs: Final = ["Hello world", "How are you?"]
response: Final = litellm.embedding(
model="openrouter/openai/text-embedding-3-small",
input=inputs,
api_key="test-api-key",
client=HTTPHandler(),
)
request_body: Final = json.loads(route.calls[0].request.content)
assert request_body == {
"model": "openai/text-embedding-3-small",
"input": inputs,
}
assert route.calls[0].request.headers["Authorization"] == "Bearer test-api-key"
assert [item["embedding"] for item in response.data] == [[0.1, 0.2], [0.3, 0.4]]
assert response.usage.total_tokens == 4

132
tests/unit/llms/publicai/test_publicai_chat_transformation.py
View file

@ -5,15 +5,48 @@ These tests validate the PublicAI configuration which is now JSON-based.
PublicAI is an OpenAI-compatible provider with minor customizations.
"""
import json
from typing import Final
from unittest.mock import patch
import httpx
import pytest
import respx
import litellm
from litellm.llms.openai_like.json_loader import JSONProviderRegistry
from litellm.llms.openai_like.dynamic_config import create_config_class
PUBLICAI_COMPLETION_RESPONSE: Final = {
"id": "publicai-completion",
"object": "chat.completion",
"created": 11111,
"model": "swiss-ai/apertus-8b-instruct",
"choices": [
{
"index": 0,
"message": {"role": "assistant", "content": "Hello from PublicAI"},
"finish_reason": "stop",
}
],
"usage": {"prompt_tokens": 3, "completion_tokens": 4, "total_tokens": 7},
}
PUBLICAI_STREAMING_RESPONSE: Final = b"".join(
(
b'data: {"id":"publicai-stream-1","object":"chat.completion.chunk","created":11111,'
b'"model":"swiss-ai/apertus-8b-instruct","choices":[{"index":0,"delta":{"content":"Hello "},'
b'"finish_reason":null}]}\n\n',
b'data: {"id":"publicai-stream-2","object":"chat.completion.chunk","created":11111,'
b'"model":"swiss-ai/apertus-8b-instruct","choices":[{"index":0,"delta":{"content":"there"},'
b'"finish_reason":null}]}\n\n',
b'data: {"id":"publicai-stream-3","object":"chat.completion.chunk","created":11111,'
b'"model":"swiss-ai/apertus-8b-instruct","choices":[{"index":0,"delta":{},"finish_reason":"stop"}]}\n\n',
b"data: [DONE]\n\n",
)
)
class TestPublicAIConfig:
"""Test class for PublicAI functionality"""
@ -137,3 +170,100 @@ class TestPublicAIConfig:
)
assert url == "https://custom.publicai.co/v1/chat/completions"
@pytest.mark.respx(assert_all_called=True)
def test_publicai_content_list_conversion_reaches_provider_as_text(respx_mock: respx.MockRouter) -> None:
route: Final = respx_mock.post("https://api.publicai.co/v1/chat/completions")
route.return_value = httpx.Response(200, json=PUBLICAI_COMPLETION_RESPONSE)
response: Final = litellm.completion(
model="publicai/swiss-ai/apertus-8b-instruct",
messages=[{"role": "user", "content": [{"type": "text", "text": "Say hello"}]}],
max_tokens=10,
api_key="test-api-key",
)
request_body: Final = json.loads(route.calls[0].request.content)
assert request_body == {
"model": "swiss-ai/apertus-8b-instruct",
"messages": [{"role": "user", "content": "Say hello"}],
"max_tokens": 10,
}
assert response.choices[0].message.content == "Hello from PublicAI"
assert response.usage is not None
assert response.usage.total_tokens == 7
@pytest.mark.respx(assert_all_called=True)
def test_publicai_parameter_mapping_reaches_provider_as_max_tokens(respx_mock: respx.MockRouter) -> None:
route: Final = respx_mock.post("https://api.publicai.co/v1/chat/completions")
route.return_value = httpx.Response(200, json=PUBLICAI_COMPLETION_RESPONSE)
response: Final = litellm.completion(
model="publicai/swiss-ai/apertus-8b-instruct",
messages=[{"role": "user", "content": "Hi"}],
max_completion_tokens=5,
api_key="test-api-key",
)
request_body: Final = json.loads(route.calls[0].request.content)
assert request_body == {
"model": "swiss-ai/apertus-8b-instruct",
"messages": [{"role": "user", "content": "Hi"}],
"max_tokens": 5,
}
assert response.choices[0].message.content == "Hello from PublicAI"
@pytest.mark.respx(assert_all_called=True)
def test_publicai_completion_basic_uses_provider_request(respx_mock: respx.MockRouter) -> None:
route: Final = respx_mock.post("https://api.publicai.co/v1/chat/completions")
route.return_value = httpx.Response(200, json=PUBLICAI_COMPLETION_RESPONSE)
messages: Final = [{"role": "user", "content": "Say hello"}]
response: Final = litellm.completion(
model="publicai/swiss-ai/apertus-8b-instruct",
messages=messages,
max_tokens=10,
api_key="test-api-key",
)
assert json.loads(route.calls[0].request.content) == {
"model": "swiss-ai/apertus-8b-instruct",
"messages": messages,
"max_tokens": 10,
}
assert response.choices[0].message.content == "Hello from PublicAI"
@pytest.mark.respx(assert_all_called=True)
def test_publicai_completion_with_streaming_uses_provider_request(respx_mock: respx.MockRouter) -> None:
route: Final = respx_mock.post("https://api.publicai.co/v1/chat/completions")
route.return_value = httpx.Response(
200,
headers={"content-type": "text/event-stream"},
content=PUBLICAI_STREAMING_RESPONSE,
)
messages: Final = [{"role": "user", "content": "Say hello"}]
chunks: Final = tuple(
litellm.completion(
model="publicai/swiss-ai/apertus-8b-instruct",
messages=messages,
max_tokens=10,
stream=True,
api_key="test-api-key",
)
)
assert json.loads(route.calls[0].request.content) == {
"model": "swiss-ai/apertus-8b-instruct",
"messages": messages,
"max_tokens": 10,
"stream": True,
}
assert tuple(chunk.choices[0].delta.content for chunk in chunks if chunk.choices[0].delta.content) == (
"Hello ",
"there",
)

58
tests/unit/rerank_api/test_main.py
View file

@ -1,6 +1,5 @@
import json
import logging
from datetime import datetime
from typing import Final
from unittest.mock import MagicMock, patch
@ -10,10 +9,7 @@ import respx
import litellm
from litellm.integrations.custom_logger import CustomLogger
from litellm.litellm_core_utils.litellm_logging import Logging
from litellm.llms.custom_httpx.http_handler import AsyncHTTPHandler, HTTPHandler
from litellm.types.rerank import RerankResponse
MARKER_QUERY = "MARKER_QUERY_do_not_log_at_info"
MARKER_DOC = "MARKER_DOC_sensitive_customer_text"
@ -541,46 +537,22 @@ def test_cohere_rerank_custom_api_base_selects_versioned_route(
@pytest.mark.asyncio
async def test_cohere_rerank_success_callback_receives_cost() -> None:
class CostCallback(CustomLogger):
def __init__(self) -> None:
self.response_cost: float | None = None
self.result_index: int | None = None
super().__init__()
@pytest.mark.respx(assert_all_called=True)
async def test_cohere_rerank_success_response_includes_cost(respx_mock: respx.MockRouter) -> None:
route: Final = respx_mock.post("https://api.cohere.com/v2/rerank")
route.return_value = httpx.Response(200, json=COHERE_RERANK_RESPONSE)
client: Final = AsyncHTTPHandler(transport=RespxAsyncTransport(respx_mock))
async def async_log_success_event(
self,
kwargs: dict[str, object],
response_obj: RerankResponse,
start_time: datetime,
end_time: datetime,
) -> None:
response_cost: Final = kwargs["response_cost"]
assert isinstance(response_cost, (int, float))
self.response_cost = float(response_cost)
self.result_index = response_obj.results[0]["index"]
callback: Final = CostCallback()
logging_obj: Final = Logging(
response: Final = await litellm.arerank(
model="cohere/rerank-v4.0-pro",
messages=[],
stream=False,
call_type="rerank",
start_time=datetime(2025, 1, 1),
litellm_call_id="rerank-test-call",
function_id="rerank-test-function",
dynamic_async_success_callbacks=[callback],
kwargs={"custom_llm_provider": "cohere"},
query="hello",
documents=["hello", "world"],
top_n=2,
api_key="test-api-key",
client=client,
)
response: Final = RerankResponse.model_validate(COHERE_RERANK_RESPONSE)
logging_obj.model_call_details["response_cost"] = 0.01
await client.client.aclose()
await logging_obj.async_success_handler(
result=response,
start_time=datetime(2025, 1, 1),
end_time=datetime(2025, 1, 1),
)
assert callback.response_cost is not None
assert callback.response_cost > 0
assert callback.result_index == 0
search_units: Final = COHERE_RERANK_RESPONSE["meta"]["billed_units"]["search_units"]
expected_cost: Final = litellm.model_cost["rerank-v4.0-pro"]["input_cost_per_query"] * search_units
assert response._hidden_params["response_cost"] == expected_cost

21
tests/unit/test_utils_get_optional_params.py
View file

@ -2013,7 +2013,7 @@ def test_get_optional_params_preserves_num_retries_as_max_retries() -> None:
def test_responses_allowed_openai_params_are_sent(
respx_mock: respx.MockRouter,
) -> None:
route: Final = respx_mock.post("https://api.openai.com/v1/responses")
route: Final = respx_mock.post("https://api.perplexity.ai/v1/responses")
route.return_value = httpx.Response(
200,
json={
@ -2035,19 +2035,26 @@ def test_responses_allowed_openai_params_are_sent(
},
)
with pytest.raises(litellm.UnsupportedParamsError):
litellm.responses(
model="perplexity/sonar",
input="Hello",
store=True,
api_key="test-api-key",
client=HTTPHandler(),
)
litellm.responses(
model="openai/gpt-5.6",
model="perplexity/sonar",
input="Hello",
max_output_tokens=100,
top_logprobs=10,
allowed_openai_params=["top_logprobs"],
store=True,
allowed_openai_params=["store"],
api_key="test-api-key",
client=HTTPHandler(),
)
request_body: Final = json.loads(route.calls[0].request.content)
assert request_body["top_logprobs"] == 10
assert request_body["max_output_tokens"] == 100
assert request_body["store"] is True
def test_supports_system_message_merges_system_prompt_into_user_message() -> None: