scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-10-29 06:33:28 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 200

fix(bedrock): adapt native output backports to release interfaces

Browse source
This commit is contained in:
Joshua Valluru 2026-10-20 17:41:23 -07:00
parent a9bad2b5e0
commit 71b6509656
22 changed files with 669 additions and 1054 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
litellm/litellm_core_utils/prompt_templates/common_utils.py
View file

@ -1,3 +1,5 @@
from pydantic import TypeAdapter
"""
Common utility functions used for translating messages across providers
"""

5
litellm/llms/anthropic/chat/transformation.py
View file

@ -1454,6 +1454,11 @@ class AnthropicConfig(AnthropicModelInfo, BaseConfig):
return None
return AnthropicConfig.filter_anthropic_output_schema(dict(json_schema))
def uses_native_structured_output(self, model: str) -> bool:
return AnthropicConfig._supports_model_capability(
model, "supports_native_structured_output", self._resolved_provider
)
def uses_native_structured_output_for_schema(
self,
model: str,

1
litellm/llms/bedrock/chat/converse_transformation.py
View file

@ -54,6 +54,7 @@ from litellm.llms.anthropic.chat.transformation import (
from litellm.llms.anthropic.common_utils import AnthropicModelInfo
from litellm.llms.base_llm.chat.transformation import REQUEST_PARAMS, BaseConfig, BaseLLMException
from litellm.llms.bedrock.common_utils import (
bedrock_capability_model,
bedrock_chat_uses_native_structured_output,
bedrock_cleaned_output_schema,
bedrock_model_supports_regex_lookaround,

13
litellm/llms/bedrock/chat/invoke_transformations/anthropic_claude3_transformation.py
View file

@ -1,6 +1,8 @@
from collections.abc import Mapping
from typing import TYPE_CHECKING, Any, Final
import httpx
from pydantic import TypeAdapter
from litellm.anthropic_beta_headers_manager import filter_and_transform_beta_headers
from litellm.constants import RESPONSE_FORMAT_TOOL_NAME
@ -112,20 +114,9 @@ class AmazonAnthropicClaudeConfig(AmazonInvokeConfig, AnthropicConfig):
model: str,
drop_params: bool,
) -> dict:
# Force tool-based structured outputs for Bedrock Invoke
# (similar to VertexAI fix in #19201) unless the model map advertises
# native structured output
from litellm.utils import supports_native_structured_output
original_model: Final = model
requested_thinking: Final = non_default_params.get("thinking")
requested_display_updates: Final = self.is_thinking_display_updates_used(requested_thinking)
if "response_format" in non_default_params and not supports_native_structured_output(
model=model, custom_llm_provider="bedrock"
):
# Use a model name that forces tool-based approach
model = "claude-3-sonnet-20240229"
# Clamp ``reasoning_effort`` to the Bedrock effort ceiling before the
# parent mapping converts it to ``output_config.effort`` and the
# downstream effort gate runs. Mirrors the converse path's

6
litellm/llms/bedrock/common_utils.py
View file

@ -968,6 +968,12 @@ def is_bedrock_application_inference_profile_arn(model: str) -> bool:
return ":application-inference-profile/" in model
def bedrock_capability_model(model: str, base_model: object) -> str:
if isinstance(base_model, str) and base_model and is_bedrock_application_inference_profile_arn(model):
return strip_bedrock_routing_prefix(base_model)
return model
def strip_bedrock_routing_prefix(model: str) -> str:
"""Strip LiteLLM routing prefixes from model name."""
for prefix in ["bedrock/", "chat_completions/", "converse/", "invoke/", "openai/", "mantle/", "nova-2/", "nova/"]:

2
litellm/llms/bedrock/messages/invoke_transformations/anthropic_claude3_transformation.py
View file

@ -39,8 +39,8 @@ from litellm.llms.bedrock.common_utils import (
bedrock_supports_tool_search,
ensure_bedrock_anthropic_messages_tool_names,
get_anthropic_beta_from_headers,
is_claude_4_5_on_bedrock,
is_bedrock_native_output_rejection,
is_claude_4_5_on_bedrock,
normalize_bedrock_opus_output_config_effort,
normalize_custom_field_on_tools,
normalize_tool_input_schema_types_for_bedrock_invoke,

9
litellm/llms/custom_httpx/llm_http_handler.py
View file

@ -28,6 +28,7 @@ import httpx
from httpx import USE_CLIENT_DEFAULT
from httpx._types import FileContent
from openai.types.file_deleted import FileDeleted
from pydantic import TypeAdapter
from typing_extensions import ReadOnly
import litellm
@ -2292,7 +2293,9 @@ class BaseLLMHTTPHandler:
native_output_fallback_params: Final = anthropic_messages_provider_config.native_output_fallback_litellm_params(
model=model,
anthropic_messages_optional_request_params=_OPTIONAL_PARAMS.validate_python(anthropic_messages_optional_request_params),
anthropic_messages_optional_request_params=_OPTIONAL_PARAMS.validate_python(
anthropic_messages_optional_request_params
),
litellm_params=litellm_params,
)
rebuild_without_native_output: Final = (
@ -2302,7 +2305,9 @@ class BaseLLMHTTPHandler:
anthropic_messages_provider_config.transform_anthropic_messages_request,
model=model,
messages=copy.deepcopy(_JSON_OBJECTS.validate_python(messages)),
anthropic_messages_optional_request_params=copy.deepcopy(_JSON_OBJECT.validate_python(anthropic_messages_optional_request_params)),
anthropic_messages_optional_request_params=copy.deepcopy(
_JSON_OBJECT.validate_python(anthropic_messages_optional_request_params)
),
litellm_params=native_output_fallback_params,
headers=_JSON_OBJECT.validate_python(headers),
)

2
tests/integration/providers/test_vertex_anthropic_structured_output_wire.py
View file

@ -24,11 +24,11 @@ import pytest
import yaml
from integration._support.client import Gateway, Scenario, eventually, list_value, object_value, string_value
from integration._support.database import read_rows
from integration._support.ports import allocate_port
from integration._support.process import OwnedProxy, graceful_stop_seconds, owned_proxy_process
from integration._support.upstream import aws_event_stream_frame
from integration._support.vertex import service_account_json
from integration._support.wire import Reply, Request, Wire, wire_server
from integration._support.ports import allocate_port
from pydantic import JsonValue, TypeAdapter
_JSON_TOOL: Final = "json_tool_call"

4
tests/integration/sdk/test_bedrock_native_output_register_model_sdk.py
View file

@ -10,7 +10,9 @@ from pydantic import JsonValue, TypeAdapter
import litellm
from litellm import ModelResponse, Router
from litellm.utils import _invalidate_model_cost_lowercase_map # pyright: ignore[reportPrivateUsage] # required cache invalidation
from litellm.utils import (
_invalidate_model_cost_lowercase_map, # pyright: ignore[reportPrivateUsage] # required cache invalidation
)
_OPUS_4_8: Final = "us.anthropic.claude-opus-4-8"
_JSON_TOOL: Final = "json_tool_call"

4
tests/unit/conftest.py
View file

@ -253,7 +253,9 @@ def isolate_litellm_globals() -> Iterator[None]:
original_callbacks: Final = {name: list(getattr(litellm, name) or []) for name in CALLBACK_LISTS}
original_reset: Final = {name: getattr(litellm, name) for name in RESET_TO_NONE_GLOBALS}
original_restored: Final = {name: getattr(litellm, name) for name in RESTORED_GLOBALS if hasattr(litellm, name)}
original_clients: Final = {name: litellm.__dict__[name] for name in MODULE_LEVEL_CLIENTS if name in litellm.__dict__}
original_clients: Final = {
name: litellm.__dict__[name] for name in MODULE_LEVEL_CLIENTS if name in litellm.__dict__
}
original_loggers: Final = {
logger: (logger.level, logger.disabled, logger.propagate, list(logger.handlers), list(logger.filters))
for logger in ALL_LOGGERS

64
tests/unit/litellm_core_utils/prompt_templates/test_litellm_core_utils_prompt_templates_common_utils.py
View file

@ -2298,23 +2298,66 @@ def _object_with(properties: Mapping[str, object], **defs: Mapping[str, object])
("schema", "unresolvable"),
[
(_object_with({"a": {"type": "string"}}), False),
(_object_with({"a": {"$ref": "#/$defs/A"}, "b": {"$ref": "#/$defs/A"}}, **{"$defs": {"A": {"type": "string"}}}), False),
(_object_with({"a": {"$ref": "#/definitions/A"}}, definitions={"A": {"$ref": "#/definitions/B"}, "B": {"type": "integer"}}), False),
(_object_with({"a": {"$ref": "/$defs/A"}}, **{"$defs": {"A": {"items": {"$ref": "/$defs/B"}}, "B": {"type": "string"}}}), False),
(
_object_with(
{"a": {"$ref": "#/$defs/A"}, "b": {"$ref": "#/$defs/A"}}, **{"$defs": {"A": {"type": "string"}}}
),
False,
),
(
_object_with(
{"a": {"$ref": "#/definitions/A"}},
definitions={"A": {"$ref": "#/definitions/B"}, "B": {"type": "integer"}},
),
False,
),
(
_object_with(
{"a": {"$ref": "/$defs/A"}},
**{"$defs": {"A": {"items": {"$ref": "/$defs/B"}}, "B": {"type": "string"}}},
),
False,
),
(_object_with({"a": {"$ref": "#/$defs/Missing"}}, **{"$defs": {"A": {"type": "string"}}}), True),
(_object_with({"a": {"$ref": "other.json#/A"}}), True),
(_object_with({"parent": {"$ref": "#"}}), True),
(_object_with({"a": {"$ref": "#/$defs/A"}}, **{"$defs": {"A": {"type": "array", "items": {"$ref": "#/$defs/A"}}}}), True),
(_object_with({"a": {"$ref": "/$defs/A"}}, **{"$defs": {"A": {"type": "array", "items": {"$ref": "/$defs/A"}}}}), True),
(
_object_with(
{"a": {"$ref": "#/$defs/A"}}, **{"$defs": {"A": {"type": "array", "items": {"$ref": "#/$defs/A"}}}}
),
True,
),
(
_object_with(
{"a": {"$ref": "/$defs/A"}}, **{"$defs": {"A": {"type": "array", "items": {"$ref": "/$defs/A"}}}}
),
True,
),
(
_object_with(
{"a": {"$ref": "#/$defs/A"}},
**{"$defs": {"A": {"properties": {"b": {"$ref": "#/$defs/B"}}}, "B": {"properties": {"a": {"$ref": "#/$defs/A"}}}}},
**{
"$defs": {
"A": {"properties": {"b": {"$ref": "#/$defs/B"}}},
"B": {"properties": {"a": {"$ref": "#/$defs/A"}}},
}
},
),
True,
),
],
ids=["no-refs", "shared-def", "chained-definitions", "root-path-defs", "missing-def", "external", "root", "self-cycle", "root-path-self-cycle", "mutual-cycle"],
ids=[
"no-refs",
"shared-def",
"chained-definitions",
"root-path-defs",
"missing-def",
"external",
"root",
"self-cycle",
"root-path-self-cycle",
"mutual-cycle",
],
)
def test_schema_has_unresolvable_ref_flags_refs_that_cannot_be_inlined(schema, unresolvable):
assert schema_has_unresolvable_ref(schema) is unresolvable
@ -2323,7 +2366,11 @@ def test_schema_has_unresolvable_ref_flags_refs_that_cannot_be_inlined(schema, u
@pytest.mark.parametrize(
("schema", "properties", "branches"),
[
(_object_with({"a": {"type": "string"}, "b": {"type": "string"}, "c": {"type": "string"}}, required=["a"]), 3, 8),
(
_object_with({"a": {"type": "string"}, "b": {"type": "string"}, "c": {"type": "string"}}, required=["a"]),
3,
8,
),
(
_object_with(
{"x": {"$ref": "#/$defs/Pair"}, "y": {"$ref": "#/$defs/Pair"}},
@ -2340,4 +2387,3 @@ def test_schema_has_unresolvable_ref_flags_refs_that_cannot_be_inlined(schema, u
def test_schema_size_counts_every_written_object_once(schema, properties, branches):
assert schema_property_count(schema) == properties
assert schema_grammar_branch_count(schema) == branches

77
tests/unit/litellm_core_utils/prompt_templates/test_litellm_core_utils_prompt_templates_factory.py
View file

@ -1,61 +1,61 @@
import asyncio, base64, importlib, uuid
import asyncio
import base64
import importlib
import json
import logging
import os
import re
import uuid
from datetime import datetime
from typing import Final, List
from unittest.mock import MagicMock, patch
import pytest
import litellm
from litellm.litellm_core_utils.logging_worker import GLOBAL_LOGGING_WORKER
from litellm.litellm_core_utils.prompt_templates.common_utils import get_completion_messages
from litellm.litellm_core_utils.prompt_templates.factory import (
BEDROCK_DOCUMENT_PLACEHOLDER_TEXT,
THOUGHT_SIGNATURE_SEPARATOR,
BedrockConverseMessagesProcessor,
BedrockImageProcessor,
_bedrock_converse_messages_pt,
_bedrock_tools_pt,
_rename_duplicate_bedrock_document_names,
_convert_to_bedrock_tool_call_invoke,
_sanitize_anthropic_tool_use_id,
_convert_to_bedrock_tool_call_result,
_rename_duplicate_bedrock_document_names,
_sanitize_anthropic_tool_use_id,
anthropic_messages_pt,
convert_to_anthropic_tool_result,
convert_to_gemini_tool_call_result,
encode_tool_call_id_with_signature,
function_call_prompt,
get_thought_signature_from_tool,
get_tool_calls_from_response,
make_valid_bedrock_tool_name,
ollama_pt,
parse_mime_type,
sanitize_messages_for_tool_calling,
THOUGHT_SIGNATURE_SEPARATOR,
)
from litellm.types.llms.openai import ChatCompletionAssistantMessage, ChatCompletionToolMessage
from litellm.utils import (
_invalidate_model_cost_lowercase_map,
function_setup,
Rules,
validate_and_fix_openai_messages,
)
from datetime import datetime
from litellm.litellm_core_utils.logging_worker import GLOBAL_LOGGING_WORKER
from tests._vcr_conftest_common import install_live_call_probe, record_vcr_outcome
from litellm.litellm_core_utils.prompt_templates.factory import (
anthropic_pt,
claude_2_1_pt,
convert_to_anthropic_image_obj,
convert_to_anthropic_tool_invoke,
convert_to_anthropic_tool_result,
convert_to_gemini_tool_call_result,
convert_url_to_base64,
create_anthropic_image_param,
encode_tool_call_id_with_signature,
function_call_prompt,
get_thought_signature_from_tool,
get_tool_calls_from_response,
has_tool_with_name,
llama_2_chat_pt,
make_valid_bedrock_tool_name,
ollama_pt,
parse_mime_type,
prompt_factory,
sanitize_messages_for_tool_calling,
)
from litellm.litellm_core_utils.prompt_templates.common_utils import get_completion_messages
from litellm.llms.vertex_ai.gemini.transformation import gemini_convert_messages_with_history
from litellm.types.llms.openai import AllMessageValues
from litellm.types.llms.openai import AllMessageValues, ChatCompletionAssistantMessage, ChatCompletionToolMessage
from litellm.utils import (
Rules,
_invalidate_model_cost_lowercase_map,
function_setup,
validate_and_fix_openai_messages,
)
from tests._vcr_conftest_common import install_live_call_probe, record_vcr_outcome
def test_function_call_prompt_preserves_append_failure_for_non_string_content() -> None:
@ -208,7 +208,11 @@ def test_ollama_pt_consecutive_user_messages():
def _ollama_tool_turn(*results: object) -> list[dict]:
call: Final = {"id": "call_1", "type": "function", "function": {"name": "get_weather", "arguments": '{"city": "Paris"}'}}
call: Final = {
"id": "call_1",
"type": "function",
"function": {"name": "get_weather", "arguments": '{"city": "Paris"}'},
}
return [
{"role": "user", "content": "Weather in Paris?"},
{"role": "assistant", "content": None, "tool_calls": [call]},
@ -2148,8 +2152,8 @@ def test_bedrock_tools_unpack_defs_no_oom_with_nested_refs():
This test creates a schema with multiple nested $defs that reference each other
to verify the fix prevents memory explosion while still correctly resolving refs.
"""
import sys
import copy
import sys
from litellm.litellm_core_utils.prompt_templates.factory import _bedrock_tools_pt
@ -4366,7 +4370,9 @@ def test_bedrock_converse_messages_pt_user_message_without_content_adds_no_block
assert _bedrock_converse_messages_pt(
messages=with_message, model="anthropic.claude-haiku-4-5", llm_provider="bedrock"
) == _bedrock_converse_messages_pt(messages=without_message, model="anthropic.claude-haiku-4-5", llm_provider="bedrock")
) == _bedrock_converse_messages_pt(
messages=without_message, model="anthropic.claude-haiku-4-5", llm_provider="bedrock"
)
@pytest.mark.asyncio
@ -4496,6 +4502,7 @@ def _vcr_outcome_gate(request, vcr):
yield
record_vcr_outcome(request, vcr)
@pytest.fixture(scope="function")
def isolate_litellm_state():
"""
@ -4548,6 +4555,7 @@ def isolate_litellm_state():
setattr(litellm, attr, original_value)
_invalidate_model_cost_lowercase_map()
_SCALAR_DEFAULTS = {
"num_retries": getattr(litellm, "num_retries", None),
"num_retries_per_request": getattr(litellm, "num_retries_per_request", None),
@ -4568,6 +4576,7 @@ _SCALAR_DEFAULTS = {
"api_key": getattr(litellm, "api_key", None),
}
@pytest.fixture(scope="module")
def setup_and_teardown():
"""
@ -4590,6 +4599,7 @@ def setup_and_teardown():
litellm.in_memory_llm_clients_cache.flush_cache()
yield
@pytest.mark.usefixtures("_vcr_outcome_gate", "isolate_litellm_state", "setup_and_teardown")
def test_function_call_non_openai_model():
try:
@ -4618,6 +4628,7 @@ def test_function_call_non_openai_model():
print(e)
pass
@pytest.mark.usefixtures("_vcr_outcome_gate", "isolate_litellm_state", "setup_and_teardown")
def test_empty_content():
"""
@ -4636,6 +4647,7 @@ def test_empty_content():
litellm_call_id=str(uuid.uuid4()),
)
@pytest.mark.usefixtures("_vcr_outcome_gate", "isolate_litellm_state", "setup_and_teardown")
def test_thought_signature_removal_for_non_gemini():
"""
@ -4684,6 +4696,7 @@ def test_thought_signature_removal_for_non_gemini():
assert THOUGHT_SIGNATURE_SEPARATOR not in processed_messages[1]["tool_calls"][0]["id"]
assert THOUGHT_SIGNATURE_SEPARATOR not in processed_messages[2]["tool_call_id"]
@pytest.mark.usefixtures("_vcr_outcome_gate", "isolate_litellm_state", "setup_and_teardown")
def test_thought_signature_preserved_for_gemini():
"""
@ -4730,6 +4743,7 @@ def test_thought_signature_preserved_for_gemini():
assert THOUGHT_SIGNATURE_SEPARATOR in processed_messages[1]["tool_calls"][0]["id"]
assert THOUGHT_SIGNATURE_SEPARATOR in processed_messages[2]["tool_call_id"]
@pytest.mark.usefixtures("_vcr_outcome_gate", "isolate_litellm_state", "setup_and_teardown")
def test_thought_signature_removal_with_multiple_tool_calls():
"""
@ -4784,6 +4798,7 @@ def test_thought_signature_removal_with_multiple_tool_calls():
assert processed_messages[2]["tool_call_id"] == "call_1"
assert processed_messages[3]["tool_call_id"] == "call_2"
@pytest.mark.usefixtures("_vcr_outcome_gate", "isolate_litellm_state", "setup_and_teardown")
def test_messages_without_tool_calls_unchanged():
"""

57
tests/unit/llms/anthropic/chat/test_anthropic_chat_transformation.py
View file

@ -4,9 +4,9 @@ from typing import Final
from unittest.mock import MagicMock, patch
import httpx
from httpx import Headers
import pytest
import respx
from httpx import Headers
import litellm
from litellm.constants import (
@ -21,8 +21,8 @@ from litellm.constants import (
from litellm.litellm_core_utils.prompt_templates.common_utils import encrypted_reasoning_signature
from litellm.litellm_core_utils.prompt_templates.factory import anthropic_messages_pt
from litellm.llms.anthropic.chat import ModelResponseIterator
from litellm.llms.anthropic.common_utils import process_anthropic_headers
from litellm.llms.anthropic.chat.transformation import AnthropicConfig
from litellm.llms.anthropic.common_utils import process_anthropic_headers
from litellm.llms.anthropic.pass_through.messages.transformation import (
AnthropicMessagesConfig,
)
@ -1336,7 +1336,6 @@ def test_dict_fields_take_the_json_tool_on_a_native_structured_output_model(fiel
assert ([tool["name"] for tool in optional_params.get("tools", [])] == ["json_tool_call"]) is not native
@pytest.mark.parametrize(
("model", "thinking_params"),
[
@ -1349,7 +1348,13 @@ def test_dict_fields_take_the_json_tool_on_a_native_structured_output_model(fiel
{"tools": [{"type": "function", "function": {"name": "lookup", "parameters": {"type": "object"}}}]},
),
],
ids=["adaptive-thinking", "reasoning-effort", "reasoning-effort-none", "model-without-forced-tools", "caller-tools"],
ids=[
"adaptive-thinking",
"reasoning-effort",
"reasoning-effort-none",
"model-without-forced-tools",
"caller-tools",
],
)
def test_dict_fields_keep_native_output_when_the_json_tool_would_not_be_forced(model, thinking_params):
optional_params: Final = AnthropicConfig().map_openai_params(
@ -1375,6 +1380,7 @@ def test_dict_fields_keep_native_output_when_the_json_tool_would_not_be_forced(m
assert "output_format" in optional_params
assert "json_tool_call" not in [tool["name"] for tool in optional_params.get("tools", [])]
def test_native_structured_output_uses_bundled_capability_when_remote_map_lags(
monkeypatch: pytest.MonkeyPatch,
) -> None:
@ -7640,19 +7646,18 @@ async def test_anthropic_api_max_completion_tokens(model: str):
print("request_body: ", request_body)
assert request_body == {
"messages": [
{"role": "user", "content": [{"type": "text", "text": "Hello!"}]}
],
"messages": [{"role": "user", "content": [{"type": "text", "text": "Hello!"}]}],
"max_tokens": 10,
"model": model.split("/")[-1],
}
def test_anthropic_tool_cache_control():
from litellm.utils import return_raw_request
from litellm.types.utils import CallTypes
import json
from litellm.types.utils import CallTypes
from litellm.utils import return_raw_request
tool_content = "Result: 4. " * 1000
messages = [
{"role": "user", "content": "Calculate 2+2"},
@ -7860,9 +7865,7 @@ async def test_anthropic_api_prompt_caching_with_content_str():
"cache_control": {"type": "ephemeral"},
},
]
translated_system_message = litellm.AnthropicConfig().translate_system_message(
messages=system_message
)
translated_system_message = litellm.AnthropicConfig().translate_system_message(messages=system_message)
assert translated_system_message == [
# System Message
@ -7932,9 +7935,9 @@ async def test_anthropic_api_prompt_caching_with_content_str():
assert len(translated_messages) == len(expected_messages)
for idx, i in enumerate(translated_messages):
assert (
i == expected_messages[idx]
), "Error on idx={}. Got={}, Expected={}".format(idx, i, expected_messages[idx])
assert i == expected_messages[idx], "Error on idx={}. Got={}, Expected={}".format(
idx, i, expected_messages[idx]
)
@pytest.fixture
@ -8048,3 +8051,27 @@ def test_json_mode_keeps_the_thinking_that_preceded_the_json_tool_call():
assert message.reasoning_content == thinking_block["thinking"]
assert [block["signature"] for block in message.thinking_blocks] == [thinking_block["signature"]]
def _json_mode_message_body(stop_reason: str, stop_details: dict | None) -> dict:
return {
"id": "msg_json_mode",
"type": "message",
"role": "assistant",
"model": "claude-fable-5",
"stop_reason": stop_reason,
"stop_details": stop_details,
"usage": {"input_tokens": 12, "output_tokens": 8},
"content": [
{"type": "tool_use", "id": "toolu_json", "name": RESPONSE_FORMAT_TOOL_NAME, "input": {"answer": "Hello"}}
],
}
def _transform_json_mode_body(body: dict) -> litellm.ModelResponse:
return AnthropicConfig().transform_parsed_response(
completion_response=body,
raw_response=httpx.Response(status_code=200, headers={}),
model_response=litellm.ModelResponse(),
json_mode=True,
prefix_prompt=None,
)

1
tests/unit/llms/anthropic/test_anthropic_schema_filter.py
View file

@ -5,6 +5,7 @@ Related to: https://platform.claude.com/docs/en/build-with-claude/structured-out
"""
import pytest
from litellm.llms.anthropic.chat.transformation import AnthropicConfig

15
tests/unit/llms/bedrock/chat/invoke_transformations/test_bedrock_chat_invoke_transformations_anthropic_claude3_transformation.py
View file

@ -1,9 +1,11 @@
import asyncio, base64, importlib
import asyncio
import base64
import copy
import importlib
import json
import uuid
from types import SimpleNamespace
from collections.abc import Mapping
from types import SimpleNamespace
from typing import Final
from unittest.mock import patch
@ -634,7 +636,6 @@ def test_shared_defs_reach_native_output_once_instead_of_inlined_per_use(set_nat
assert "tools" not in result
class Relabel(BaseModel):
model_config = ConfigDict(frozen=True)
@ -671,6 +672,7 @@ def test_nested_pydantic_response_format_reaches_native_output_with_refs_that_re
assert item_ref.removeprefix("#/$defs/") in sent["$defs"]
assert "tools" not in result
def test_output_config_format_forwarded_for_bedrock_chat_invoke_request(local_model_cost_map):
"""Bedrock Invoke chat path forwards ``output_config.format`` alongside effort
for models with native structured-output support (Claude Sonnet 4.6)."""
@ -1386,6 +1388,7 @@ def _vcr_outcome_gate(request, vcr):
yield
record_vcr_outcome(request, vcr)
@pytest.fixture(scope="session")
def event_loop():
try:
@ -1395,6 +1398,7 @@ def event_loop():
yield loop
loop.close()
@pytest.fixture(scope="function")
def setup_and_teardown(event_loop):
import litellm
@ -1428,6 +1432,7 @@ def setup_and_teardown(event_loop):
if pending:
event_loop.run_until_complete(asyncio.gather(*pending, return_exceptions=True))
_SCALAR_DEFAULTS = {
"num_retries": getattr(litellm, "num_retries", None),
"set_verbose": getattr(litellm, "set_verbose", False),
@ -1499,6 +1504,7 @@ IN WITNESS WHEREOF, the parties have executed this Agreement.
* 8
) # Repeat to ensure we have enough tokens (need 1024+ for Claude models)
@pytest.mark.usefixtures("_vcr_outcome_gate", "setup_and_teardown")
class TestBedrockAnthropicPromptCachingRegression:
"""
@ -1661,6 +1667,7 @@ class TestBedrockAnthropicPromptCachingRegression:
if "anthropic_beta" in additional_fields:
assert "prompt-caching-2024-07-31" not in additional_fields["anthropic_beta"]
@pytest.mark.usefixtures("_vcr_outcome_gate", "setup_and_teardown")
class TestBedrockAnthropic1MContextRegression:
"""
@ -1846,6 +1853,7 @@ class TestBedrockAnthropic1MContextRegression:
assert "context-1m-2025-08-07" in beta_headers
assert "computer-use-2024-10-22" in beta_headers
@pytest.mark.usefixtures("_vcr_outcome_gate", "setup_and_teardown")
class TestBedrockAnthropicCombinedRegressions:
"""
@ -1992,4 +2000,3 @@ def test_invoke_uses_opt_in_native_output_is_false_when_native_output_did_not_co
assert not AmazonAnthropicClaudeConfig().uses_opt_in_native_output(
model, optional_params, {"model_info": model_info}
)

621
tests/unit/llms/bedrock/chat/test_converse_transformation.py
View file

File diff suppressed because it is too large Load diff

397
tests/unit/llms/bedrock/files/test_bedrock_files_transformation.py
View file

@ -34,9 +34,7 @@ class TestBedrockFilesTransformation:
transformation = BedrockJsonlFilesTransformation()
# Load input JSONL file
input_file_path = os.path.join(
os.path.dirname(__file__), "input_batch_completions.jsonl"
)
input_file_path = os.path.join(os.path.dirname(__file__), "input_batch_completions.jsonl")
# Read and parse the JSONL content
openai_jsonl_content = []
@ -131,28 +129,20 @@ class TestBedrockFilesTransformation:
model_input = record["modelInput"]
# Must have inferenceConfig with maxTokens, NOT top-level max_tokens
assert (
"inferenceConfig" in model_input
), "Nova modelInput must contain inferenceConfig"
assert "inferenceConfig" in model_input, "Nova modelInput must contain inferenceConfig"
assert model_input["inferenceConfig"]["maxTokens"] == 50
assert model_input["inferenceConfig"]["temperature"] == 0.7
assert (
"max_tokens" not in model_input
), "max_tokens must NOT be at the top level for Nova"
assert (
"temperature" not in model_input
), "temperature must NOT be at the top level for Nova"
assert "max_tokens" not in model_input, "max_tokens must NOT be at the top level for Nova"
assert "temperature" not in model_input, "temperature must NOT be at the top level for Nova"
# Must have messages
assert "messages" in model_input
# Nova Pro rejects empty additionalModelRequestFields / system — they must be absent
assert (
"additionalModelRequestFields" not in model_input
), "Nova: empty additionalModelRequestFields must be omitted, not serialized as {}"
assert (
"system" not in model_input
), "Nova: empty system must be omitted, not serialized as []"
assert "additionalModelRequestFields" not in model_input, (
"Nova: empty additionalModelRequestFields must be omitted, not serialized as {}"
)
assert "system" not in model_input, "Nova: empty system must be omitted, not serialized as []"
def test_nova_batch_jsonl_omits_empty_converse_fields(self):
"""
@ -220,10 +210,7 @@ class TestBedrockFilesTransformation:
config = BedrockFilesConfig()
# 1x1 transparent PNG
img_b64 = (
"iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJAAAADUlEQVR4"
"2mP8z8BQDwADhQGAWjR9awAAAABJRU5ErkJggg=="
)
img_b64 = "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJAAAADUlEQVR42mP8z8BQDwADhQGAWjR9awAAAABJRU5ErkJggg=="
openai_jsonl_content = [
{
@ -239,9 +226,7 @@ class TestBedrockFilesTransformation:
{"type": "text", "text": "Describe this image."},
{
"type": "image_url",
"image_url": {
"url": "data:image/png;base64," + img_b64
},
"image_url": {"url": "data:image/png;base64," + img_b64},
},
],
}
@ -277,16 +262,10 @@ class TestBedrockFilesTransformation:
# Verify Converse image format
assert "format" in block["image"], "Image block must have format field"
assert "source" in block["image"], "Image block must have source field"
assert (
"bytes" in block["image"]["source"]
), "Image source must have bytes field"
assert "bytes" in block["image"]["source"], "Image source must have bytes field"
# Must NOT have OpenAI-style image_url
assert (
"image_url" not in block
), "image_url must not appear in Converse format"
assert (
block.get("type") != "image_url"
), "type=image_url must not appear in Converse format"
assert "image_url" not in block, "image_url must not appear in Converse format"
assert block.get("type") != "image_url", "type=image_url must not appear in Converse format"
assert has_text, "Should have a text content block"
assert has_image, "Should have an image content block"
@ -369,9 +348,7 @@ class TestBedrockFilesTransformation:
)
assert "us-gov-west-1" in url, f"Expected us-gov-west-1 in URL but got: {url}"
assert (
"us-west-2" not in url
), f"us-west-2 must not appear when s3_region_name is set, got: {url}"
assert "us-west-2" not in url, f"us-west-2 must not appear when s3_region_name is set, got: {url}"
assert "litellm-batch-352026" in url
def test_get_complete_file_url_sanitizes_untrusted_filename(self):
@ -403,9 +380,7 @@ class TestBedrockFilesTransformation:
from litellm.llms.bedrock.files.transformation import BedrockFilesConfig
config = BedrockFilesConfig()
object_name = config._get_s3_object_name_from_batch_jsonl(
[{"body": {"model": "bedrock/../../secret:model"}}]
)
object_name = config._get_s3_object_name_from_batch_jsonl([{"body": {"model": "bedrock/../../secret:model"}}])
assert object_name.startswith("litellm-bedrock-files-")
assert object_name.endswith(".jsonl")
@ -460,9 +435,9 @@ class TestBedrockFilesTransformation:
litellm_params=litellm_params,
)
assert (
captured_optional_params.get("aws_region_name") == "us-gov-west-1"
), "s3_region_name must be forwarded as aws_region_name for SigV4 signing"
assert captured_optional_params.get("aws_region_name") == "us-gov-west-1", (
"s3_region_name must be forwarded as aws_region_name for SigV4 signing"
)
def test_s3_region_name_wins_over_aws_region_name_for_signing(self):
"""
@ -516,9 +491,9 @@ class TestBedrockFilesTransformation:
litellm_params=litellm_params,
)
assert (
captured_optional_params.get("aws_region_name") == "us-gov-west-1"
), "s3_region_name must override aws_region_name for SigV4 signing"
assert captured_optional_params.get("aws_region_name") == "us-gov-west-1", (
"s3_region_name must override aws_region_name for SigV4 signing"
)
def _signed_upload_request(self, litellm_params: dict) -> dict:
from litellm.llms.bedrock.files.transformation import BedrockFilesConfig
@ -1109,9 +1084,9 @@ class TestBedrockFilesEmbeddingTransformation:
}
]
)
assert result[0]["modelInput"] == {
"inputText": "Hi"
}, f"model id {model_id} did not route to Titan v2 embedding path"
assert result[0]["modelInput"] == {"inputText": "Hi"}, (
f"model id {model_id} did not route to Titan v2 embedding path"
)
def test_pretokenized_input_list_of_ints_raises(self):
"""`input: List[int]` (pre-tokenized) is rejected, not silently mis-shaped."""
@ -1179,7 +1154,6 @@ class TestBedrockFilesEmbeddingTransformation:
assert "messages" in result[0]["modelInput"]
assert "inputText" not in result[0]["modelInput"]
def test_titan_v2_marker_boundary_rejects_lookalikes(self):
"""The marker must end at `:`, `/`, or end-of-string to avoid false positives."""
from litellm.llms.bedrock.files.transformation import BedrockFilesConfig
@ -1190,9 +1164,9 @@ class TestBedrockFilesEmbeddingTransformation:
"bedrock/amazon.titan-embed-text-v2-experimental:0",
"bedrock/amazon.titan-embed-text-v2foo",
):
assert not BedrockFilesConfig._is_titan_v2_embed_model(
model
), f"{model} unexpectedly matched the Titan v2 marker"
assert not BedrockFilesConfig._is_titan_v2_embed_model(model), (
f"{model} unexpectedly matched the Titan v2 marker"
)
# Real Titan v2 ids that MUST match
for model in (
@ -1201,9 +1175,9 @@ class TestBedrockFilesEmbeddingTransformation:
"us.amazon.titan-embed-text-v2:0",
"arn:aws:bedrock:us-east-1:123:foundation-model/amazon.titan-embed-text-v2:0",
):
assert BedrockFilesConfig._is_titan_v2_embed_model(
model
), f"{model} unexpectedly missed the Titan v2 marker"
assert BedrockFilesConfig._is_titan_v2_embed_model(model), (
f"{model} unexpectedly missed the Titan v2 marker"
)
def test_titan_v2_accepted_when_registry_schema_field_matches(self, mocker):
"""Registry-driven happy path: nested
@ -1213,13 +1187,9 @@ class TestBedrockFilesEmbeddingTransformation:
mocker.patch(
"litellm.get_model_info",
return_value={
"provider_specific_entry": {"bedrock_invocation_schema": "titan_v2"}
},
)
assert BedrockFilesConfig._is_titan_v2_embed_model(
"amazon.titan-embed-text-v2:0"
return_value={"provider_specific_entry": {"bedrock_invocation_schema": "titan_v2"}},
)
assert BedrockFilesConfig._is_titan_v2_embed_model("amazon.titan-embed-text-v2:0")
def test_titan_v2_rejected_when_registry_schema_field_differs(self, mocker):
"""Registry resolves with a different schema value (e.g. a hypothetical
@ -1229,19 +1199,13 @@ class TestBedrockFilesEmbeddingTransformation:
mocker.patch(
"litellm.get_model_info",
return_value={
"provider_specific_entry": {"bedrock_invocation_schema": "cohere_v3"}
},
return_value={"provider_specific_entry": {"bedrock_invocation_schema": "cohere_v3"}},
)
# Even though the model id looks like Titan v2, the registry says
# otherwise and we trust it.
assert not BedrockFilesConfig._is_titan_v2_embed_model(
"amazon.titan-embed-text-v2:0"
)
assert not BedrockFilesConfig._is_titan_v2_embed_model("amazon.titan-embed-text-v2:0")
def test_titan_v2_falls_back_to_marker_when_registry_lacks_schema_field(
self, mocker
):
def test_titan_v2_falls_back_to_marker_when_registry_lacks_schema_field(self, mocker):
"""Registry resolves but the entry has no
`provider_specific_entry.bedrock_invocation_schema` field yet (e.g.
a stale local registry) -> fall through to substring."""
@ -1252,9 +1216,7 @@ class TestBedrockFilesEmbeddingTransformation:
"litellm.get_model_info",
return_value={"mode": "embedding"},
)
assert BedrockFilesConfig._is_titan_v2_embed_model(
"amazon.titan-embed-text-v2:0"
)
assert BedrockFilesConfig._is_titan_v2_embed_model("amazon.titan-embed-text-v2:0")
# provider_specific_entry present but missing the schema key
mocker.patch(
@ -1264,9 +1226,7 @@ class TestBedrockFilesEmbeddingTransformation:
"provider_specific_entry": {"unrelated": "value"},
},
)
assert BedrockFilesConfig._is_titan_v2_embed_model(
"amazon.titan-embed-text-v2:0"
)
assert BedrockFilesConfig._is_titan_v2_embed_model("amazon.titan-embed-text-v2:0")
def test_titan_v2_accepted_when_registry_silent(self, mocker):
"""Marker-only match is fine for ids the registry can't resolve
@ -1274,9 +1234,7 @@ class TestBedrockFilesEmbeddingTransformation:
from litellm.llms.bedrock.files.transformation import BedrockFilesConfig
mocker.patch("litellm.get_model_info", side_effect=Exception("not mapped"))
assert BedrockFilesConfig._is_titan_v2_embed_model(
"us.amazon.titan-embed-text-v2:0"
)
assert BedrockFilesConfig._is_titan_v2_embed_model("us.amazon.titan-embed-text-v2:0")
assert BedrockFilesConfig._is_titan_v2_embed_model(
"arn:aws:bedrock:us-east-1:123:foundation-model/amazon.titan-embed-text-v2:0"
)
@ -1288,87 +1246,49 @@ class TestBedrockFilesEmbeddingTransformation:
# Happy path: returns the nested field's string value
mocker.patch(
"litellm.get_model_info",
return_value={
"provider_specific_entry": {"bedrock_invocation_schema": "titan_v2"}
},
)
assert (
BedrockFilesConfig._lookup_provider_specific_field(
"anything", "bedrock_invocation_schema"
)
== "titan_v2"
return_value={"provider_specific_entry": {"bedrock_invocation_schema": "titan_v2"}},
)
assert BedrockFilesConfig._lookup_provider_specific_field("anything", "bedrock_invocation_schema") == "titan_v2"
# Registry raises -> None
mocker.patch("litellm.get_model_info", side_effect=Exception("not mapped"))
assert (
BedrockFilesConfig._lookup_provider_specific_field("anything", "any")
is None
)
assert BedrockFilesConfig._lookup_provider_specific_field("anything", "any") is None
# Registry returns non-dict -> None
mocker.patch("litellm.get_model_info", return_value="not a dict")
assert (
BedrockFilesConfig._lookup_provider_specific_field("anything", "any")
is None
)
assert BedrockFilesConfig._lookup_provider_specific_field("anything", "any") is None
# Registry returns dict without provider_specific_entry -> None
mocker.patch("litellm.get_model_info", return_value={"mode": "embedding"})
assert (
BedrockFilesConfig._lookup_provider_specific_field(
"anything", "bedrock_invocation_schema"
)
is None
)
assert BedrockFilesConfig._lookup_provider_specific_field("anything", "bedrock_invocation_schema") is None
# provider_specific_entry exists but isn't a dict -> None
mocker.patch(
"litellm.get_model_info",
return_value={"provider_specific_entry": "not a dict"},
)
assert (
BedrockFilesConfig._lookup_provider_specific_field(
"anything", "bedrock_invocation_schema"
)
is None
)
assert BedrockFilesConfig._lookup_provider_specific_field("anything", "bedrock_invocation_schema") is None
# provider_specific_entry dict missing the requested field -> None
mocker.patch(
"litellm.get_model_info",
return_value={"provider_specific_entry": {"unrelated": "x"}},
)
assert (
BedrockFilesConfig._lookup_provider_specific_field(
"anything", "bedrock_invocation_schema"
)
is None
)
assert BedrockFilesConfig._lookup_provider_specific_field("anything", "bedrock_invocation_schema") is None
# Non-string nested value -> None
mocker.patch(
"litellm.get_model_info",
return_value={"provider_specific_entry": {"bedrock_invocation_schema": 42}},
)
assert (
BedrockFilesConfig._lookup_provider_specific_field(
"anything", "bedrock_invocation_schema"
)
is None
)
assert BedrockFilesConfig._lookup_provider_specific_field("anything", "bedrock_invocation_schema") is None
# Empty-string nested value -> None
mocker.patch(
"litellm.get_model_info",
return_value={"provider_specific_entry": {"bedrock_invocation_schema": ""}},
)
assert (
BedrockFilesConfig._lookup_provider_specific_field(
"anything", "bedrock_invocation_schema"
)
is None
)
assert BedrockFilesConfig._lookup_provider_specific_field("anything", "bedrock_invocation_schema") is None
def test_classify_batch_record_helper(self):
"""Helper classifies by `url` first, then by body shape."""
@ -1376,28 +1296,18 @@ class TestBedrockFilesEmbeddingTransformation:
from litellm.types.llms.bedrock import BedrockBatchRecordKind
assert (
BedrockFilesConfig._classify_batch_record(
{"url": "/v1/embeddings", "body": {"input": "x"}}
)
BedrockFilesConfig._classify_batch_record({"url": "/v1/embeddings", "body": {"input": "x"}})
is BedrockBatchRecordKind.EMBEDDING
)
# body-only fallback
assert (
BedrockFilesConfig._classify_batch_record({"body": {"input": "x"}})
is BedrockBatchRecordKind.EMBEDDING
)
assert BedrockFilesConfig._classify_batch_record({"body": {"input": "x"}}) is BedrockBatchRecordKind.EMBEDDING
# chat shape
assert (
BedrockFilesConfig._classify_batch_record(
{"url": "/v1/chat/completions", "body": {"messages": []}}
)
BedrockFilesConfig._classify_batch_record({"url": "/v1/chat/completions", "body": {"messages": []}})
is BedrockBatchRecordKind.CHAT
)
# ambiguous body without any recognized key is treated as chat
assert (
BedrockFilesConfig._classify_batch_record({"body": {}})
is BedrockBatchRecordKind.CHAT
)
assert BedrockFilesConfig._classify_batch_record({"body": {}}) is BedrockBatchRecordKind.CHAT
@pytest.mark.parametrize("body", ["not a mapping", ["messages"], 7, None], ids=["str", "list", "int", "missing"])
def test_classify_batch_record_falls_back_to_chat_for_non_mapping_body(self, body):
@ -1437,7 +1347,6 @@ class TestBedrockFilesEmbeddingTransformation:
transformer, corrupting the modelInput.
"""
from litellm.llms.bedrock.files.transformation import BedrockFilesConfig
from litellm.types.llms.bedrock import BedrockBatchRecordKind
# Direct helper assertion
@ -1475,9 +1384,7 @@ class TestBedrockFilesEmbeddingTransformation:
)
model_input = result[0]["modelInput"]
assert (
"inputText" not in model_input
), "explicit chat URL must not produce an embedding-shaped modelInput"
assert "inputText" not in model_input, "explicit chat URL must not produce an embedding-shaped modelInput"
def test_coerce_embedding_input_helper_isolated(self):
"""Direct coverage of the extracted input-normalization helper."""
@ -1487,9 +1394,7 @@ class TestBedrockFilesEmbeddingTransformation:
# Happy paths
assert BedrockFilesConfig._coerce_embedding_input_to_string("hello") == "hello"
assert (
BedrockFilesConfig._coerce_embedding_input_to_string(["hello"]) == "hello"
)
assert BedrockFilesConfig._coerce_embedding_input_to_string(["hello"]) == "hello"
# Error paths
with pytest.raises(ValueError, match="missing required `input`"):
@ -1517,22 +1422,16 @@ class TestBedrockFilesEmbeddingTransformation:
# /v1/completions (legacy completions endpoint)
assert (
BedrockFilesConfig._classify_batch_record(
{"url": "/v1/completions", "body": {"input": "x"}}
)
BedrockFilesConfig._classify_batch_record({"url": "/v1/completions", "body": {"input": "x"}})
is BedrockBatchRecordKind.TEXT_COMPLETION
)
assert (
BedrockFilesConfig._classify_batch_record(
{"url": "/v1/responses", "body": {"input": "x"}}
)
BedrockFilesConfig._classify_batch_record({"url": "/v1/responses", "body": {"input": "x"}})
is BedrockBatchRecordKind.RESPONSES
)
# Arbitrary unknown url - caller's explicit signal still wins
assert (
BedrockFilesConfig._classify_batch_record(
{"url": "/v1/moderations", "body": {"input": "x"}}
)
BedrockFilesConfig._classify_batch_record({"url": "/v1/moderations", "body": {"input": "x"}})
is BedrockBatchRecordKind.CHAT
)
@ -1684,7 +1583,10 @@ class TestBedrockBatchNonChatEndpointRecords:
assert model_input["messages"][0]["content"] == [{"type": "text", "text": "Weather in Paris?"}]
tool = model_input["tools"][0]
function = tool.get("function", tool)
assert (function["name"], function.get("parameters", function.get("input_schema"))) == ("get_weather", parameters)
assert (function["name"], function.get("parameters", function.get("input_schema"))) == (
"get_weather",
parameters,
)
@pytest.mark.parametrize(
("url", "body"),
@ -1842,9 +1744,7 @@ class TestBedrockBatchNonChatEndpointRecords:
"embedding",
]
for record in result[:3]:
assert record["modelInput"]["messages"] == [
{"role": "user", "content": [{"type": "text", "text": "hi"}]}
]
assert record["modelInput"]["messages"] == [{"role": "user", "content": [{"type": "text", "text": "hi"}]}]
assert result[3]["modelInput"] == {"inputText": "hi"}
@pytest.mark.parametrize(
@ -1889,9 +1789,7 @@ class TestBedrockBatchNonChatEndpointRecords:
}
)
assert model_input["messages"] == [
{"role": "user", "content": [{"type": "text", "text": "from messages"}]}
]
assert model_input["messages"] == [{"role": "user", "content": [{"type": "text", "text": "from messages"}]}]
class TestBedrockBatchAnthropicRowParams:
@ -1972,7 +1870,10 @@ class TestBedrockBatchAnthropicRowParams:
@pytest.mark.parametrize(
("url", "body"),
[
("/v1/chat/completions", {"messages": [{"role": "user", "content": "17 * 23?"}], "reasoning_effort": "low"}),
(
"/v1/chat/completions",
{"messages": [{"role": "user", "content": "17 * 23?"}], "reasoning_effort": "low"},
),
("/v1/responses", {"input": "17 * 23?", "reasoning": {"effort": "low"}}),
],
ids=["chat", "responses"],
@ -2132,16 +2033,24 @@ class TestBedrockFileDeletion:
with respx.mock, closing(HTTPHandler()) as client:
route: Final = respx.delete(self.URL).mock(return_value=httpx.Response(204))
deleted: Final = litellm.file_delete(
file_id=self.S3_URI, custom_llm_provider="bedrock", client=client,
aws_access_key_id="AKIAEXAMPLE", aws_secret_access_key="test-secret", aws_region_name="us-west-2",
file_id=self.S3_URI,
custom_llm_provider="bedrock",
client=client,
aws_access_key_id="AKIAEXAMPLE",
aws_secret_access_key="test-secret",
aws_region_name="us-west-2",
)
assert route.call_count == 1
request: Final = route.calls[0].request
assert request.content == b""
signed: Final = AWSRequest(method="DELETE", url=self.URL, headers={
"X-Amz-Date": request.headers["X-Amz-Date"],
"X-Amz-Content-SHA256": request.headers["X-Amz-Content-SHA256"],
})
signed: Final = AWSRequest(
method="DELETE",
url=self.URL,
headers={
"X-Amz-Date": request.headers["X-Amz-Date"],
"X-Amz-Content-SHA256": request.headers["X-Amz-Content-SHA256"],
},
)
signed.context["timestamp"] = request.headers["X-Amz-Date"]
auth: Final = S3SigV4Auth(Credentials("AKIAEXAMPLE", "test-secret"), "s3", "us-west-2")
signature: Final = auth.signature(auth.string_to_sign(signed, auth.canonical_request(signed)), signed)
@ -2169,15 +2078,22 @@ class TestBedrockFileDeletion:
with pytest.raises(BedrockError, match="AccessDenied"):
await litellm.afile_delete(
file_id=self.S3_URI, custom_llm_provider="bedrock", client=client,
aws_access_key_id="AKIAEXAMPLE", aws_secret_access_key="test-secret", aws_region_name="us-west-2",
file_id=self.S3_URI,
custom_llm_provider="bedrock",
client=client,
aws_access_key_id="AKIAEXAMPLE",
aws_secret_access_key="test-secret",
aws_region_name="us-west-2",
)
assert route.call_count == 1
@pytest.mark.parametrize("file_id, message", [
("s3://other-bucket/litellm-bedrock-files-model-abc.jsonl", "configured storage bucket"),
("s3://my-bucket/private/data.jsonl", "LiteLLM-managed"),
])
@pytest.mark.parametrize(
"file_id, message",
[
("s3://other-bucket/litellm-bedrock-files-model-abc.jsonl", "configured storage bucket"),
("s3://my-bucket/private/data.jsonl", "LiteLLM-managed"),
],
)
def test_delete_rejects_untrusted_objects_before_signing(
self, file_id: str, message: str, monkeypatch: pytest.MonkeyPatch
) -> None:
@ -2225,11 +2141,7 @@ class TestBedrockFileContentTransformation:
assert params == {}
signed_headers = litellm_params[S3_SIGNED_REQUEST_HEADERS_PARAM]
content_hashes = {
value
for name, value in signed_headers.items()
if name.lower() == "x-amz-content-sha256"
}
content_hashes = {value for name, value in signed_headers.items() if name.lower() == "x-amz-content-sha256"}
assert content_hashes == {hashlib.sha256(b"").hexdigest()}, (
"GET has no payload, so the content hash must be the empty-body hash."
" The header name is matched case-insensitively because botocore picks"
@ -2462,9 +2374,7 @@ class TestBedrockFileContentTransformation:
unified_file_id = SpecialEnums.LITELLM_MANAGED_FILE_COMPLETE_STR.value.format(
"application/json", "unified-id", "", self.S3_URI, "model-id"
)
encoded_file_id = (
base64.urlsafe_b64encode(unified_file_id.encode()).decode().rstrip("=")
)
encoded_file_id = base64.urlsafe_b64encode(unified_file_id.encode()).decode().rstrip("=")
url, _ = BedrockFilesConfig().transform_file_content_request(
file_content_request={"file_id": encoded_file_id},
@ -2482,16 +2392,13 @@ class TestBedrockFileContentTransformation:
with pytest.raises(BedrockError, match="configured storage bucket") as rejection:
BedrockFilesConfig().transform_file_content_request(
file_content_request={
"file_id": "s3://other-bucket/litellm-batch-outputs/job/x.jsonl.out"
},
file_content_request={"file_id": "s3://other-bucket/litellm-batch-outputs/job/x.jsonl.out"},
optional_params={},
litellm_params=self._litellm_params(),
)
assert rejection.value.status_code == 400
def test_transform_file_content_request_rejects_unmanaged_key(self, monkeypatch):
from litellm.llms.bedrock.common_utils import BedrockError
from litellm.llms.bedrock.files.transformation import BedrockFilesConfig
@ -2507,7 +2414,6 @@ class TestBedrockFileContentTransformation:
assert rejection.value.status_code == 400
def test_extract_s3_uri_rejects_non_managed_file_id(self):
"""A file id that is neither an s3:// URI nor a unified id must be rejected."""
from litellm.llms.bedrock.files.transformation import (
@ -2517,9 +2423,7 @@ class TestBedrockFileContentTransformation:
with pytest.raises(ValueError, match="managed LiteLLM S3 file id"):
extract_s3_uri_from_file_id("file-1234567890")
def test_transform_file_content_request_requires_configured_bucket(
self, monkeypatch
):
def test_transform_file_content_request_requires_configured_bucket(self, monkeypatch):
"""Without a server-configured bucket (env or snapshot), the request must fail
before any S3 call rather than guessing a bucket from the file id."""
from litellm.llms.bedrock.files.transformation import BedrockFilesConfig
@ -2556,9 +2460,7 @@ class TestBedrockFileContentTransformation:
from litellm.types.router import CredentialLiteLLMParams
snapshot = CredentialLiteLLMParams(**deployment_litellm_params).model_dump(
exclude_none=True
)
snapshot = CredentialLiteLLMParams(**deployment_litellm_params).model_dump(exclude_none=True)
params = self._litellm_params()
params["_litellm_internal_model_credentials"] = MappingProxyType(snapshot)
return params
@ -2575,19 +2477,12 @@ class TestBedrockFileContentTransformation:
monkeypatch.delenv("AWS_S3_OUTPUT_BUCKET_NAME", raising=False)
url, _ = BedrockFilesConfig().transform_file_content_request(
file_content_request={
"file_id": "s3://out-bucket/litellm-batch-outputs/job/in.jsonl.out"
},
file_content_request={"file_id": "s3://out-bucket/litellm-batch-outputs/job/in.jsonl.out"},
optional_params={},
litellm_params=self._trusted(
s3_bucket_name="in-bucket", s3_output_bucket_name="out-bucket"
),
litellm_params=self._trusted(s3_bucket_name="in-bucket", s3_output_bucket_name="out-bucket"),
)
assert (
url
== "https://s3.us-west-2.amazonaws.com/out-bucket/litellm-batch-outputs/job/in.jsonl.out"
)
assert url == "https://s3.us-west-2.amazonaws.com/out-bucket/litellm-batch-outputs/job/in.jsonl.out"
def test_output_bucket_falls_back_to_env(self, monkeypatch):
"""The output bucket resolves from AWS_S3_OUTPUT_BUCKET_NAME when not in
@ -2598,17 +2493,12 @@ class TestBedrockFileContentTransformation:
monkeypatch.setenv("AWS_S3_OUTPUT_BUCKET_NAME", "env-out-bucket")
url, _ = BedrockFilesConfig().transform_file_content_request(
file_content_request={
"file_id": "s3://env-out-bucket/litellm-batch-outputs/job/in.jsonl.out"
},
file_content_request={"file_id": "s3://env-out-bucket/litellm-batch-outputs/job/in.jsonl.out"},
optional_params={},
litellm_params=self._litellm_params(),
)
assert (
url
== "https://s3.us-west-2.amazonaws.com/env-out-bucket/litellm-batch-outputs/job/in.jsonl.out"
)
assert url == "https://s3.us-west-2.amazonaws.com/env-out-bucket/litellm-batch-outputs/job/in.jsonl.out"
def test_input_bucket_still_validates_when_output_bucket_set(self, monkeypatch):
"""Adding output-bucket support must not break retrieval of input-bucket
@ -2619,19 +2509,12 @@ class TestBedrockFileContentTransformation:
monkeypatch.delenv("AWS_S3_OUTPUT_BUCKET_NAME", raising=False)
url, _ = BedrockFilesConfig().transform_file_content_request(
file_content_request={
"file_id": "s3://in-bucket/litellm-batch-outputs/job/in.jsonl.out"
},
file_content_request={"file_id": "s3://in-bucket/litellm-batch-outputs/job/in.jsonl.out"},
optional_params={},
litellm_params=self._trusted(
s3_bucket_name="in-bucket", s3_output_bucket_name="out-bucket"
),
litellm_params=self._trusted(s3_bucket_name="in-bucket", s3_output_bucket_name="out-bucket"),
)
assert (
url
== "https://s3.us-west-2.amazonaws.com/in-bucket/litellm-batch-outputs/job/in.jsonl.out"
)
assert url == "https://s3.us-west-2.amazonaws.com/in-bucket/litellm-batch-outputs/job/in.jsonl.out"
def test_rejects_bucket_outside_input_and_output(self, monkeypatch):
"""A file id whose bucket is neither the input nor the output bucket is
@ -2644,18 +2527,13 @@ class TestBedrockFileContentTransformation:
with pytest.raises(BedrockError, match="configured storage bucket") as rejection:
BedrockFilesConfig().transform_file_content_request(
file_content_request={
"file_id": "s3://other-bucket/litellm-batch-outputs/job/x.jsonl.out"
},
file_content_request={"file_id": "s3://other-bucket/litellm-batch-outputs/job/x.jsonl.out"},
optional_params={},
litellm_params=self._trusted(
s3_bucket_name="in-bucket", s3_output_bucket_name="out-bucket"
),
litellm_params=self._trusted(s3_bucket_name="in-bucket", s3_output_bucket_name="out-bucket"),
)
assert rejection.value.status_code == 400
def test_sign_request_without_botocore_raises_helpful_error(self, monkeypatch):
"""A missing botocore must surface an actionable 'install boto3' error
rather than a raw import failure."""
@ -2681,9 +2559,7 @@ class TestBedrockFileContentTransformation:
monkeypatch.delenv("AWS_S3_BUCKET_NAME", raising=False)
litellm_params = self._litellm_params()
litellm_params["_litellm_internal_model_credentials"] = MappingProxyType(
{"s3_bucket_name": "my-bucket"}
)
litellm_params["_litellm_internal_model_credentials"] = MappingProxyType({"s3_bucket_name": "my-bucket"})
url, _ = BedrockFilesConfig().transform_file_content_request(
file_content_request={"file_id": self.S3_URI},
@ -2723,9 +2599,10 @@ class TestBedrockFileContentTransformation:
s3_endpoint_url="https://bucket.vpce-abc.s3.us-east-1.vpce.amazonaws.com",
)
assert BedrockFilesConfig()._s3_request_target(
optional_params={}, litellm_params=lp
).endpoint_url == "https://bucket.vpce-abc.s3.us-east-1.vpce.amazonaws.com"
assert (
BedrockFilesConfig()._s3_request_target(optional_params={}, litellm_params=lp).endpoint_url
== "https://bucket.vpce-abc.s3.us-east-1.vpce.amazonaws.com"
)
def test_validate_environment_merges_and_pops_signed_get_headers(self):
from litellm.llms.bedrock.files.transformation import (
@ -2733,9 +2610,7 @@ class TestBedrockFileContentTransformation:
BedrockFilesConfig,
)
litellm_params = {
S3_SIGNED_REQUEST_HEADERS_PARAM: {"Authorization": "AWS4-HMAC-SHA256 test"}
}
litellm_params = {S3_SIGNED_REQUEST_HEADERS_PARAM: {"Authorization": "AWS4-HMAC-SHA256 test"}}
headers = BedrockFilesConfig().validate_environment(
headers={"x-custom": "kept"},
@ -2801,9 +2676,7 @@ class TestBedrockFileContentTransformation:
monkeypatch.setenv("AWS_S3_BUCKET_NAME", "my-bucket")
with respx.mock:
route = respx.get(self.EXPECTED_URL).mock(
return_value=httpx.Response(200, content=b'{"recordId": "x"}')
)
route = respx.get(self.EXPECTED_URL).mock(return_value=httpx.Response(200, content=b'{"recordId": "x"}'))
response = litellm.file_content(
file_id=self.S3_URI,
@ -2831,9 +2704,7 @@ class TestBedrockFileContentTransformation:
litellm.in_memory_llm_clients_cache.flush_cache()
with respx.mock:
route = respx.get(self.EXPECTED_URL).mock(
return_value=httpx.Response(200, content=b'{"recordId": "x"}')
)
route = respx.get(self.EXPECTED_URL).mock(return_value=httpx.Response(200, content=b'{"recordId": "x"}'))
response = await litellm.afile_content(
file_id=self.S3_URI,
@ -2842,11 +2713,7 @@ class TestBedrockFileContentTransformation:
)
assert route.called
assert (
route.calls[0]
.request.headers["Authorization"]
.startswith("AWS4-HMAC-SHA256")
)
assert route.calls[0].request.headers["Authorization"].startswith("AWS4-HMAC-SHA256")
assert response.content == b'{"recordId": "x"}'
@ -2880,9 +2747,7 @@ class TestBedrockFilesS3SignatureEncoding:
headers: Mapping[str, str],
) -> str:
sent = {name.lower(): value for name, value in headers.items()}
signed_names = (
sent["authorization"].split("SignedHeaders=")[1].split(",")[0].split(";")
)
signed_names = sent["authorization"].split("SignedHeaders=")[1].split(",")[0].split(";")
request = AWSRequest(
method=method,
url=url,
@ -2890,12 +2755,8 @@ class TestBedrockFilesS3SignatureEncoding:
headers={name: sent[name] for name in signed_names if name in sent},
)
request.context["timestamp"] = sent["x-amz-date"]
signer = signer_cls(
Credentials(self.ACCESS_KEY, self.SECRET_KEY), "s3", self.REGION
)
return signer.signature(
signer.string_to_sign(request, signer.canonical_request(request)), request
)
signer = signer_cls(Credentials(self.ACCESS_KEY, self.SECRET_KEY), "s3", self.REGION)
return signer.signature(signer.string_to_sign(request, signer.canonical_request(request)), request)
def _assert_signed_the_way_s3_reads_it(
self,
@ -2906,12 +2767,8 @@ class TestBedrockFilesS3SignatureEncoding:
) -> None:
assert "%20" in url, "the object key must reach the wire percent-encoded"
sent_signature = headers["Authorization"].split("Signature=")[1].strip()
assert sent_signature == self._signature_under(
S3SigV4Auth, method, url, body, headers
)
assert sent_signature != self._signature_under(
SigV4Auth, method, url, body, headers
)
assert sent_signature == self._signature_under(S3SigV4Auth, method, url, body, headers)
assert sent_signature != self._signature_under(SigV4Auth, method, url, body, headers)
def test_create_file_signs_spaced_object_key_the_way_s3_does(self) -> None:
from litellm.llms.bedrock.files.transformation import BedrockFilesConfig
@ -2945,9 +2802,7 @@ class TestBedrockFilesS3SignatureEncoding:
headers=signed["headers"],
)
def test_file_content_signs_spaced_object_key_the_way_s3_does(
self, monkeypatch: pytest.MonkeyPatch
) -> None:
def test_file_content_signs_spaced_object_key_the_way_s3_does(self, monkeypatch: pytest.MonkeyPatch) -> None:
from litellm.llms.bedrock.files.transformation import (
S3_SIGNED_REQUEST_HEADERS_PARAM,
BedrockFilesConfig,
@ -2961,9 +2816,7 @@ class TestBedrockFilesS3SignatureEncoding:
}
url, _ = BedrockFilesConfig().transform_file_content_request(
file_content_request={
"file_id": "s3://my-bucket/LLM AI Projects/litellm-bedrock-files-model-abc.jsonl"
},
file_content_request={"file_id": "s3://my-bucket/LLM AI Projects/litellm-bedrock-files-model-abc.jsonl"},
optional_params={},
litellm_params=litellm_params,
)
@ -3294,7 +3147,6 @@ class TestBedrockFileDeletionTransformation:
assert rejection.value.status_code == 400
def test_transform_delete_file_request_rejects_unmanaged_key(self, monkeypatch):
from litellm.llms.bedrock.common_utils import BedrockError
from litellm.llms.bedrock.files.transformation import BedrockFilesConfig
@ -3310,7 +3162,6 @@ class TestBedrockFileDeletionTransformation:
assert rejection.value.status_code == 400
def test_transform_delete_file_response_echoes_the_deleted_id(self):
import httpx

6
tests/unit/llms/bedrock/messages/invoke_transformations/test_anthropic_claude3_transformation.py
View file

@ -14,7 +14,9 @@ from unittest.mock import Mock
import httpx
import pytest
from pydantic import TypeAdapter
import litellm
from litellm.constants import (
BEDROCK_MIN_THINKING_BUDGET_TOKENS,
DEFAULT_REASONING_EFFORT_HIGH_THINKING_BUDGET,
@ -39,6 +41,7 @@ from litellm.llms.bedrock.messages.invoke_transformations.anthropic_claude3_tran
)
from litellm.llms.bedrock.messages.mantle_transformation import AmazonMantleMessagesConfig
from litellm.llms.bedrock_mantle.messages.transformation import BedrockMantleAnthropicMessagesConfig
from litellm.llms.custom_httpx.http_handler import AsyncHTTPHandler
@pytest.mark.asyncio
@ -3943,3 +3946,6 @@ async def test_bedrock_messages_400_is_not_retried_unless_it_rejects_opt_in_nati
assert isinstance(raised, litellm.BadRequestError)
assert len(bedrock.bodies) == 1
REASONING_BODY_ADAPTER: Final = TypeAdapter(dict[str, object])

141
tests/unit/llms/bedrock/test_bedrock_common_utils.py
View file

@ -1,9 +1,15 @@
import asyncio, importlib, litellm, litellm.litellm_core_utils.get_model_cost_map as bedrock_govcloud_model_cost_map, pytest
import asyncio
import importlib
from collections.abc import Iterator
from typing import Final
from unittest.mock import Mock, patch
import pytest
from litellm.llms.bedrock.common_utils import(
import litellm
import litellm.litellm_core_utils.get_model_cost_map as bedrock_govcloud_model_cost_map
from litellm import completion
from litellm.llms.bedrock.common_utils import (
AmazonBedrockGlobalConfig,
BedrockModelInfo,
bedrock_native_output_is_opt_in,
@ -16,11 +22,8 @@ from litellm.llms.bedrock.common_utils import(
strip_bedrock_throughput_suffix,
without_bedrock_native_output_opt_in,
)
from collections.abc import Iterator
from litellm import completion
from litellm.llms.bedrock.count_tokens.bedrock_token_counter import BedrockTokenCounter
from tests._vcr_conftest_common import install_live_call_probe, record_vcr_outcome
from unittest.mock import Mock, patch
# --------------------------------------------------------------------------- #
# get_bedrock_response_stream_shape lazy-load tests #
@ -48,9 +51,7 @@ def test_bedrock_response_stream_shape_lazy_loads_once():
import litellm.llms.bedrock.common_utils as mod
sentinel = MagicMock()
with patch.object(
mod, "_load_bedrock_response_stream_shape", return_value=sentinel
) as mock_load:
with patch.object(mod, "_load_bedrock_response_stream_shape", return_value=sentinel) as mock_load:
assert mod.get_bedrock_response_stream_shape() is sentinel
assert mod.get_bedrock_response_stream_shape() is sentinel
mock_load.assert_called_once()
@ -97,9 +98,7 @@ def test_bedrock_response_stream_shape_is_structure_shape():
from litellm.llms.bedrock.common_utils import get_bedrock_response_stream_shape
loaded_shape = get_bedrock_response_stream_shape()
assert (
loaded_shape is not None
), "get_bedrock_response_stream_shape() is None — botocore may not be installed"
assert loaded_shape is not None, "get_bedrock_response_stream_shape() is None — botocore may not be installed"
shape: StructureShape = loaded_shape
assert isinstance(shape, StructureShape)
assert shape.name == "ResponseStream"
@ -164,9 +163,7 @@ def test_deepseek_cris():
Test that DeepSeek models with cross-region inference prefix use converse route
"""
bedrock_model_info = BedrockModelInfo
bedrock_route = bedrock_model_info.get_bedrock_route(
model="bedrock/us.deepseek.r1-v1:0"
)
bedrock_route = bedrock_model_info.get_bedrock_route(model="bedrock/us.deepseek.r1-v1:0")
assert bedrock_route == "converse"
@ -239,27 +236,19 @@ def test_govcloud_cross_region_inference_prefix():
bedrock_model_info = BedrockModelInfo
# Test us-gov prefix is stripped correctly for Claude models
base_model = bedrock_model_info.get_base_model(
model="bedrock/us-gov.anthropic.claude-haiku-4-5-20251001-v1:0"
)
base_model = bedrock_model_info.get_base_model(model="bedrock/us-gov.anthropic.claude-haiku-4-5-20251001-v1:0")
assert base_model == "anthropic.claude-haiku-4-5-20251001-v1:0"
# Test us-gov prefix is stripped correctly for different Claude versions
base_model = bedrock_model_info.get_base_model(
model="bedrock/us-gov.anthropic.claude-sonnet-4-5-20250929-v1:0"
)
base_model = bedrock_model_info.get_base_model(model="bedrock/us-gov.anthropic.claude-sonnet-4-5-20250929-v1:0")
assert base_model == "anthropic.claude-sonnet-4-5-20250929-v1:0"
# Test us-gov prefix is stripped correctly for Haiku models
base_model = bedrock_model_info.get_base_model(
model="bedrock/us-gov.anthropic.claude-3-haiku-20240307-v1:0"
)
base_model = bedrock_model_info.get_base_model(model="bedrock/us-gov.anthropic.claude-3-haiku-20240307-v1:0")
assert base_model == "anthropic.claude-3-haiku-20240307-v1:0"
# Test us-gov prefix is stripped correctly for Meta models
base_model = bedrock_model_info.get_base_model(
model="bedrock/us-gov.meta.llama3-8b-instruct-v1:0"
)
base_model = bedrock_model_info.get_base_model(model="bedrock/us-gov.meta.llama3-8b-instruct-v1:0")
assert base_model == "meta.llama3-8b-instruct-v1:0"
@ -273,23 +262,14 @@ def test_context_window_suffix_stripped_for_cost_lookup():
"""
from litellm.llms.bedrock.common_utils import get_bedrock_base_model
assert (
get_bedrock_base_model("us.anthropic.claude-opus-4-6-v1[1m]")
== "anthropic.claude-opus-4-6-v1"
)
assert (
get_bedrock_base_model("us.anthropic.claude-sonnet-4-6[1m]")
== "anthropic.claude-sonnet-4-6"
)
assert get_bedrock_base_model("us.anthropic.claude-opus-4-6-v1[1m]") == "anthropic.claude-opus-4-6-v1"
assert get_bedrock_base_model("us.anthropic.claude-sonnet-4-6[1m]") == "anthropic.claude-sonnet-4-6"
assert (
get_bedrock_base_model("global.anthropic.claude-opus-4-5-20251101-v1:0[1m]")
== "anthropic.claude-opus-4-5-20251101-v1:0"
)
# Ensure models without suffix are unaffected
assert (
get_bedrock_base_model("us.anthropic.claude-opus-4-6-v1")
== "anthropic.claude-opus-4-6-v1"
)
assert get_bedrock_base_model("us.anthropic.claude-opus-4-6-v1") == "anthropic.claude-opus-4-6-v1"
# Ensure :51k throughput suffix still works
assert (
get_bedrock_base_model("anthropic.claude-3-5-sonnet-20241022-v2:0:51k")
@ -339,9 +319,7 @@ def test_output_config_effort_normalization_uses_model_info_ceiling(monkeypatch)
("us.anthropic.claude-opus-4-7", "xhigh"),
],
)
def test_bundled_bedrock_opus_model_info_declares_output_config_effort_ceiling(
model, expected_ceiling
):
def test_bundled_bedrock_opus_model_info_declares_output_config_effort_ceiling(model, expected_ceiling):
from litellm.litellm_core_utils.get_model_cost_map import GetModelCostMap
model_info = GetModelCostMap.load_local_model_cost_map()[model]
@ -360,54 +338,24 @@ def test_route_prefix_matched_as_path_segment_not_substring():
or a ``/`` boundary.
"""
# The bedrock_mantle/ provider prefix must NOT be read as the mantle/ route.
assert (
BedrockModelInfo.get_bedrock_route("bedrock_mantle/openai.gpt-5.5") != "mantle"
)
assert (
BedrockModelInfo.get_bedrock_route("bedrock_mantle/openai.gpt-5.4") == "converse"
)
assert (
BedrockModelInfo._explicit_mantle_route("bedrock_mantle/openai.gpt-5.5")
is False
)
assert BedrockModelInfo.get_bedrock_route("bedrock_mantle/openai.gpt-5.5") != "mantle"
assert BedrockModelInfo.get_bedrock_route("bedrock_mantle/openai.gpt-5.4") == "converse"
assert BedrockModelInfo._explicit_mantle_route("bedrock_mantle/openai.gpt-5.5") is False
# A genuine mantle route still resolves, via the startswith branch...
assert (
BedrockModelInfo.get_bedrock_route("mantle/anthropic.claude-mythos-preview")
== "mantle"
)
assert BedrockModelInfo.get_bedrock_route("mantle/anthropic.claude-mythos-preview") == "mantle"
# ...and via the mid-path "/mantle/" branch (after the bedrock/ provider prefix).
assert (
BedrockModelInfo.get_bedrock_route(
"bedrock/mantle/anthropic.claude-mythos-preview"
)
== "mantle"
)
assert BedrockModelInfo.get_bedrock_route("bedrock/mantle/anthropic.claude-mythos-preview") == "mantle"
def test_model_has_route_prefix_exercises_both_branches():
"""``_model_has_route_prefix`` matches on ``startswith`` or a ``/`` boundary only."""
# startswith branch
assert (
BedrockModelInfo._model_has_route_prefix(
"mantle/anthropic.claude-mythos-preview", "mantle/"
)
is True
)
assert BedrockModelInfo._model_has_route_prefix("mantle/anthropic.claude-mythos-preview", "mantle/") is True
# f"/{prefix}" boundary branch
assert (
BedrockModelInfo._model_has_route_prefix(
"bedrock/mantle/anthropic.claude-mythos-preview", "mantle/"
)
is True
)
assert BedrockModelInfo._model_has_route_prefix("bedrock/mantle/anthropic.claude-mythos-preview", "mantle/") is True
# neither branch: the token only appears glued to another segment
assert (
BedrockModelInfo._model_has_route_prefix(
"bedrock_mantle/openai.gpt-5.5", "mantle/"
)
is False
)
assert BedrockModelInfo._model_has_route_prefix("bedrock_mantle/openai.gpt-5.5", "mantle/") is False
@pytest.mark.parametrize(
@ -457,16 +405,10 @@ def test_explicit_invoke_route_does_not_match_async_invoke():
"""
async_invoke_model = "async_invoke/twelvelabs.marengo-embed-2-7-v1:0"
assert BedrockModelInfo._explicit_invoke_route(async_invoke_model) is False
assert (
BedrockModelInfo._explicit_invoke_route(f"bedrock/{async_invoke_model}")
is False
)
assert BedrockModelInfo._explicit_invoke_route(f"bedrock/{async_invoke_model}") is False
# ...while async_invoke/ is still detected as its own route.
assert BedrockModelInfo._explicit_async_invoke_route(async_invoke_model) is True
assert (
BedrockModelInfo._explicit_async_invoke_route(f"bedrock/{async_invoke_model}")
is True
)
assert BedrockModelInfo._explicit_async_invoke_route(f"bedrock/{async_invoke_model}") is True
def test_capability_lookups_fall_back_to_base_model_when_regional_entry_lacks_field(monkeypatch):
@ -1183,6 +1125,7 @@ def _vcr_outcome_gate(request, vcr):
yield
record_vcr_outcome(request, vcr)
@pytest.fixture(scope="session")
def event_loop():
try:
@ -1192,6 +1135,7 @@ def event_loop():
yield loop
loop.close()
@pytest.fixture(scope="function")
def setup_and_teardown(event_loop):
import litellm
@ -1225,6 +1169,7 @@ def setup_and_teardown(event_loop):
if pending:
event_loop.run_until_complete(asyncio.gather(*pending, return_exceptions=True))
_SCALAR_DEFAULTS = {
"num_retries": getattr(litellm, "num_retries", None),
"set_verbose": getattr(litellm, "set_verbose", False),
@ -1239,6 +1184,7 @@ _SCALAR_DEFAULTS = {
"cohere_key": getattr(litellm, "cohere_key", None),
}
@pytest.mark.usefixtures("_vcr_outcome_gate", "setup_and_teardown")
class TestStripBedrockRoutingPrefix:
"""Tests for strip_bedrock_routing_prefix function."""
@ -1269,6 +1215,7 @@ class TestStripBedrockRoutingPrefix:
== "anthropic.claude-3-sonnet-20240229-v1:0"
)
@pytest.mark.usefixtures("_vcr_outcome_gate", "setup_and_teardown")
class TestStripBedrockThroughputSuffix:
"""Tests for strip_bedrock_throughput_suffix function."""
@ -1296,6 +1243,7 @@ class TestStripBedrockThroughputSuffix:
def test_strip_throughput_suffix(self, input_model, expected):
assert strip_bedrock_throughput_suffix(input_model) == expected
@pytest.mark.usefixtures("_vcr_outcome_gate", "setup_and_teardown")
class TestExtractModelNameFromBedrockArn:
"""Tests for extract_model_name_from_bedrock_arn function."""
@ -1316,6 +1264,7 @@ class TestExtractModelNameFromBedrockArn:
arn = "ARN:aws:bedrock:us-east-1:123456789012:model/my-model"
assert extract_model_name_from_bedrock_arn(arn) == "my-model"
@pytest.mark.usefixtures("_vcr_outcome_gate", "setup_and_teardown")
class TestGetBedrockCrossRegionInferenceRegions:
"""Tests for get_bedrock_cross_region_inference_regions function."""
@ -1331,6 +1280,7 @@ class TestGetBedrockCrossRegionInferenceRegions:
regions = get_bedrock_cross_region_inference_regions()
assert isinstance(regions, list)
@pytest.mark.usefixtures("_vcr_outcome_gate", "setup_and_teardown")
class TestGetBedrockBaseModel:
"""Tests for get_bedrock_base_model function."""
@ -1394,6 +1344,7 @@ class TestGetBedrockBaseModel:
"""Test that throughput tier suffixes like :51k are stripped. Issue #19113."""
assert get_bedrock_base_model(input_model) == expected
@pytest.mark.usefixtures("_vcr_outcome_gate", "setup_and_teardown")
class TestBedrockModelInfoWrappers:
"""Tests that BedrockModelInfo methods correctly wrap standalone functions."""
@ -1415,6 +1366,7 @@ class TestBedrockModelInfoWrappers:
model = "bedrock/converse/claude-3"
assert BedrockModelInfo.get_non_litellm_routing_model_name(model) == strip_bedrock_routing_prefix(model)
@pytest.mark.usefixtures("_vcr_outcome_gate", "setup_and_teardown")
class TestBedrockTokenCounter:
"""Tests for BedrockTokenCounter class."""
@ -1451,12 +1403,14 @@ class TestBedrockTokenCounter:
)
assert result is None
@pytest.fixture
def _pr4_bedrock_env(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setenv("AWS_ACCESS_KEY_ID", "pr4-test-aws-access-key")
monkeypatch.setenv("AWS_SECRET_ACCESS_KEY", "pr4-test-aws-secret-key")
monkeypatch.setenv("AWS_REGION_NAME", "us-east-1")
@pytest.fixture(scope="module")
def _use_local_model_cost_map() -> Iterator[None]:
with pytest.MonkeyPatch.context() as monkeypatch:
@ -1465,6 +1419,7 @@ def _use_local_model_cost_map() -> Iterator[None]:
importlib.reload(litellm)
yield
@pytest.mark.usefixtures(
"_pr4_bedrock_env",
"_use_local_model_cost_map",
@ -2005,6 +1960,7 @@ class TestBedrockGovCloudSupport:
@pytest.mark.parametrize("missing", ["botocore", "unrelated_dependency"])
def test_event_decoder_reports_only_missing_aws_dependency(missing):
from unittest.mock import patch
from litellm.llms.bedrock.common_utils import BedrockEventStreamDecoderBase
failure = ModuleNotFoundError("missing dependency", name=missing)
@ -2018,10 +1974,14 @@ def test_event_decoder_reports_only_missing_aws_dependency(missing):
assert error.value is failure
_FORMAT_REJECTION: Final = "The model returned the following errors: output_config.format: Extra inputs are not permitted"
_FORMAT_REJECTION: Final = (
"The model returned the following errors: output_config.format: Extra inputs are not permitted"
)
_COMPILE_REJECTION: Final = "The model returned the following errors: compilation exceeded limits: Schema is too complex."
_COMPILE_REJECTION: Final = (
"The model returned the following errors: compilation exceeded limits: Schema is too complex."
)
_SCHEMA_PATTERN_REJECTION: Final = (
@ -2087,4 +2047,3 @@ def test_without_bedrock_native_output_opt_in_turns_the_opt_in_off_and_keeps_the
assert turned_off["api_base"] == "https://bedrock.example"
assert deployment_model_info(turned_off)["id"] == "deployment-1"
assert litellm_params["model_info"] == {"id": "deployment-1"}

218
tests/unit/test_utils.py
View file

@ -1,13 +1,15 @@
import asyncio, importlib, re
import asyncio
import base64
import copy
import contextlib
import contextvars
import copy
import importlib
import io
import json
import logging
import os
import queue
import re
import threading
from collections.abc import Callable, Iterator, Mapping
from concurrent.futures import Future, ThreadPoolExecutor
@ -38,7 +40,6 @@ from litellm.constants import DEFAULT_MOCK_RESPONSE_COMPLETION_TOKEN_COUNT
from litellm.integrations.custom_guardrail import CustomGuardrail
from litellm.integrations.custom_logger import CustomLogger
from litellm.integrations.humanloop import HumanloopLogger
from litellm.llms.base_llm.audio_transcription.transformation import BaseAudioTranscriptionConfig
from litellm.integrations.langfuse.langfuse_prompt_management import LangfusePromptManagement
from litellm.litellm_core_utils import litellm_logging
from litellm.litellm_core_utils.duration_parser import (
@ -47,6 +48,7 @@ from litellm.litellm_core_utils.duration_parser import (
)
from litellm.litellm_core_utils.get_litellm_params import get_litellm_params
from litellm.litellm_core_utils.thread_pool_executor import executor as logging_executor
from litellm.llms.base_llm.audio_transcription.transformation import BaseAudioTranscriptionConfig
from litellm.llms.base_llm.base_model_iterator import MockResponseIterator
from litellm.llms.custom_httpx.http_handler import AsyncHTTPHandler, headers
from litellm.llms.openai_like.json_loader import JSONProviderRegistry
@ -77,28 +79,28 @@ from litellm.types.utils import (
)
from litellm.types.videos.main import VideoObject
from litellm.utils import (
_invalidate_model_cost_lowercase_map,
add_custom_logger_callback_to_specific_event,
check_valid_key,
CustomStreamWrapper,
filter_out_litellm_params,
get_applied_guardrails,
get_llm_provider,
get_optional_params_embeddings,
ProviderConfigManager,
TextCompletionStreamWrapper,
_check_provider_match,
get_potential_model_names,
_invalidate_model_cost_lowercase_map,
_is_litellm_router_call,
_is_streaming_request,
_run_success_deployment_hook_on_converted_chat_stream,
_snapshot_exception_for_hook,
add_custom_logger_callback_to_specific_event,
async_post_call_failure_deployment_hook,
async_post_call_success_deployment_hook,
calculate_max_parallel_requests,
check_valid_key,
client,
filter_out_litellm_params,
get_applied_guardrails,
get_llm_provider,
get_non_default_completion_params,
get_optional_params_embeddings,
get_optional_params_image_gen,
get_potential_model_names,
get_prompt_cache_min_tokens,
get_supported_openai_params,
get_token_count,
@ -106,8 +108,8 @@ from litellm.utils import (
is_cached_message,
is_prompt_caching_valid_prompt,
trim_messages,
validate_environment,
validate_chat_completion_tool_choice,
validate_environment,
)
@ -1604,10 +1606,10 @@ def test_vertex_params_not_stripped_for_vertex_family(model, custom_llm_provider
assert optional_params["vertex_location"] == "us-central1"
from litellm.utils import supports_function_calling
from litellm.litellm_core_utils.logging_worker import GLOBAL_LOGGING_WORKER
from tests._vcr_conftest_common import install_live_call_probe, record_vcr_outcome
from litellm import acompletion, completion
from litellm.litellm_core_utils.logging_worker import GLOBAL_LOGGING_WORKER
from litellm.utils import supports_function_calling
from tests._vcr_conftest_common import install_live_call_probe, record_vcr_outcome
class TestProxyFunctionCalling:
@ -3247,9 +3249,7 @@ class TestExtraBodyCannotOverrideModel:
}
]
untouched = litellm.get_optional_params(
model="my-vllm-model", custom_llm_provider="hosted_vllm", tools=tools()
)
untouched = litellm.get_optional_params(model="my-vllm-model", custom_llm_provider="hosted_vllm", tools=tools())
assert untouched["tools"][0]["function"]["custom_marker"] == "LEAK", untouched
result = litellm.get_optional_params(
@ -4240,7 +4240,9 @@ def test_is_prompt_caching_valid_prompt_stops_counting_once_the_minimum_is_reach
assert is_prompt_caching_valid_prompt(model="claude-opus-4-8", messages=long_prompt, min_token_count=1024) is True
assert sum(counted_messages) < len(long_prompt), sum(counted_messages)
full_count = litellm.token_counter(model="claude-opus-4-8", messages=long_prompt, use_default_image_token_count=True)
full_count = litellm.token_counter(
model="claude-opus-4-8", messages=long_prompt, use_default_image_token_count=True
)
assert (
is_prompt_caching_valid_prompt(model="claude-opus-4-8", messages=long_prompt, min_token_count=full_count)
is True
@ -4707,7 +4709,13 @@ _SUCCESS_RESPONSES_BY_CALL_TYPE: Final = (
pytest.param(EmbeddingResponse(model="text-embedding-3-small"), CallTypes.aembedding, id="embedding"),
pytest.param(
ResponsesAPIResponse(
id="resp_abc", created_at=1, output=[], parallel_tool_calls=False, tool_choice="auto", tools=[], model="gpt-5.6"
id="resp_abc",
created_at=1,
output=[],
parallel_tool_calls=False,
tool_choice="auto",
tools=[],
model="gpt-5.6",
),
CallTypes.aresponses,
id="responses",
@ -4735,7 +4743,9 @@ async def test_success_deployment_hook_raising_keeps_response_and_runs_later_hoo
assert result is response
assert second_hook.seen_responses == (response,)
failure_logs: Final = tuple(r for r in caplog.records if "async_post_call_success_deployment_hook error" in r.message)
failure_logs: Final = tuple(
r for r in caplog.records if "async_post_call_success_deployment_hook error" in r.message
)
assert len(failure_logs) == 1
assert "_ChatShapedSuccessDeploymentHook" in failure_logs[0].message
assert str(call_type) in failure_logs[0].message
@ -4872,17 +4882,13 @@ async def test_wrapper_async_logs_converted_responses_stream_with_standard_loggi
({"metadata": None}, True, False),
],
)
def test_is_litellm_router_call_is_async_aware(
kwargs: Mapping[str, object], is_async: bool, expected: bool
) -> None:
def test_is_litellm_router_call_is_async_aware(kwargs: Mapping[str, object], is_async: bool, expected: bool) -> None:
assert _is_litellm_router_call(kwargs, is_async=is_async) is expected
@pytest.mark.asyncio
@pytest.mark.parametrize("stream", [False, True], ids=["non_streaming", "streaming"])
async def test_router_aresponses_does_not_run_sdk_retries(
monkeypatch: pytest.MonkeyPatch, stream: bool
) -> None:
async def test_router_aresponses_does_not_run_sdk_retries(monkeypatch: pytest.MonkeyPatch, stream: bool) -> None:
monkeypatch.setattr(litellm, "disable_aiohttp_transport", True)
litellm.in_memory_llm_clients_cache.flush_cache()
model_list: Final = [
@ -4896,9 +4902,7 @@ async def test_router_aresponses_does_not_run_sdk_retries(
},
}
]
router: Final = litellm.Router(
model_list=model_list, num_retries=0, retry_after=0, disable_cooldowns=True
)
router: Final = litellm.Router(model_list=model_list, num_retries=0, retry_after=0, disable_cooldowns=True)
try:
with respx.mock(assert_all_called=True) as respx_mock:
@ -4934,9 +4938,7 @@ def test_router_responses_keeps_sdk_retries_for_sync_router_call(
},
}
]
router: Final = litellm.Router(
model_list=model_list, num_retries=0, retry_after=0, disable_cooldowns=True
)
router: Final = litellm.Router(model_list=model_list, num_retries=0, retry_after=0, disable_cooldowns=True)
try:
with respx.mock(assert_all_called=True) as respx_mock:
@ -6780,7 +6782,9 @@ def test_function_setup_logs_the_search_query_edit_prompt_and_ocr_document_summa
@pytest.mark.parametrize("original_function", ("atext_completion", "text_completion"))
def test_function_setup_without_a_prompt_leaves_the_missing_prompt_to_request_validation(original_function: str) -> None:
def test_function_setup_without_a_prompt_leaves_the_missing_prompt_to_request_validation(
original_function: str,
) -> None:
assert _logged_request_messages(original_function, model="gpt-4o") is None
@ -6822,6 +6826,7 @@ def _vcr_outcome_gate(request, vcr):
yield
record_vcr_outcome(request, vcr)
@pytest.fixture(scope="function")
def setup_and_teardown():
"""
@ -6835,6 +6840,7 @@ def setup_and_teardown():
loop.close()
asyncio.set_event_loop(None)
MODEL: Final = "anthropic/claude-haiku-4-5"
@ -6844,6 +6850,7 @@ def test_validate_tool_choice_none():
result = validate_chat_completion_tool_choice(None, model=MODEL)
assert result is None
@pytest.mark.usefixtures("_vcr_outcome_gate", "setup_and_teardown")
def test_validate_tool_choice_string():
"""Test that string values are returned as-is."""
@ -6851,6 +6858,7 @@ def test_validate_tool_choice_string():
assert validate_chat_completion_tool_choice("none", model=MODEL) == "none"
assert validate_chat_completion_tool_choice("required", model=MODEL) == "required"
@pytest.mark.usefixtures("_vcr_outcome_gate", "setup_and_teardown")
def test_validate_tool_choice_standard_dict():
"""Test standard OpenAI format with function."""
@ -6858,6 +6866,7 @@ def test_validate_tool_choice_standard_dict():
result = validate_chat_completion_tool_choice(tool_choice, model=MODEL)
assert result == tool_choice
@pytest.mark.usefixtures("_vcr_outcome_gate", "setup_and_teardown")
def test_validate_tool_choice_cursor_format():
"""Cursor IDE format {"type": "auto"} is unwrapped to the bare string."""
@ -6865,6 +6874,7 @@ def test_validate_tool_choice_cursor_format():
assert validate_chat_completion_tool_choice({"type": "none"}, model=MODEL) == "none"
assert validate_chat_completion_tool_choice({"type": "required"}, model=MODEL) == "required"
@pytest.mark.usefixtures("_vcr_outcome_gate", "setup_and_teardown")
@pytest.mark.parametrize(
"tool_choice",
@ -6886,6 +6896,7 @@ def test_validate_tool_choice_invalid_dict_is_a_400(tool_choice):
assert exc_info.value.status_code == 400
assert exc_info.value.model == MODEL
@pytest.mark.usefixtures("_vcr_outcome_gate", "setup_and_teardown")
@pytest.mark.parametrize("tool_choice", [123, []])
def test_validate_tool_choice_invalid_type_is_a_400(tool_choice):
@ -6896,6 +6907,7 @@ def test_validate_tool_choice_invalid_type_is_a_400(tool_choice):
validate_chat_completion_tool_choice(tool_choice, model=MODEL)
assert exc_info.value.status_code == 400
@pytest.mark.usefixtures("_vcr_outcome_gate", "setup_and_teardown")
def test_validate_tool_choice_without_model_is_still_a_400():
"""Callers that predate the model argument keep getting a 400, with an empty model on the error."""
@ -6904,12 +6916,14 @@ def test_validate_tool_choice_without_model_is_still_a_400():
assert exc_info.value.status_code == 400
assert exc_info.value.model == ""
@pytest.fixture()
def _vcr_outcome_gate_local_testing(request, vcr):
install_live_call_probe(request, vcr)
yield
record_vcr_outcome(request, vcr)
@pytest.fixture(scope="function")
def isolate_litellm_state():
"""
@ -6962,6 +6976,7 @@ def isolate_litellm_state():
setattr(litellm, attr, original_value)
_invalidate_model_cost_lowercase_map()
_SCALAR_DEFAULTS = {
"num_retries": getattr(litellm, "num_retries", None),
"num_retries_per_request": getattr(litellm, "num_retries_per_request", None),
@ -7005,6 +7020,7 @@ def setup_and_teardown_local_testing():
litellm.in_memory_llm_clients_cache.flush_cache()
yield
@pytest.mark.usefixtures(
"_vcr_outcome_gate_local_testing",
"isolate_litellm_state",
@ -7030,6 +7046,7 @@ def test_vertex_projects():
assert "vertex_ai_project" in optional_params
assert "vertex_ai_location" in optional_params
@pytest.mark.usefixtures(
"_vcr_outcome_gate_local_testing",
"isolate_litellm_state",
@ -7045,6 +7062,7 @@ def test_bedrock_embed_v2_regular():
print(f"received optional_params: {optional_params}")
assert optional_params == {"dimensions": 512}
@pytest.mark.usefixtures(
"_vcr_outcome_gate_local_testing",
"isolate_litellm_state",
@ -7063,6 +7081,7 @@ def test_bedrock_embed_v2_with_drop_params():
print(f"received optional_params: {optional_params}")
assert optional_params == {"dimensions": 512, "embeddingTypes": ["binary"]}
@pytest.mark.usefixtures(
"_vcr_outcome_gate_local_testing",
"isolate_litellm_state",
@ -7084,6 +7103,7 @@ def test_openai_non_text_embedding_3_with_allowed_openai_params():
print(f"received optional_params: {optional_params}")
assert optional_params.get("dimensions") == 1024
@pytest.mark.usefixtures(
"_vcr_outcome_gate_local_testing",
"isolate_litellm_state",
@ -7110,6 +7130,7 @@ def test_openai_non_text_embedding_3_without_allowed_openai_params_raises():
finally:
litellm.drop_params = prev_drop_params
@pytest.mark.usefixtures(
"_vcr_outcome_gate_local_testing",
"isolate_litellm_state",
@ -7138,6 +7159,7 @@ def test_openai_non_text_embedding_3_drop_params_per_call():
finally:
litellm.drop_params = prev_drop_params
@pytest.mark.usefixtures(
"_vcr_outcome_gate_local_testing",
"isolate_litellm_state",
@ -7165,12 +7187,14 @@ def test_openai_non_text_embedding_3_drop_params_global():
finally:
litellm.drop_params = prev_drop_params
@pytest.fixture()
def _vcr_outcome_gate_search_tests(request, vcr):
install_live_call_probe(request, vcr)
yield
record_vcr_outcome(request, vcr)
@pytest.mark.usefixtures("_vcr_outcome_gate_search_tests")
def test_search_tool_name_in_all_litellm_params():
"""
@ -7180,6 +7204,7 @@ def test_search_tool_name_in_all_litellm_params():
"""
assert "search_tool_name" in all_litellm_params
@pytest.mark.usefixtures("_vcr_outcome_gate_search_tests")
def test_filter_out_search_tool_name():
"""
@ -7206,6 +7231,7 @@ def test_filter_out_search_tool_name():
assert filtered["query"] == "latest ai developments"
assert filtered["max_results"] == 5
@pytest.mark.asyncio
async def test_nested_wrapper_exits_schedule_one_async_success_log(monkeypatch: pytest.MonkeyPatch) -> None:
"""Chat over the Responses bridge exits two @client wrappers with one logging object. Issue
@ -7358,9 +7384,7 @@ def test_trimming_with_system_message_within_max_tokens():
"content": "This is a medium normal message, let's say litellm is awesome.",
},
]
trimmed_messages = trim_messages(
messages, max_tokens=30, model="gpt-4-0613"
)
trimmed_messages = trim_messages(messages, max_tokens=30, model="gpt-4-0613")
assert len(trimmed_messages) == 2
assert trimmed_messages[0]["content"] == "This is a short system message"
@ -7543,9 +7567,7 @@ def test_aget_valid_models():
print(valid_models)
# list of openai supported llms on litellm
expected_models = (
litellm.open_ai_chat_completion_models | litellm.open_ai_text_completion_models
)
expected_models = litellm.open_ai_chat_completion_models | litellm.open_ai_text_completion_models
assert set(valid_models) == set(expected_models)
@ -7634,9 +7656,7 @@ def test_get_chat_completion_prompt():
prompt_variables=None,
)
assert litellm_logging_obj.messages == [
{"role": "user", "content": updated_message}
]
assert litellm_logging_obj.messages == [{"role": "user", "content": updated_message}]
def test_redact_msgs_from_logs():
@ -7708,9 +7728,7 @@ def test_redact_embedding_response():
litellm.turn_off_message_logging = True
# Create a test EmbeddingResponse with usage data
original_usage = litellm.Usage(
prompt_tokens=10, completion_tokens=0, total_tokens=10
)
original_usage = litellm.Usage(prompt_tokens=10, completion_tokens=0, total_tokens=10)
original_data = [
{"object": "embedding", "index": 0, "embedding": [0.1, 0.2, 0.3, 0.4, 0.5]},
{"object": "embedding", "index": 1, "embedding": [0.6, 0.7, 0.8, 0.9, 1.0]},
@ -7746,9 +7764,7 @@ def test_redact_embedding_response():
# Assert the redacted response preserves critical metadata
assert _redacted_response_obj.usage == original_usage # usage should be preserved
assert (
_redacted_response_obj.model == "text-embedding-3-small"
) # model should be preserved
assert _redacted_response_obj.model == "text-embedding-3-small" # model should be preserved
assert _redacted_response_obj.object == "list" # object should be preserved
# Assert sensitive data is cleared
@ -7802,12 +7818,8 @@ def test_redact_msgs_from_logs_with_dynamic_params():
)
# Test Case 1: standard_callback_dynamic_params = False (or not set)
standard_callback_dynamic_params = StandardCallbackDynamicParams(
turn_off_message_logging=False
)
litellm_logging_obj.model_call_details["standard_callback_dynamic_params"] = (
standard_callback_dynamic_params
)
standard_callback_dynamic_params = StandardCallbackDynamicParams(turn_off_message_logging=False)
litellm_logging_obj.model_call_details["standard_callback_dynamic_params"] = standard_callback_dynamic_params
_redacted_response_obj = redact_message_input_output_from_logging(
result=response_obj,
model_call_details=litellm_logging_obj.model_call_details,
@ -7816,12 +7828,8 @@ def test_redact_msgs_from_logs_with_dynamic_params():
assert _redacted_response_obj.choices[0].message.content == test_content
# Test Case 2: standard_callback_dynamic_params = True
standard_callback_dynamic_params = StandardCallbackDynamicParams(
turn_off_message_logging=True
)
litellm_logging_obj.model_call_details["standard_callback_dynamic_params"] = (
standard_callback_dynamic_params
)
standard_callback_dynamic_params = StandardCallbackDynamicParams(turn_off_message_logging=True)
litellm_logging_obj.model_call_details["standard_callback_dynamic_params"] = standard_callback_dynamic_params
_redacted_response_obj = redact_message_input_output_from_logging(
result=response_obj,
model_call_details=litellm_logging_obj.model_call_details,
@ -7832,9 +7840,7 @@ def test_redact_msgs_from_logs_with_dynamic_params():
# Test Case 3: standard_callback_dynamic_params does not set turn_off_message_logging
# since litellm.turn_off_message_logging is True redaction should occur
standard_callback_dynamic_params = StandardCallbackDynamicParams()
litellm_logging_obj.model_call_details["standard_callback_dynamic_params"] = (
standard_callback_dynamic_params
)
litellm_logging_obj.model_call_details["standard_callback_dynamic_params"] = standard_callback_dynamic_params
_redacted_response_obj = redact_message_input_output_from_logging(
result=response_obj,
model_call_details=litellm_logging_obj.model_call_details,
@ -8022,8 +8028,10 @@ def test_async_http_handler_force_ipv4(mock_async_client):
This is prod test - we need to ensure that httpx always uses ipv4 when litellm.force_ipv4 is True
"""
import httpx
import ssl
import httpx
from litellm.llms.custom_httpx.http_handler import AsyncHTTPHandler
# Set force_ipv4 to True
@ -8176,10 +8184,7 @@ def test_models_by_provider():
continue
elif k == "sample_spec":
continue
elif (
v["litellm_provider"] == "sagemaker"
or v["litellm_provider"] == "bedrock_converse"
):
elif v["litellm_provider"] == "sagemaker" or v["litellm_provider"] == "bedrock_converse":
continue
elif v.get("mode") in ("search", "decisions", "evaluation"):
continue
@ -8187,9 +8192,7 @@ def test_models_by_provider():
providers.add(v["litellm_provider"])
for provider in providers:
assert provider in models_by_provider.keys() or JSONProviderRegistry.exists(
provider
)
assert provider in models_by_provider.keys() or JSONProviderRegistry.exists(provider)
@pytest.fixture
@ -8211,16 +8214,11 @@ def restore_end_user_cost_tracking_flags(monkeypatch: pytest.MonkeyPatch) -> Non
({"user_api_key_end_user_id": "123"}, True, None),
],
)
def test_get_end_user_id_for_cost_tracking(
litellm_params, disable_end_user_cost_tracking, expected_end_user_id
):
def test_get_end_user_id_for_cost_tracking(litellm_params, disable_end_user_cost_tracking, expected_end_user_id):
from litellm.utils import get_end_user_id_for_cost_tracking
litellm.disable_end_user_cost_tracking = disable_end_user_cost_tracking
assert (
get_end_user_id_for_cost_tracking(litellm_params=litellm_params)
== expected_end_user_id
)
assert get_end_user_id_for_cost_tracking(litellm_params=litellm_params) == expected_end_user_id
@pytest.mark.usefixtures("restore_end_user_cost_tracking_flags")
@ -8237,13 +8235,9 @@ def test_get_end_user_id_for_cost_tracking_prometheus_only(
):
from litellm.utils import get_end_user_id_for_cost_tracking
litellm.enable_end_user_cost_tracking_prometheus_only = (
enable_end_user_cost_tracking_prometheus_only
)
litellm.enable_end_user_cost_tracking_prometheus_only = enable_end_user_cost_tracking_prometheus_only
assert (
get_end_user_id_for_cost_tracking(
litellm_params=litellm_params, service_type="prometheus"
)
get_end_user_id_for_cost_tracking(litellm_params=litellm_params, service_type="prometheus")
== expected_end_user_id
)
@ -8258,20 +8252,14 @@ def test_get_end_user_id_for_cost_tracking_prometheus_only(
),
# Test with only litellm_metadata field (new behavior)
(
{
"litellm_metadata": {
"user_api_key_end_user_id": "user_from_litellm_metadata"
}
},
{"litellm_metadata": {"user_api_key_end_user_id": "user_from_litellm_metadata"}},
"user_from_litellm_metadata",
),
# Test with both fields - metadata should take precedence for user_api_key fields
(
{
"metadata": {"user_api_key_end_user_id": "user_from_metadata"},
"litellm_metadata": {
"user_api_key_end_user_id": "user_from_litellm_metadata"
},
"litellm_metadata": {"user_api_key_end_user_id": "user_from_litellm_metadata"},
},
"user_from_metadata",
),
@ -8287,9 +8275,7 @@ def test_get_end_user_id_for_cost_tracking_prometheus_only(
(
{
"metadata": {},
"litellm_metadata": {
"user_api_key_end_user_id": "user_from_litellm_metadata"
},
"litellm_metadata": {"user_api_key_end_user_id": "user_from_litellm_metadata"},
},
"user_from_litellm_metadata",
),
@ -8297,9 +8283,7 @@ def test_get_end_user_id_for_cost_tracking_prometheus_only(
({}, None),
],
)
def test_get_end_user_id_for_cost_tracking_metadata_handling(
litellm_params, expected_end_user_id
):
def test_get_end_user_id_for_cost_tracking_metadata_handling(litellm_params, expected_end_user_id):
"""
Test that get_end_user_id_for_cost_tracking correctly handles both metadata and litellm_metadata
fields using the get_litellm_metadata_from_kwargs helper function.
@ -8339,10 +8323,11 @@ def test_token_counter_with_image_url_with_detail_high():
PROD TEST this is importat - Can impact latency very badly
"""
from litellm.constants import DEFAULT_IMAGE_TOKEN_COUNT
from litellm._logging import verbose_logger
import logging
from litellm._logging import verbose_logger
from litellm.constants import DEFAULT_IMAGE_TOKEN_COUNT
verbose_logger.setLevel(logging.DEBUG)
_tokens = litellm.utils.token_counter(
@ -8384,8 +8369,8 @@ def test_logprobs_type():
def test_get_valid_models_openai_proxy(monkeypatch):
from litellm.utils import get_valid_models
import litellm
from litellm.utils import get_valid_models
litellm.turn_on_debug()
@ -8557,17 +8542,15 @@ def test_add_custom_logger_callback_to_specific_event_e2e_failure(monkeypatch):
assert len(litellm.success_callback) == curr_len_success_callback
assert len(litellm.failure_callback) == curr_len_failure_callback
assert any(
isinstance(callback, OpenMeterLogger) for callback in litellm.failure_callback
)
assert any(isinstance(callback, OpenMeterLogger) for callback in litellm.failure_callback)
@pytest.mark.asyncio
async def test_wrapper_kwargs_passthrough():
from litellm.utils import client
from litellm.litellm_core_utils.litellm_logging import (
Logging as LiteLLMLoggingObject,
)
from litellm.utils import client
mock_original = AsyncMock()
@ -8656,20 +8639,14 @@ def test_validate_user_messages_invalid_content_type():
[
{
"name": "default_on_guardrail",
"callbacks": [
CustomGuardrail(guardrail_name="test_guardrail", default_on=True)
],
"callbacks": [CustomGuardrail(guardrail_name="test_guardrail", default_on=True)],
"kwargs": {"metadata": {"requester_metadata": {"guardrails": []}}},
"expected": ["test_guardrail"],
},
{
"name": "request_specific_guardrail",
"callbacks": [
CustomGuardrail(guardrail_name="test_guardrail", default_on=False)
],
"kwargs": {
"metadata": {"requester_metadata": {"guardrails": ["test_guardrail"]}}
},
"callbacks": [CustomGuardrail(guardrail_name="test_guardrail", default_on=False)],
"kwargs": {"metadata": {"requester_metadata": {"guardrails": ["test_guardrail"]}}},
"expected": ["test_guardrail"],
},
{
@ -8678,18 +8655,12 @@ def test_validate_user_messages_invalid_content_type():
CustomGuardrail(guardrail_name="default_guardrail", default_on=True),
CustomGuardrail(guardrail_name="request_guardrail", default_on=False),
],
"kwargs": {
"metadata": {
"requester_metadata": {"guardrails": ["request_guardrail"]}
}
},
"kwargs": {"metadata": {"requester_metadata": {"guardrails": ["request_guardrail"]}}},
"expected": ["default_guardrail", "request_guardrail"],
},
{
"name": "empty_metadata",
"callbacks": [
CustomGuardrail(guardrail_name="test_guardrail", default_on=False)
],
"callbacks": [CustomGuardrail(guardrail_name="test_guardrail", default_on=False)],
"kwargs": {},
"expected": [],
},
@ -9247,6 +9218,7 @@ def test_delta_tool_calls_sequential_indices():
defaulting all tool calls to index=0.
"""
import json
from litellm.types.utils import Delta
tool_calls_without_indices = [

70
tests/unit/test_utils_get_model_info.py
View file

@ -1,8 +1,10 @@
import json
import os
import re
from collections.abc import Collection, Iterator, Mapping
from pathlib import Path
from typing import Final, Literal, cast
from unittest.mock import MagicMock, patch
import httpx
import pytest
@ -13,8 +15,6 @@ from litellm.llms.bedrock.common_utils import BedrockModelInfo
from litellm.llms.custom_httpx.http_handler import HTTPHandler
from litellm.types.utils import ModelInfoBase
from litellm.utils import _invalidate_model_cost_lowercase_map
import os
from unittest.mock import MagicMock, patch
@pytest.fixture(autouse=True)
@ -169,9 +169,7 @@ def test_model_info_bedrock_converse(monkeypatch):
except FileNotFoundError:
pytest.skip("whitelisted_bedrock_models.txt not found")
_enforce_bedrock_converse_models(
model_cost=litellm.model_cost, whitelist_models=whitelist_models
)
_enforce_bedrock_converse_models(model_cost=litellm.model_cost, whitelist_models=whitelist_models)
@pytest.mark.flaky(retries=6, delay=2)
@ -195,9 +193,7 @@ def test_model_info_bedrock_converse_enforcement(monkeypatch):
# Check for unwhitelisted models
with pytest.raises(AssertionError, match=r"fake\.bedrock-chat-model"):
_enforce_bedrock_converse_models(
model_cost=litellm.model_cost, whitelist_models=whitelist_models
)
_enforce_bedrock_converse_models(model_cost=litellm.model_cost, whitelist_models=whitelist_models)
except FileNotFoundError as e:
pytest.skip("whitelisted_bedrock_models.txt not found")
@ -256,11 +252,7 @@ def test_get_model_info_bedrock_models():
k = k.replace(f"{commitment}/", "")
base_model = BedrockModelInfo.get_base_model(k)
# get_base_model() returns model id without "bedrock/" prefix; cost map keys use "bedrock/<model>"
base_model_key = (
base_model
if base_model in litellm.model_cost
else f"bedrock/{base_model}"
)
base_model_key = base_model if base_model in litellm.model_cost else f"bedrock/{base_model}"
if base_model_key not in litellm.model_cost:
continue
base_model_info = litellm.model_cost[base_model_key]
@ -268,15 +260,13 @@ def test_get_model_info_bedrock_models():
if "invoke/" in k:
continue
if base_model_key.startswith("supports_"):
assert (
base_model_key in v
), f"{base_model_key} is not in model cost map for {k}"
assert base_model_key in v, f"{base_model_key} is not in model cost map for {k}"
if _govcloud_may_lag(k, base_model_key):
assert base_model_value or not v[base_model_key], f"{base_model_key} is on {k} but not its base"
continue
assert (
v[base_model_key] == base_model_value
), f"{base_model_key} is not equal to {base_model_value} for model {k}"
assert v[base_model_key] == base_model_value, (
f"{base_model_key} is not equal to {base_model_value} for model {k}"
)
def _cross_region_base_model_key(
@ -299,10 +289,7 @@ def _cross_region_profiles(
(model_key, model_info, base_model_key)
for model_key, model_info in model_cost.items()
if str(model_info.get("litellm_provider", "")).startswith("bedrock")
and (
base_model_key := _cross_region_base_model_key(model_key, prefixes, model_cost)
)
is not None
and (base_model_key := _cross_region_base_model_key(model_key, prefixes, model_cost)) is not None
)
@ -336,9 +323,7 @@ def test_get_model_info_bedrock_cross_region_capability_parity():
if _govcloud_may_lag(k, cap):
assert base_value or not v[cap], f"{cap} is on {k} but not on {base_model_key}"
continue
assert (
v[cap] == base_value
), f"{cap} is {v[cap]} on {k} but {base_value} on {base_model_key}"
assert v[cap] == base_value, f"{cap} is {v[cap]} on {k} but {base_value} on {base_model_key}"
assert checked > 0, "no cross-region bedrock profiles found - the filter is inert"
@ -367,9 +352,7 @@ def test_get_model_info_bedrock_priced_cross_region_profile_has_priced_base():
base = litellm.model_cost[base_model_key]
for cost_key in ("input_cost_per_token", "output_cost_per_token"):
if (v.get(cost_key) or 0) > 0:
assert (
base.get(cost_key) or 0
) > 0, f"{k} charges {cost_key} but its base {base_model_key} is free"
assert (base.get(cost_key) or 0) > 0, f"{k} charges {cost_key} but its base {base_model_key} is free"
assert checked > 0, "no cross-region bedrock profiles found - the filter is inert"
@ -401,23 +384,17 @@ def test_get_model_info_case_insensitive_lookup(monkeypatch):
)
# Test 1: Exact case should work
info = litellm.get_model_info(
model="Qwen/Qwen3-Next-80B-A3B-Thinking", custom_llm_provider="together_ai"
)
info = litellm.get_model_info(model="Qwen/Qwen3-Next-80B-A3B-Thinking", custom_llm_provider="together_ai")
assert info is not None
assert info["supports_function_calling"] is True
# Test 2: Lowercase should also work (case-insensitive lookup)
info_lower = litellm.get_model_info(
model="qwen/qwen3-next-80b-a3b-thinking", custom_llm_provider="together_ai"
)
info_lower = litellm.get_model_info(model="qwen/qwen3-next-80b-a3b-thinking", custom_llm_provider="together_ai")
assert info_lower is not None
assert info_lower["supports_function_calling"] is True
# Test 3: Mixed case should also work
info_mixed = litellm.get_model_info(
model="QWEN/qwen3-NEXT-80b-a3b-thinking", custom_llm_provider="together_ai"
)
info_mixed = litellm.get_model_info(model="QWEN/qwen3-NEXT-80b-a3b-thinking", custom_llm_provider="together_ai")
assert info_mixed is not None
assert info_mixed["supports_function_calling"] is True
@ -445,21 +422,14 @@ def test_get_model_info_case_insensitive_supports_function_calling(monkeypatch):
from litellm.utils import supports_function_calling
# Exact case
assert (
supports_function_calling("TestModel-ABC", custom_llm_provider="test_provider")
is True
)
assert supports_function_calling("TestModel-ABC", custom_llm_provider="test_provider") is True
# Lowercase (should now work with case-insensitive lookup)
assert (
supports_function_calling("testmodel-abc", custom_llm_provider="test_provider")
is True
)
assert supports_function_calling("testmodel-abc", custom_llm_provider="test_provider") is True
def test_get_model_info_custom_model_router():
from litellm import Router
from litellm import get_model_info
from litellm import Router, get_model_info
litellm.turn_on_debug()
@ -517,6 +487,4 @@ def test_get_model_info_custom_provider():
# Get registered model info
from litellm import get_model_info
get_model_info(
model="my-custom-llm/my-fake-model"
) # 💥 "Exception: This model isn't mapped yet." in v1.56.10
get_model_info(model="my-custom-llm/my-fake-model") # 💥 "Exception: This model isn't mapped yet." in v1.56.10

8
tests/unit/types/test_completion.py
View file

@ -1,7 +1,7 @@
"""
Tests for litellm.types.completion module
This test suite validates the CompletionRequest model and its compatibility with
This test suite validates the CompletionRequest model and its compatibility with
OpenAI ChatCompletion API message formats.
Usage:
@ -15,8 +15,8 @@ import pytest
from litellm.types.completion import (
ChatCompletionMessageParam,
CompletionRequest,
CompletionDispatchContext,
CompletionRequest,
)
@ -102,9 +102,7 @@ def test_completion_request_multimodal_content():
{"type": "text", "text": "What's in this image?"},
{
"type": "image_url",
"image_url": {
"url": "data:image/jpeg;base64,/9j/4AAQSkZJRgABAQAAAQABAAD..."
},
"image_url": {"url": "data:image/jpeg;base64,/9j/4AAQSkZJRgABAQAAAQABAAD..."},
},
],
}