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

chore: merge main into litellm_migrate_tests_p1

Browse source
This commit is contained in:
yuneng 2026-09-20 12:54:53 +00:00
commit b2ae0151e7
348 changed files with 624 additions and 1851 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
litellm/a2a_protocol/providers/bedrock_agentcore/transformation.py
View file

@ -85,7 +85,7 @@ def _filter_reserved_headers(
def _request_scoped_runtime_session_id(
params: Mapping[str, Any],
params: Mapping[str, object],
litellm_params: Mapping[str, Any],
) -> str | None:
context_id: Final = get_session_id_from_a2a_params(params)

4
litellm/a2a_protocol/providers/watsonx_orchestrate/config.py
View file

@ -20,7 +20,7 @@ class WatsonxOrchestrateA2AConfig(BaseA2AProviderConfig):
params: dict[str, Any],
api_base: str | None = None,
**kwargs: Any,
) -> dict[str, Any]:
) -> dict[str, object]:
"""Handle a non-streaming A2A request via WXO runs API."""
litellm_params: Final = kwargs.get("litellm_params")
if not litellm_params:
@ -40,7 +40,7 @@ class WatsonxOrchestrateA2AConfig(BaseA2AProviderConfig):
params: dict[str, Any],
api_base: str | None = None,
**kwargs: Any,
) -> AsyncIterator[dict[str, Any]]:
) -> AsyncIterator[dict[str, object]]:
"""Handle a streaming A2A request via WXO streaming runs API."""
litellm_params: Final = kwargs.get("litellm_params")
if not litellm_params:

6
litellm/a2a_protocol/utils.py
View file

@ -17,7 +17,7 @@ class A2ARequestUtils:
"""Utility class for A2A request/response processing."""
@staticmethod
def extract_text_from_message(message: Any) -> str:
def extract_text_from_message(message: object) -> str:
"""
Extract text content from A2A message parts.
@ -142,7 +142,7 @@ class A2ARequestUtils:
return prompt_tokens, completion_tokens, total_tokens
def get_session_id_from_a2a_params(params: Mapping[str, Any]) -> str | None:
def get_session_id_from_a2a_params(params: Mapping[str, object]) -> str | None:
message: Final = params.get("message", {})
if isinstance(message, dict):
return message.get("contextId")
@ -166,7 +166,7 @@ def scope_session_to_principal(session_id: str, principal: str | None) -> str:
# Backwards compatibility aliases
def extract_text_from_a2a_message(message: Any) -> str:
def extract_text_from_a2a_message(message: object) -> str:
return A2ARequestUtils.extract_text_from_message(message)

4
litellm/integrations/gitlab/gitlab_prompt_manager.py
View file

@ -200,8 +200,8 @@ class GitLabTemplateManager:
metadata=metadata,
)
def _parse_yaml_basic(self, yaml_str: str) -> dict[str, Any]:
result: Final[dict[str, Any]] = {}
def _parse_yaml_basic(self, yaml_str: str) -> dict[str, bool | int | float | str]:
result: Final[dict[str, bool | int | float | str]] = {}
for line in yaml_str.split("\n"):
line = line.strip()
if ":" in line and not line.startswith("#"):

4
litellm/integrations/vantage/vantage_logger.py
View file

@ -59,7 +59,7 @@ class VantageLogger(FocusLogger):
raw_interval,
)
destination_config: Final[dict[str, Any]] = {}
destination_config: Final[dict[str, str]] = {}
if resolved_api_key:
destination_config["api_key"] = resolved_api_key
if resolved_token:
@ -93,7 +93,7 @@ class VantageLogger(FocusLogger):
pod_lock_manager = None
if proxy_logging_obj is not None:
writer: Final = getattr(proxy_logging_obj, "db_spend_update_writer", None)
writer: Final[object] = getattr(proxy_logging_obj, "db_spend_update_writer", None)
if writer is not None:
pod_lock_manager = getattr(writer, "pod_lock_manager", None)

22
litellm/interactions/agents/http_handler.py
View file

@ -7,7 +7,7 @@ duplicated. BaseAgentsAPIConfig stays as pure transform code.
"""
from collections.abc import Coroutine, Mapping
from typing import Any, Final
from typing import Final
import httpx
@ -38,7 +38,7 @@ class AgentsHTTPHandler(InteractionsHTTPHandler):
name: str,
litellm_params: GenericLiteLLMParams,
logging_obj: LiteLLMLoggingObj,
extra_headers: dict[str, Any] | None = None,
extra_headers: dict[str, str] | None = None,
extra_body: Mapping[str, object] | None = None,
timeout: float | httpx.Timeout | None = None,
client: HTTPHandler | None = None,
@ -93,7 +93,7 @@ class AgentsHTTPHandler(InteractionsHTTPHandler):
name: str,
litellm_params: GenericLiteLLMParams,
logging_obj: LiteLLMLoggingObj,
extra_headers: dict[str, Any] | None = None,
extra_headers: dict[str, str] | None = None,
extra_body: Mapping[str, object] | None = None,
timeout: float | httpx.Timeout | None = None,
client: AsyncHTTPHandler | None = None,
@ -141,7 +141,7 @@ class AgentsHTTPHandler(InteractionsHTTPHandler):
agents_api_config: BaseAgentsAPIConfig,
litellm_params: GenericLiteLLMParams,
logging_obj: LiteLLMLoggingObj,
extra_headers: dict[str, Any] | None = None,
extra_headers: dict[str, str] | None = None,
timeout: float | httpx.Timeout | None = None,
client: HTTPHandler | None = None,
_is_async: bool = False,
@ -181,7 +181,7 @@ class AgentsHTTPHandler(InteractionsHTTPHandler):
agents_api_config: BaseAgentsAPIConfig,
litellm_params: GenericLiteLLMParams,
logging_obj: LiteLLMLoggingObj,
extra_headers: dict[str, Any] | None = None,
extra_headers: dict[str, str] | None = None,
timeout: float | httpx.Timeout | None = None,
client: AsyncHTTPHandler | None = None,
) -> AgentListResponse:
@ -216,7 +216,7 @@ class AgentsHTTPHandler(InteractionsHTTPHandler):
name: str,
litellm_params: GenericLiteLLMParams,
logging_obj: LiteLLMLoggingObj,
extra_headers: dict[str, Any] | None = None,
extra_headers: dict[str, str] | None = None,
timeout: float | httpx.Timeout | None = None,
client: HTTPHandler | None = None,
_is_async: bool = False,
@ -259,7 +259,7 @@ class AgentsHTTPHandler(InteractionsHTTPHandler):
name: str,
litellm_params: GenericLiteLLMParams,
logging_obj: LiteLLMLoggingObj,
extra_headers: dict[str, Any] | None = None,
extra_headers: dict[str, str] | None = None,
timeout: float | httpx.Timeout | None = None,
client: AsyncHTTPHandler | None = None,
) -> AgentCreateResponse:
@ -295,7 +295,7 @@ class AgentsHTTPHandler(InteractionsHTTPHandler):
name: str,
litellm_params: GenericLiteLLMParams,
logging_obj: LiteLLMLoggingObj,
extra_headers: dict[str, Any] | None = None,
extra_headers: dict[str, str] | None = None,
timeout: float | httpx.Timeout | None = None,
client: HTTPHandler | None = None,
_is_async: bool = False,
@ -338,7 +338,7 @@ class AgentsHTTPHandler(InteractionsHTTPHandler):
name: str,
litellm_params: GenericLiteLLMParams,
logging_obj: LiteLLMLoggingObj,
extra_headers: dict[str, Any] | None = None,
extra_headers: dict[str, str] | None = None,
timeout: float | httpx.Timeout | None = None,
client: AsyncHTTPHandler | None = None,
) -> AgentDeleteResult:
@ -374,7 +374,7 @@ class AgentsHTTPHandler(InteractionsHTTPHandler):
name: str,
litellm_params: GenericLiteLLMParams,
logging_obj: LiteLLMLoggingObj,
extra_headers: dict[str, Any] | None = None,
extra_headers: dict[str, str] | None = None,
timeout: float | httpx.Timeout | None = None,
client: HTTPHandler | None = None,
_is_async: bool = False,
@ -417,7 +417,7 @@ class AgentsHTTPHandler(InteractionsHTTPHandler):
name: str,
litellm_params: GenericLiteLLMParams,
logging_obj: LiteLLMLoggingObj,
extra_headers: dict[str, Any] | None = None,
extra_headers: dict[str, str] | None = None,
timeout: float | httpx.Timeout | None = None,
client: AsyncHTTPHandler | None = None,
) -> AgentVersionsResponse:

4
litellm/litellm_core_utils/logging_utils.py
View file

@ -67,7 +67,9 @@ def _truncate_base64_in_string(value: str) -> str:
return _DATA_URI_RE.sub(_base64_data_uri_replacer, value)
def _truncate_base64_in_value(value: Any) -> Any:
def _truncate_base64_in_value(
value: str | dict[str, object] | list[object] | None,
) -> str | dict[str, object] | list[object] | None:
"""Iteratively truncate base64 data URIs in a JSON-like value (str/list/dict).
Uses an explicit stack instead of recursion to satisfy the project's

4
litellm/litellm_core_utils/url_utils.py
View file

@ -418,7 +418,7 @@ def _extract_redirect_url(response: httpx.Response, request_url: str) -> str:
return str(httpx.URL(request_url).join(location))
def safe_get(client: Any, url: str, **kwargs: Any) -> httpx.Response:
def safe_get(client: _UrlFetcher, url: str, **kwargs: Any) -> httpx.Response:
"""
Fetch a user-supplied URL with SSRF protection on every redirect hop.
@ -461,7 +461,7 @@ def safe_get(client: Any, url: str, **kwargs: Any) -> httpx.Response:
raise SSRFError("Too many redirects")
async def async_safe_get(client: Any, url: str, **kwargs: Any) -> httpx.Response:
async def async_safe_get(client: _AsyncUrlFetcher, url: str, **kwargs: Any) -> httpx.Response:
"""Async version of safe_get."""
if not getattr(litellm, "user_url_validation", True):
kwargs.setdefault("follow_redirects", True)

2
litellm/llms/anthropic/chat/handler.py
View file

@ -596,7 +596,7 @@ class ModelResponseIterator:
self.reasoning_content_chunks: list[str] = []
# Track server tool use inputs and results for code_interpreter_results
self._server_tool_inputs: dict[str, Any] = {}
self._server_tool_inputs: dict[str, object] = {}
self.tool_results: list[dict[str, Any]] = []
self._current_server_tool_id: str | None = None
self._container_id: str | None = None

9
litellm/llms/azure/completion/handler.py
View file

@ -1,6 +1,7 @@
from collections.abc import Callable
from typing import Any, Final
from typing import Final
import httpx
from openai import AsyncAzureOpenAI, AzureOpenAI
from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
@ -191,7 +192,7 @@ class AzureTextCompletion(BaseAzureLLM):
model: str,
api_base: str,
data: dict,
timeout: Any,
timeout: float | httpx.Timeout | None,
model_response: ModelResponse,
logging_obj: LiteLLMLoggingObj,
max_retries: int,
@ -253,7 +254,7 @@ class AzureTextCompletion(BaseAzureLLM):
api_version: str,
data: dict,
model: str,
timeout: Any,
timeout: float | httpx.Timeout | None,
azure_ad_token: str | None = None,
client=None,
litellm_params: dict = {},
@ -306,7 +307,7 @@ class AzureTextCompletion(BaseAzureLLM):
api_version: str,
data: dict,
model: str,
timeout: Any,
timeout: float | httpx.Timeout | None,
azure_ad_token: str | None = None,
client=None,
litellm_params: dict = {},

8
litellm/llms/azure_ai/ocr/document_intelligence/transformation.py
View file

@ -12,7 +12,7 @@ import asyncio
import re
import time
from collections.abc import Mapping
from typing import TYPE_CHECKING, Any, Final
from typing import TYPE_CHECKING, Final
from urllib.parse import quote
import httpx
@ -127,7 +127,7 @@ class AzureDocumentIntelligenceOCRConfig(BaseOCRConfig):
def map_ocr_params(
self,
non_default_params: dict,
non_default_params: Mapping[str, object],
optional_params: dict,
model: str,
) -> dict:
@ -164,7 +164,7 @@ class AzureDocumentIntelligenceOCRConfig(BaseOCRConfig):
raise UnsupportedParamsError(message=f"{e}", model=model, llm_provider="azure_ai") from e
@staticmethod
def _normalize_pages_param(pages: Any) -> str:
def _normalize_pages_param(pages: object) -> str:
"""
Convert a caller-provided `pages` value to Azure DI's query-string
form. Azure expects 1-based page numbers, grammar: `^(\\d+(-\\d+)?)(,\\s*(\\d+(-\\d+)?))*$`.
@ -412,7 +412,7 @@ class AzureDocumentIntelligenceOCRConfig(BaseOCRConfig):
raise ValueError("Document URL is required")
# Build Azure DI request
data: Final[dict[str, Any]] = {}
data: Final[dict[str, str]] = {}
# Check if it's a data URI (base64)
if document_url.startswith("data:"):

10
litellm/llms/bedrock/batches/transformation.py
View file

@ -2,7 +2,7 @@ import os
import re
import time
from collections.abc import Mapping
from typing import TYPE_CHECKING, Any, Final, Literal, cast
from typing import TYPE_CHECKING, Final, Literal, cast
from httpx import Headers, Response
from pydantic import TypeAdapter, ValidationError
@ -170,7 +170,7 @@ class BedrockBatchesConfig(BaseAWSLLM, BaseBatchesConfig):
create_batch_data: CreateBatchRequest,
optional_params: dict,
litellm_params: dict,
) -> dict[str, Any]:
) -> dict[str, object]:
"""
Transform the batch creation request to Bedrock format.
@ -354,7 +354,7 @@ class BedrockBatchesConfig(BaseAWSLLM, BaseBatchesConfig):
)
@staticmethod
def _get_openai_compatible_batch_metadata(metadata: Any) -> dict[str, str]:
def _get_openai_compatible_batch_metadata(metadata: object) -> dict[str, str]:
"""
OpenAI Batch metadata only accepts string values.
"""
@ -379,7 +379,7 @@ class BedrockBatchesConfig(BaseAWSLLM, BaseBatchesConfig):
batch_id: str,
optional_params: dict,
litellm_params: dict,
) -> dict[str, Any]:
) -> dict[str, object]:
"""
Transform batch retrieval request for Bedrock.
@ -523,7 +523,7 @@ class BedrockBatchesConfig(BaseAWSLLM, BaseBatchesConfig):
)
# Enrich metadata with useful Bedrock fields
enriched_metadata_raw: Final[dict[str, Any]] = {
enriched_metadata_raw: Final[dict[str, object]] = {
"jobName": response_data.get("jobName"),
"clientRequestToken": response_data.get("clientRequestToken"),
"modelId": response_data.get("modelId"),

2
litellm/llms/bedrock/chat/invoke_transformations/amazon_twelvelabs_pegasus_transformation.py
View file

@ -110,7 +110,7 @@ class AmazonTwelveLabsPegasusConfig(AmazonInvokeConfig, BaseConfig):
headers: dict,
) -> dict:
input_prompt: Final = self._convert_messages_to_prompt(messages=messages)
request_data: Final[dict[str, Any]] = {"inputPrompt": input_prompt}
request_data: Final[dict[str, object]] = {"inputPrompt": input_prompt}
media_source: Final = self._build_media_source(optional_params)
if media_source is not None:

6
litellm/llms/bytez/chat/transformation.py
View file

@ -335,10 +335,10 @@ class BytezChatConfig(BaseConfig):
class BytezCustomStreamWrapper(CustomStreamWrapper):
def chunk_creator(self, chunk: Any):
def chunk_creator(self, chunk: object):
try:
model_response: Final = self.model_response_creator()
response_obj: dict[str, Any] = {}
response_obj: dict[str, object] = {}
response_obj = {
"text": chunk,
@ -346,7 +346,7 @@ class BytezCustomStreamWrapper(CustomStreamWrapper):
"finish_reason": "",
}
completion_obj: Final[dict[str, Any]] = {"content": chunk}
completion_obj: Final[dict[str, object]] = {"content": chunk}
return self.return_processed_chunk_logic(
completion_obj=completion_obj,

4
litellm/llms/custom_httpx/aiohttp_handler.py
View file

@ -1,5 +1,5 @@
import ssl
from collections.abc import Callable
from collections.abc import AsyncIterable, Callable, Iterable
from typing import TYPE_CHECKING, Any, Final, cast
import aiohttp
@ -212,7 +212,7 @@ class BaseLLMAIOHTTPHandler:
litellm_params: dict,
stream: bool = False,
files: dict | None = None,
content: Any = None,
content: str | bytes | Iterable[bytes] | AsyncIterable[bytes] | None = None,
params: dict | None = None,
) -> httpx.Response:
max_retry_on_unprocessable_entity_error: Final = provider_config.max_retry_on_unprocessable_entity_error

2
litellm/llms/deprecated_providers/aleph_alpha.py
View file

@ -146,7 +146,7 @@ class AlephAlphaConfig:
setattr(self.__class__, key, value)
@classmethod
def get_config(cls):
def get_config(cls) -> dict[str, object]:
return {
k: v
for k, v in cls.__dict__.items()

2
litellm/llms/lemonade/chat/transformation.py
View file

@ -170,7 +170,7 @@ class LemonadeChatConfig(OpenAILikeChatConfig):
model: str,
api_base: str | None = None,
api_key: str | None = None,
) -> Any:
) -> dict[str, object]:
if model.startswith("lemonade/"):
model = model.split("/", 1)[1]

2
litellm/llms/openai/image_edit/transformation.py
View file

@ -66,7 +66,7 @@ class OpenAIImageEditConfig(BaseImageEditConfig):
def _add_image_to_files(
self,
files_list: list[tuple[str, Any]],
image: Any,
image: object,
field_name: str,
) -> None:
"""Add an image to the files list with appropriate content type"""

2
litellm/llms/runwayml/text_to_speech/transformation.py
View file

@ -78,7 +78,7 @@ class RunwayMLTextToSpeechConfig(BaseTextToSpeechConfig):
aspeech: bool,
api_base: str | None,
api_key: str | None,
**kwargs: Any,
**kwargs: object,
) -> Union[
"HttpxBinaryResponseContent",
Coroutine[object, object, "HttpxBinaryResponseContent"],

6
litellm/llms/vertex_ai/files/transformation.py
View file

@ -651,7 +651,7 @@ def _openai_batch_jsonl_entry_to_vertex_embeddings_rows(
def _openai_batch_jsonl_entry_to_vertex_rows(
openai_entry: dict[str, Any],
map_openai_to_vertex_params: Callable[[dict[str, Any]], dict[str, Any]],
map_openai_to_vertex_params: Callable[[dict[str, Any]], dict[str, object]],
) -> tuple[Mapping[str, object], ...]:
"""
Transforms a single OpenAI JSONL batch entry into the Vertex rows it maps to.
@ -774,7 +774,7 @@ class _OpenAIToVertexBatchUploadStream(BaseFileUploadStream):
def __init__(
self,
openai_file_content: FileTypes,
map_openai_to_vertex_params: Callable[[dict[str, Any]], dict[str, Any]],
map_openai_to_vertex_params: Callable[[dict[str, Any]], dict[str, object]],
) -> None:
self._openai_file_content = openai_file_content
self._map_openai_to_vertex_params = map_openai_to_vertex_params
@ -948,7 +948,7 @@ class VertexAIFilesConfig(VertexBase, BaseFilesConfig):
def _map_openai_to_vertex_params(
self,
openai_request_body: dict[str, Any],
) -> dict[str, Any]:
) -> dict[str, object]:
"""
wrapper to call VertexGeminiConfig.map_openai_params
"""

4
litellm/llms/vertex_ai/gemini_embeddings/batch_embed_content_handler.py
View file

@ -3,7 +3,7 @@ Google AI Studio /batchEmbedContents Embeddings Endpoint
"""
import json
from typing import TYPE_CHECKING, Any, Final, Literal
from typing import TYPE_CHECKING, Final, Literal
import httpx
@ -210,7 +210,7 @@ class GoogleBatchEmbeddings(VertexLLM):
)
### TRANSFORMATION (sync path) ###
request_data: Any
request_data: VertexAIBatchEmbeddingsRequestBody | dict[str, object]
if use_embed_content:
resolved_files = {}
if api_key:

2
litellm/llms/vertex_ai/vertex_ai_non_gemini.py
View file

@ -64,7 +64,7 @@ def _get_client_from_cache(client_cache_key: str):
return litellm.in_memory_llm_clients_cache.get_cache(client_cache_key)
def _set_client_in_cache(client_cache_key: str, vertex_llm_model: Any):
def _set_client_in_cache(client_cache_key: str, vertex_llm_model: object):
litellm.in_memory_llm_clients_cache.set_cache(
key=client_cache_key,
value=vertex_llm_model,

4
litellm/llms/watsonx/audio_transcription/transformation.py
View file

@ -4,7 +4,7 @@ Translates from OpenAI's `/v1/audio/transcriptions` to IBM WatsonX's `/ml/v1/aud
WatsonX follows the OpenAI spec for audio transcription.
"""
from typing import Any, Final
from typing import Final
from httpx import Response
@ -124,7 +124,7 @@ class IBMWatsonXAudioTranscriptionConfig(IBMWatsonXMixin, OpenAIWhisperAudioTran
}
# Convert TypedDict to regular dict for AudioTranscriptionRequestData
form_data_dict: Final[dict[str, Any]] = dict(form_data)
form_data_dict: Final[dict[str, object]] = dict(form_data)
return AudioTranscriptionRequestData(data=form_data_dict, files=files)

16
litellm/proxy/a2a/agent_card.py
View file

@ -10,7 +10,7 @@ and uses LiteLLM auth.
import re
from collections.abc import Mapping
from copy import deepcopy
from typing import Any, Final, Literal
from typing import Final, Literal
SupportedA2AVersion = Literal["0.3", "1.0"]
@ -44,7 +44,7 @@ def normalize_protocol_version(version: object) -> SupportedA2AVersion | None:
return next((supported for supported in SUPPORTED_A2A_PROTOCOL_VERSIONS if supported == major_minor), None)
def resolve_served_protocol_version(card: Mapping[str, Any] | None) -> str:
def resolve_served_protocol_version(card: Mapping[str, object] | None) -> str:
"""Return the validated protocol version an agent card pins, else the default."""
normalized: Final = normalize_protocol_version(card.get("protocolVersion") if card else None)
return normalized if normalized is not None else LITELLM_A2A_PROTOCOL_VERSION
@ -53,7 +53,7 @@ def resolve_served_protocol_version(card: Mapping[str, Any] | None) -> str:
# Security scheme exposed by the LiteLLM-fronted agent card. Always replaces
# whatever upstream advertised — the client must authenticate to the proxy,
# not the upstream agent.
LITELLM_SECURITY_SCHEMES: Final[dict[str, dict[str, Any]]] = {
LITELLM_SECURITY_SCHEMES: Final[dict[str, dict[str, str]]] = {
"LiteLLMKey": {
"type": "http",
"scheme": "bearer",
@ -112,7 +112,7 @@ _ALLOWED_TOP_LEVEL_KEYS: Final = {
"url",
}
_DEFAULT_SKILLS: Final[list[dict[str, Any]]] = [
_DEFAULT_SKILLS: Final[list[dict[str, str | list[str]]]] = [
{
"id": "chat",
"name": "Chat",
@ -129,7 +129,7 @@ _DEFAULT_MODES: Final[list[str]] = ["text"]
_DEFAULT_AGENT_VERSION: Final = "1.0.0"
def _filter_capabilities(upstream_capabilities: Any) -> dict[str, Any]:
def _filter_capabilities(upstream_capabilities: object) -> dict[str, object]:
"""Return a capabilities dict containing only allowlisted, truthy keys."""
if not isinstance(upstream_capabilities, dict):
return {}
@ -143,13 +143,13 @@ def _default_litellm_provider(proxy_base_url: str) -> dict[str, str]:
def merge_agent_card(
upstream_card: Mapping[str, Any] | None,
upstream_card: Mapping[str, object] | None,
*,
proxy_url: str,
proxy_base_url: str,
name: str | None = None,
description: str | None = None,
) -> dict[str, Any]:
) -> dict[str, object]:
"""
Build the LiteLLM-fronted agent card.
@ -169,7 +169,7 @@ def merge_agent_card(
A dict suitable for serving as the proxy's agent card. Only keys in
the v1.0 AgentCard schema (plus ``supportedInterfaces``) are emitted.
"""
base: Final[dict[str, Any]] = deepcopy(dict(upstream_card)) if upstream_card else {}
base: Final[dict[str, object]] = deepcopy(dict(upstream_card)) if upstream_card else {}
# Keep the upstream ``url`` on the stored card: the runtime A2A
# invocation path reads it from ``agent_card_params`` to know where to

5
litellm/proxy/client/credentials.py
View file

@ -1,3 +1,4 @@
from collections.abc import Mapping
from typing import Any, Final
import requests
@ -69,8 +70,8 @@ class CredentialsManagementClient:
def create(
self,
credential_name: str,
credential_info: dict[str, Any],
credential_values: dict[str, Any],
credential_info: Mapping[str, object],
credential_values: Mapping[str, object],
return_request: bool = False,
) -> dict[str, Any] | requests.Request:
"""

4
litellm/proxy/guardrails/guardrail_hooks/guardrails_ai/guardrails_ai.py
View file

@ -7,7 +7,7 @@
import json
import os
from typing import TYPE_CHECKING, Any, Final, Literal, TypedDict
from typing import TYPE_CHECKING, Final, Literal, TypedDict
from fastapi import HTTPException
@ -181,7 +181,7 @@ class GuardrailsAI(CustomGuardrail):
): # raise exception if invalid, return a str for the user to receive - if rejected, or return a modified dictionary for passing into litellm
return await self.process_input(data=data, call_type=call_type)
async def async_logging_hook(self, kwargs: dict, result: Any, call_type: str) -> tuple[dict, Any]:
async def async_logging_hook(self, kwargs: dict, result: object, call_type: str) -> tuple[dict, object]:
if call_type == "acompletion" or call_type == "completion":
kwargs = await self.process_input(data=kwargs, call_type=call_type)

6
litellm/proxy/guardrails/guardrail_hooks/singulr/singulr.py
View file

@ -24,7 +24,7 @@ from litellm.llms.custom_httpx.http_handler import (
)
from litellm.proxy._types import UserAPIKeyAuth
from litellm.types.guardrails import GuardrailEventHooks
from litellm.types.llms.openai import AllMessageValues
from litellm.types.llms.openai import AllMessageValues, ChatCompletionToolCallChunk
from litellm.types.proxy.guardrails.guardrail_hooks.base import (
GuardrailConfigModel,
)
@ -36,7 +36,7 @@ from litellm.types.proxy.guardrails.guardrail_hooks.singulr import (
ToolCall,
ToolCallFunction,
)
from litellm.types.utils import CallTypes, GenericGuardrailAPIInputs
from litellm.types.utils import CallTypes, ChatCompletionMessageToolCall, GenericGuardrailAPIInputs
_DEFAULT_API_BASE: Final = "http://localhost:8003"
_GUARD_ENDPOINT: Final = "/api/v1/ai-gateway/litellm-v2"
@ -339,7 +339,7 @@ class SingulrGuardrail(CustomGuardrail):
return inputs
@staticmethod
def _build_tool_call(tool_call: Mapping[str, Any]) -> "ToolCall | None":
def _build_tool_call(tool_call: ChatCompletionToolCallChunk | ChatCompletionMessageToolCall) -> "ToolCall | None":
tool_call_id: Final = tool_call.get("id")
fun: Final = tool_call.get("function")
if not tool_call_id or not fun:

12
litellm/proxy/policy_engine/pipeline_executor.py
View file

@ -8,7 +8,7 @@ pass/fail actions (allow, block, next, modify_response) and data forwarding.
import copy
import time
from collections.abc import Callable, Mapping, Sequence
from typing import TYPE_CHECKING, Any, Final, Literal, TypeVar
from typing import TYPE_CHECKING, Final, Literal, TypeVar
from pydantic import BaseModel
@ -314,11 +314,11 @@ class PipelineExecutor:
steps: list[PipelineStep],
mode: str,
data: dict,
user_api_key_dict: Any,
user_api_key_dict: "UserAPIKeyAuth",
call_type: str,
policy_name: str,
raw_request_snapshot: dict | None = None, # mutable-ok: same request-payload shape as data
streaming_chunks: list[Any] | None = None, # mutable-ok: shared buffered-stream chunks, read per step
streaming_chunks: list[object] | None = None, # mutable-ok: shared buffered-stream chunks, read per step
endpoint_translation: "BaseTranslation | None" = None,
) -> PipelineExecutionResult:
"""
@ -490,10 +490,10 @@ class PipelineExecutor:
step: PipelineStep,
mode: str,
data: dict,
user_api_key_dict: Any,
user_api_key_dict: "UserAPIKeyAuth",
call_type: str,
raw_request_snapshot: dict | None = None, # mutable-ok: same request-payload shape as data
streaming_chunks: list[Any] | None = None, # mutable-ok: shared buffered-stream chunks, read per step
streaming_chunks: list[object] | None = None, # mutable-ok: shared buffered-stream chunks, read per step
endpoint_translation: "BaseTranslation | None" = None,
) -> tuple[
Literal["pass", "fail", "error"],
@ -722,7 +722,7 @@ def _extract_error_message(e: Exception) -> str:
if isinstance(e, ModifyResponseException):
return str(e)
if HTTPException is not None and isinstance(e, HTTPException):
detail: Final = getattr(e, "detail", None)
detail: Final[object] = getattr(e, "detail", None)
if detail:
return str(detail)
return str(e)

2
litellm/router_strategy/adaptive_router/hooks.py
View file

@ -86,7 +86,7 @@ def _resolve_session_key(kwargs: dict[str, Any]) -> str | None:
return hashlib.sha256(payload.encode("utf-8")).hexdigest()
def _last_user_content(messages: list[dict[str, Any]] | None) -> str | None:
def _last_user_content(messages: Sequence[Mapping[str, object]] | None) -> str | None:
if not messages:
return None
for msg in reversed(messages):

4
litellm/router_strategy/auto_router/auto_router.py
View file

@ -3,7 +3,7 @@ Auto-Routing Strategy that works with a Semantic Router Config
"""
import asyncio
from collections.abc import Mapping
from collections.abc import Mapping, Sequence
from typing import TYPE_CHECKING, Any, Final, Optional
from pydantic import BaseModel, ConfigDict
@ -158,7 +158,7 @@ class AutoRouter(CustomLogger):
return await asyncio.shield(build_task)
@staticmethod
def _extract_text_from_messages(messages: list[dict[str, Any]]) -> str:
def _extract_text_from_messages(messages: Sequence[Mapping[str, object]]) -> str:
"""
Extract text content from the last user message for routing.

43
scripts/budget_ratchet_check.py
View file

@ -7,13 +7,17 @@ driven DOWN over time. This check compares every budget file against its own
content at the merge-base with the target branch and fails (exits 1, red) if:
* a rule's `limit` went up,
* a rule was dropped from a budget (its ceiling effectively became infinite), or
* a rule was dropped from a budget (its ceiling effectively became infinite) while
its checker still emits it, or
* an entire budget file was deleted.
New rules and lowered/equal limits are fine. So is a rule that graduated: once a
paired config (ruff.toml for the ruff-strict budget) selects the rule outright it
hard-fails at the first violation, which is stricter than any ceiling the budget
could hold, so dropping its entry tightens the guard rather than removing it.
Likewise a retired rule: once the paired checker (check_test_quality.py for the
test-quality budget) no longer emits a code, its entry has no ceiling left to
loosen.
This is deliberately NOT a gating check. It should turn the run red so that a
loosening is impossible to miss in review, but it must stay OUT of the
@ -29,11 +33,12 @@ Usage:
from __future__ import annotations
import argparse
import importlib.util
import json
import subprocess
import sys
from pathlib import Path
from types import MappingProxyType
from types import MappingProxyType, ModuleType
from typing import Final, NamedTuple
if sys.version_info >= (3, 11):
@ -49,6 +54,7 @@ DEFAULT_BUDGETS: tuple[str, ...] = (
"test-quality-budget.json",
)
GRADUATION_CONFIGS = MappingProxyType({"ruff-strict-budget.json": "ruff.toml"})
RETIREMENT_SOURCES = MappingProxyType({"test-quality-budget.json": "check_test_quality"})
class Regression(NamedTuple):
@ -139,20 +145,40 @@ def graduated_selectors(rel: str) -> tuple[str, ...]:
)
def _load_script(name: str) -> ModuleType:
if name in sys.modules:
return sys.modules[name]
spec: Final = importlib.util.spec_from_file_location(name, REPO_ROOT / "scripts" / f"{name}.py")
assert spec is not None and spec.loader is not None
module: Final = importlib.util.module_from_spec(spec)
sys.modules[name] = module
spec.loader.exec_module(module)
return module
def retired_rules(rel: str, base: dict[str, object]) -> frozenset[str]:
"""Rules in the base budget that the paired checker can no longer emit, so there is no ceiling to loosen."""
source: Final = RETIREMENT_SOURCES.get(rel)
if source is None:
return frozenset()
return frozenset(_limits(base)) - _load_script(source).RULE_CODES
def _regression_detail(
rule: str,
base_limits: dict[str, int],
head_limits: dict[str, int],
graduated: tuple[str, ...],
retired: frozenset[str] = frozenset(),
) -> str | None:
"""Why `rule` regressed vs base, or None when it held flat, fell, or graduated.
"""Why `rule` regressed vs base, or None when it held flat, fell, or left the budget legitimately.
A dropped rule is terminal unless it graduated; otherwise the only loosening
left is a raised limit.
A dropped rule is terminal unless it graduated or retired; otherwise the only
loosening left is a raised limit.
"""
base_limit = base_limits[rule]
if rule not in head_limits:
if graduated and rule.startswith(graduated):
if rule in retired or (graduated and rule.startswith(graduated)):
return None
return f"rule dropped (limit {base_limit} -> removed)"
if head_limits[rule] > base_limit:
@ -165,6 +191,7 @@ def regressions_for(
base: dict | None,
head: dict | None,
graduated: tuple[str, ...] = (),
retired: frozenset[str] = frozenset(),
) -> list[Regression]:
if base is None:
return [] # new budget file: nothing to ratchet against yet
@ -175,7 +202,7 @@ def regressions_for(
return [
Regression(rel, rule, detail)
for rule in sorted(base_limits)
if (detail := _regression_detail(rule, base_limits, head_limits, graduated)) is not None
if (detail := _regression_detail(rule, base_limits, head_limits, graduated, retired)) is not None
]
@ -209,7 +236,7 @@ def main() -> int:
print(f"skip {rel}: new file (no base at {base_ref} to ratchet against)")
continue
checked.append(rel)
regressions.extend(regressions_for(rel, base, head, graduated_selectors(rel)))
regressions.extend(regressions_for(rel, base, head, graduated_selectors(rel), retired_rules(rel, base)))
if regressions:
print(

4
scripts/check_test_quality.py
View file

@ -146,6 +146,10 @@ SDK_MODULE: Final = "litellm"
SUBPROCESS_SPAWNS: Final = frozenset(("run", "Popen", "check_output", "check_call", "call"))
INTERPRETER_ISOLATION_FLAGS: Final = frozenset(("-I", "-P"))
RULE_CODES: Final = frozenset((
"TQ000", "TQ001", "TQ002", "TQ003", "TQ004", "TQ005", "TQ006", "TQ007", "TQ009",
))
CREDENTIAL_NAME_RE: Final = re.compile(
r"(?:API_KEY|_KEY|TOKEN|SECRET|PASSWORD|CREDENTIAL|DATABASE_URL|ACCESS_KEY_ID)$"
)

195
tests/test_litellm/llms/amazon_nova/chat/test_amazon_nova_chat_completion.py
View file

@ -1,195 +0,0 @@
import os
import pytest
# Ensure the project root is on the import path
from litellm import completion
from litellm.types.utils import ModelResponse, Usage, Choices, Message
def _has_api_key() -> bool:
"""Check if Amazon Nova API key is available"""
return (
"AMAZON_NOVA_API_KEY" in os.environ
and os.environ["AMAZON_NOVA_API_KEY"] is not None
)
def _create_mock_nova_response():
"""Helper function to create mock Amazon Nova response for testing"""
return ModelResponse(
id="chatcmpl-test-nova-micro",
choices=[
Choices(
finish_reason="stop",
index=0,
message=Message(
content="I am Amazon Nova Micro. 777 times 9 equals 6993.",
role="assistant",
),
)
],
created=1234567890,
model="amazon-nova/nova-micro-v1",
object="chat.completion",
usage=Usage(prompt_tokens=25, completion_tokens=15, total_tokens=40),
)
def test_amazon_nova_chat_completion_nova_micro():
if _has_api_key():
response: ModelResponse = completion(
model="amazon-nova/nova-micro-v1",
messages=[
{"role": "system", "content": "You are a helpful assistant"},
{
"role": "user",
"content": "What model are you? Can you calculate 777 times 9?",
},
],
api_key=os.environ["AMAZON_NOVA_API_KEY"],
)
else:
# Use mock response when API key is not available
response = _create_mock_nova_response()
# Additional mock-specific assertions for code review reference
assert (
response.choices[0].message.content
== "I am Amazon Nova Micro. 777 times 9 equals 6993."
)
assert response.model == "amazon-nova/nova-micro-v1"
assert response.usage.prompt_tokens == 25
assert response.usage.completion_tokens == 15
assert response.object == "chat.completion"
assert response.choices[0].finish_reason == "stop"
assert response.choices[0].message.role == "assistant"
# Common assertions for both real and mock responses
assert response is not None
assert hasattr(response, "choices")
assert len(response.choices) > 0
assert response.choices[0].message.content is not None
assert response.usage.total_tokens > 0
@pytest.mark.skipif(not _has_api_key(), reason="Amazon Nova API key not available")
def test_amazon_nova_chat_completion_nova_lite():
response: ModelResponse = completion(
model="amazon-nova/nova-lite-v1",
messages=[
{"role": "system", "content": "You are a helpful assistant"},
{
"role": "user",
"content": "What model are you? Please tell me a poem on rain",
},
],
api_key=os.environ["AMAZON_NOVA_API_KEY"],
)
assert response is not None
assert hasattr(response, "choices")
assert len(response.choices) > 0
assert response.choices[0].message.content is not None
assert response.usage.total_tokens > 0
@pytest.mark.skipif(not _has_api_key(), reason="Amazon Nova API key not available")
def test_amazon_nova_chat_completion_nova_pro():
response: ModelResponse = completion(
model="amazon-nova/nova-pro-v1",
messages=[
{"role": "system", "content": "You are a helpful assistant"},
{
"role": "user",
"content": "What model are you? What is MCP server and how does that help in building GenAI applications?",
},
],
timeout=30,
api_key=os.environ["AMAZON_NOVA_API_KEY"],
)
assert response is not None
assert hasattr(response, "choices")
assert len(response.choices) > 0
assert response.choices[0].message.content is not None
assert response.usage.total_tokens > 0
@pytest.mark.skipif(not _has_api_key(), reason="Amazon Nova API key not available")
def test_amazon_nova_chat_completion_nova_premier():
response: ModelResponse = completion(
model="amazon-nova/nova-premier-v1",
messages=[
{"role": "system", "content": "You are a helpful assistant"},
{
"role": "user",
"content": "What model are you? Can you help me understand what Trigonometry is?",
},
],
timeout=60,
api_key=os.environ["AMAZON_NOVA_API_KEY"],
)
assert response is not None
print(response.choices[0].message.content)
assert hasattr(response, "choices")
assert len(response.choices) > 0
assert response.choices[0].message.content is not None
assert response.usage.total_tokens > 0
@pytest.mark.skipif(not _has_api_key(), reason="Amazon Nova API key not available")
def test_amazon_nova_chat_completion_with_tool_usage():
response: ModelResponse = completion(
model="amazon-nova/nova-micro-v1",
messages=[
{"role": "system", "content": "You are a helpful assistant"},
{"role": "user", "content": "What is the temperature in SFO?"},
],
tools=[
{
"type": "function",
"function": {
"name": "getCurrentWeather",
"description": "Get the current weather in a given city",
"parameters": {
"type": "object",
"properties": {
"location": {
"type": "string",
"description": "City and country e.g. Bogotá, Colombia",
}
},
"required": ["location"],
},
},
}
],
api_key=os.environ["AMAZON_NOVA_API_KEY"],
)
assert response is not None
assert hasattr(response, "choices")
assert len(response.choices) > 0
assert response.choices[0].message is not None
@pytest.mark.skipif(not _has_api_key(), reason="Amazon Nova API key not available")
def test_amazon_nova_chat_completion_with_stream_response():
response = completion(
model="amazon-nova/nova-micro-v1",
stream=True,
messages=[
{"role": "system", "content": "You are a helpful assistant"},
{
"role": "user",
"content": "What are MMO games? Can you give me some sample references?",
},
],
api_key=os.environ["AMAZON_NOVA_API_KEY"],
)
assert response is not None
chunks = list(response)
assert chunks is not None
assert len(chunks) > 0

115
tests/test_litellm/llms/bedrock/files/test_bedrock_files_integration.py
View file

@ -1,115 +0,0 @@
"""
Test Bedrock files integration with main files API
"""
import base64
from unittest.mock import MagicMock, patch
import pytest
import litellm
from litellm.types.llms.openai import HttpxBinaryResponseContent
from litellm.types.utils import SpecialEnums
class TestBedrockFilesIntegration:
"""Test integration of Bedrock files with main litellm API"""
@pytest.mark.asyncio
async def test_litellm_afile_content_bedrock_provider_with_s3_uri(self):
"""Test litellm.afile_content with bedrock provider using direct S3 URI"""
file_id = "s3://test-bucket/test-file.jsonl"
expected_content = (
b'{"recordId": "request-1", "modelInput": {}, "modelOutput": {}}'
)
# Create a mock HttpxBinaryResponseContent response
import httpx
mock_response = httpx.Response(
status_code=200,
content=expected_content,
headers={"content-type": "application/octet-stream"},
request=httpx.Request(method="GET", url="s3://test-bucket/test-file.jsonl"),
)
mock_result = HttpxBinaryResponseContent(response=mock_response)
# Mock the base_llm_http_handler.retrieve_file_content since the code
# now routes through ProviderConfigManager -> base_llm_http_handler
with patch(
"litellm.files.main.base_llm_http_handler.retrieve_file_content",
new_callable=MagicMock,
) as mock_retrieve:
mock_retrieve.return_value = mock_result
# Call litellm.afile_content
result = await litellm.afile_content(
file_id=file_id,
custom_llm_provider="bedrock",
aws_region_name="us-west-2",
)
# Verify the result
assert isinstance(result, HttpxBinaryResponseContent)
assert result.response.content == expected_content
assert result.response.status_code == 200
# Verify the mock was called with correct parameters
mock_retrieve.assert_called_once()
call_kwargs = mock_retrieve.call_args.kwargs
assert call_kwargs["_is_async"] is True
assert call_kwargs["file_content_request"]["file_id"] == file_id
@pytest.mark.asyncio
async def test_litellm_afile_content_bedrock_provider_with_unified_file_id(self):
"""Test litellm.afile_content with bedrock provider using unified file ID"""
# Create a unified file ID
s3_uri = "s3://test-bucket/batch-outputs/output.jsonl"
unified_id = "test-unified-id-123"
model_id = "test-model-id-456"
unified_file_id_str = f"litellm_proxy:application/json;unified_id,{unified_id};target_model_names,;llm_output_file_id,{s3_uri};llm_output_file_model_id,{model_id}"
encoded_file_id = (
base64.urlsafe_b64encode(unified_file_id_str.encode()).decode().rstrip("=")
)
expected_content = (
b'{"recordId": "request-1", "modelInput": {}, "modelOutput": {}}'
)
# Create a mock HttpxBinaryResponseContent response
import httpx
mock_response = httpx.Response(
status_code=200,
content=expected_content,
headers={"content-type": "application/octet-stream"},
request=httpx.Request(method="GET", url=s3_uri),
)
mock_result = HttpxBinaryResponseContent(response=mock_response)
# Mock the base_llm_http_handler.retrieve_file_content
with patch(
"litellm.files.main.base_llm_http_handler.retrieve_file_content",
new_callable=MagicMock,
) as mock_retrieve:
mock_retrieve.return_value = mock_result
# Call litellm.afile_content with unified file ID
result = await litellm.afile_content(
file_id=encoded_file_id,
custom_llm_provider="bedrock",
aws_region_name="us-west-2",
)
# Verify the result
assert isinstance(result, HttpxBinaryResponseContent)
assert result.response.content == expected_content
assert result.response.status_code == 200
# Verify the mock was called
mock_retrieve.assert_called_once()
call_kwargs = mock_retrieve.call_args.kwargs
assert call_kwargs["_is_async"] is True
# The handler passes the encoded file_id as-is
assert call_kwargs["file_content_request"]["file_id"] == encoded_file_id

158
tests/test_litellm/llms/bedrock/image/test_bedrock_image_bearer_token.py
View file

@ -1,158 +0,0 @@
import json
import os
from unittest.mock import Mock, patch
import pytest
import litellm
from litellm.llms.custom_httpx.http_handler import HTTPHandler, AsyncHTTPHandler
# Mock response for Bedrock image generation
mock_image_response = {"images": ["base64_encoded_image_data"], "error": None}
class TestBedrockImageGeneration:
def test_image_generation_with_api_key_bearer_token(self):
"""Test image generation with bearer token authentication"""
test_api_key = "test-bearer-token-12345"
model = "bedrock/stability.sd3-large-v1:0"
prompt = "A cute baby sea otter"
with patch(
"litellm.llms.bedrock.image_generation.image_handler.BedrockImageGeneration.image_generation"
) as mock_bedrock_image_gen:
# Setup mock response
mock_image_response_obj = litellm.ImageResponse()
mock_image_response_obj.data = [{"url": "https://example.com/image.jpg"}]
mock_bedrock_image_gen.return_value = mock_image_response_obj
response = litellm.image_generation(
model=model,
prompt=prompt,
aws_region_name="us-west-2",
api_key=test_api_key,
)
assert response is not None
assert len(response.data) > 0
mock_bedrock_image_gen.assert_called_once()
for call in mock_bedrock_image_gen.call_args_list:
if "headers" in call.kwargs:
headers = call.kwargs["headers"]
if (
"Authorization" in headers
and headers["Authorization"] == f"Bearer {test_api_key}"
):
break
def test_image_generation_with_env_variable_bearer_token(self, monkeypatch):
"""Test image generation with bearer token from environment variable"""
test_api_key = "env-bearer-token-12345"
model = "bedrock/stability.sd3-large-v1:0"
prompt = "A cute baby sea otter"
# Mock the environment variable
with (
patch.dict(os.environ, {"AWS_BEARER_TOKEN_BEDROCK": test_api_key}),
patch(
"litellm.llms.bedrock.image_generation.image_handler.BedrockImageGeneration.image_generation"
) as mock_bedrock_image_gen,
):
mock_image_response_obj = litellm.ImageResponse()
mock_image_response_obj.data = [{"url": "https://example.com/image.jpg"}]
mock_bedrock_image_gen.return_value = mock_image_response_obj
response = litellm.image_generation(
model=model, prompt=prompt, aws_region_name="us-west-2"
)
assert response is not None
assert len(response.data) > 0
mock_bedrock_image_gen.assert_called_once()
for call in mock_bedrock_image_gen.call_args_list:
if "headers" in call.kwargs:
headers = call.kwargs["headers"]
if (
"Authorization" in headers
and headers["Authorization"] == f"Bearer {test_api_key}"
):
break
@pytest.mark.asyncio
async def test_async_image_generation_with_bearer_token(self):
"""Test async image generation with bearer token authentication"""
test_api_key = "async-bearer-token-12345"
model = "bedrock/stability.sd3-large-v1:0"
prompt = "A cute baby sea otter"
with patch(
"litellm.llms.bedrock.image_generation.image_handler.BedrockImageGeneration.async_image_generation"
) as mock_async_bedrock_image_gen:
mock_image_response_obj = litellm.ImageResponse()
mock_image_response_obj.data = [{"url": "https://example.com/image.jpg"}]
mock_async_bedrock_image_gen.return_value = mock_image_response_obj
# Call async image generation with api_key parameter
response = await litellm.aimage_generation(
model=model,
prompt=prompt,
aws_region_name="us-west-2",
api_key=test_api_key,
)
assert response is not None
assert len(response.data) > 0
mock_async_bedrock_image_gen.assert_called_once()
for call in mock_async_bedrock_image_gen.call_args_list:
if "headers" in call.kwargs:
headers = call.kwargs["headers"]
if (
"Authorization" in headers
and headers["Authorization"] == f"Bearer {test_api_key}"
):
break
def test_image_generation_with_sigv4(self):
"""Test image generation falls back to SigV4 auth when no bearer token is provided"""
model = "bedrock/stability.sd3-large-v1:0"
prompt = "A cute baby sea otter"
with patch(
"litellm.llms.bedrock.image_generation.image_handler.BedrockImageGeneration.image_generation"
) as mock_bedrock_image_gen:
mock_image_response_obj = litellm.ImageResponse()
mock_image_response_obj.data = [{"url": "https://example.com/image.jpg"}]
mock_bedrock_image_gen.return_value = mock_image_response_obj
response = litellm.image_generation(
model=model, prompt=prompt, aws_region_name="us-west-2"
)
assert response is not None
assert len(response.data) > 0
mock_bedrock_image_gen.assert_called_once()
def test_image_generation_bearer_token_never_runs_the_sigv4_credential_chain(monkeypatch):
"""The deployment's AWS profile does not exist, so resolving SigV4 credentials
raises; a bearer-token deployment must still sign the request with the
bearer token alone."""
from litellm.llms.bedrock.image_generation.image_handler import BedrockImageGeneration
monkeypatch.setenv("AWS_BEARER_TOKEN_BEDROCK", "env-bearer-token-12345")
request = BedrockImageGeneration()._prepare_request(
model="amazon.nova-canvas-v1:0",
prompt="A cute baby sea otter",
optional_params={"aws_region_name": "us-west-2", "aws_profile_name": "litellm-no-such-aws-profile"},
api_base=None,
extra_headers=None,
api_key=None,
logging_obj=Mock(),
)
assert request.prepped.headers["Authorization"] == "Bearer env-bearer-token-12345"

19
tests/test_litellm/llms/fal_ai/test_cost_calculator.py
View file

@ -1,19 +0,0 @@
import pytest
import litellm
from litellm.litellm_core_utils.llm_cost_calc.utils import CostCalculatorUtils
from litellm.llms.fal_ai.cost_calculator import cost_calculator
from litellm.types.utils import ImageObject, ImageResponse
@pytest.fixture(autouse=True)
def _use_local_model_cost_map(monkeypatch):
monkeypatch.setenv("LITELLM_LOCAL_MODEL_COST_MAP", "True")
monkeypatch.setattr(litellm, "model_cost", litellm.get_model_cost_map(url=""))
litellm.get_model_info.cache_clear()
yield
litellm.get_model_info.cache_clear()
def _image_response(num_images: int = 1) -> ImageResponse:
return ImageResponse(data=[ImageObject(url="https://example.com/img.png") for _ in range(num_images)])

147
tests/test_litellm/llms/hosted_vllm/chat/test_hosted_vllm_ssl_verify.py
View file

@ -1,147 +0,0 @@
"""
Test SSL verification for hosted_vllm provider.
This test ensures that the ssl_verify parameter is properly passed through
to the HTTP client when using the hosted_vllm provider.
Issue: ssl_verify parameter was being ignored because hosted_vllm fell through
to the OpenAI catch-all path in main.py, which doesn't pass ssl_verify to the HTTP client.
"""
from unittest.mock import MagicMock, patch
import pytest
import litellm
class TestHostedVLLMSSLVerify:
"""Test suite for SSL verification in hosted_vllm provider."""
@patch("litellm.llms.custom_httpx.llm_http_handler._get_httpx_client")
def test_hosted_vllm_ssl_verify_false_sync(self, mock_get_httpx_client):
"""Test that ssl_verify=False is passed to the HTTP client for sync calls."""
# Setup mock client
mock_client = MagicMock()
mock_response = MagicMock()
mock_response.status_code = 200
mock_response.headers = {"content-type": "application/json"}
mock_response.json.return_value = {
"id": "chatcmpl-test",
"object": "chat.completion",
"created": 1234567890,
"model": "test-model",
"choices": [
{
"index": 0,
"message": {
"role": "assistant",
"content": "Test response",
},
"finish_reason": "stop",
}
],
"usage": {
"prompt_tokens": 10,
"completion_tokens": 5,
"total_tokens": 15,
},
}
mock_response.text = '{"id": "chatcmpl-test", "object": "chat.completion", "created": 1234567890, "model": "test-model", "choices": [{"index": 0, "message": {"role": "assistant", "content": "Test response"}, "finish_reason": "stop"}], "usage": {"prompt_tokens": 10, "completion_tokens": 5, "total_tokens": 15}}'
mock_client.post.return_value = mock_response
mock_get_httpx_client.return_value = mock_client
try:
litellm.completion(
model="hosted_vllm/test-model",
messages=[{"role": "user", "content": "Hello"}],
api_base="https://test-vllm.example.com/v1",
ssl_verify=False,
)
except Exception:
# Even if the response parsing fails, we just need to verify
# that the mock was called with the correct ssl_verify parameter
pass
# Verify _get_httpx_client was called with ssl_verify=False
mock_get_httpx_client.assert_called()
call_args = mock_get_httpx_client.call_args
# Check that params contains ssl_verify=False
if call_args[0]:
# Positional argument
params = call_args[0][0]
else:
# Keyword argument
params = call_args[1].get("params", {})
assert (
params.get("ssl_verify") is False
), f"Expected ssl_verify=False in params, got {params}"
@patch("litellm.llms.custom_httpx.llm_http_handler.get_async_httpx_client")
@pytest.mark.asyncio
async def test_hosted_vllm_ssl_verify_false_async(
self, mock_get_async_httpx_client
):
"""Test that ssl_verify=False is passed to the HTTP client for async calls."""
# Setup mock async client
mock_client = MagicMock()
mock_response = MagicMock()
mock_response.status_code = 200
mock_response.headers = {"content-type": "application/json"}
mock_response.json.return_value = {
"id": "chatcmpl-test",
"object": "chat.completion",
"created": 1234567890,
"model": "test-model",
"choices": [
{
"index": 0,
"message": {
"role": "assistant",
"content": "Test response",
},
"finish_reason": "stop",
}
],
"usage": {
"prompt_tokens": 10,
"completion_tokens": 5,
"total_tokens": 15,
},
}
mock_response.text = '{"id": "chatcmpl-test", "object": "chat.completion", "created": 1234567890, "model": "test-model", "choices": [{"index": 0, "message": {"role": "assistant", "content": "Test response"}, "finish_reason": "stop"}], "usage": {"prompt_tokens": 10, "completion_tokens": 5, "total_tokens": 15}}'
async def mock_post(*args, **kwargs):
return mock_response
mock_client.post = mock_post
mock_get_async_httpx_client.return_value = mock_client
try:
await litellm.acompletion(
model="hosted_vllm/test-model",
messages=[{"role": "user", "content": "Hello"}],
api_base="https://test-vllm.example.com/v1",
ssl_verify=False,
)
except Exception:
# Even if the response parsing fails, we just need to verify
# that the mock was called with the correct ssl_verify parameter
pass
# Verify get_async_httpx_client was called with ssl_verify=False
mock_get_async_httpx_client.assert_called()
call_kwargs = mock_get_async_httpx_client.call_args[1]
# Check that params contains ssl_verify=False
params = call_kwargs.get("params", {})
assert (
params.get("ssl_verify") is False
), f"Expected ssl_verify=False in params, got {params}"
if __name__ == "__main__":
pytest.main([__file__, "-v", "-s"])

135
tests/test_litellm/llms/hosted_vllm/embedding/test_hosted_vllm_embedding_ssl_verify.py
View file

@ -1,135 +0,0 @@
"""
Test SSL verification for hosted_vllm provider embeddings.
This test ensures that the ssl_verify parameter is properly passed through
to the HTTP client when using the hosted_vllm provider for embeddings.
Issue: ssl_verify parameter was being ignored because hosted_vllm fell through
to the openai_like catch-all path in main.py, which doesn't pass ssl_verify to the HTTP client.
"""
from unittest.mock import MagicMock, patch
import pytest
import litellm
class TestHostedVLLMEmbeddingSSLVerify:
"""Test suite for SSL verification in hosted_vllm provider embeddings."""
@patch("litellm.llms.custom_httpx.llm_http_handler._get_httpx_client")
def test_hosted_vllm_embedding_ssl_verify_false_sync(self, mock_get_httpx_client):
"""Test that ssl_verify=False is passed to the HTTP client for sync embedding calls."""
# Setup mock client
mock_client = MagicMock()
mock_response = MagicMock()
mock_response.status_code = 200
mock_response.headers = {"content-type": "application/json"}
mock_response.json.return_value = {
"object": "list",
"data": [
{
"object": "embedding",
"index": 0,
"embedding": [0.1, 0.2, 0.3, 0.4, 0.5],
}
],
"model": "text-embedding-model",
"usage": {
"prompt_tokens": 5,
"total_tokens": 5,
},
}
mock_response.text = '{"object": "list", "data": [{"object": "embedding", "index": 0, "embedding": [0.1, 0.2, 0.3, 0.4, 0.5]}], "model": "text-embedding-model", "usage": {"prompt_tokens": 5, "total_tokens": 5}}'
mock_client.post.return_value = mock_response
mock_get_httpx_client.return_value = mock_client
try:
litellm.embedding(
model="hosted_vllm/text-embedding-model",
input=["hello world"],
api_base="https://test-vllm.example.com/v1",
ssl_verify=False,
)
except Exception:
# Even if the response parsing fails, we just need to verify
# that the mock was called with the correct ssl_verify parameter
pass
# Verify _get_httpx_client was called with ssl_verify=False
mock_get_httpx_client.assert_called()
call_args = mock_get_httpx_client.call_args
# Check that params contains ssl_verify=False
if call_args[0]:
# Positional argument
params = call_args[0][0]
else:
# Keyword argument
params = call_args[1].get("params", {})
assert (
params.get("ssl_verify") is False
), f"Expected ssl_verify=False in params, got {params}"
@patch("litellm.llms.custom_httpx.llm_http_handler.get_async_httpx_client")
@pytest.mark.asyncio
async def test_hosted_vllm_embedding_ssl_verify_false_async(
self, mock_get_async_httpx_client
):
"""Test that ssl_verify=False is passed to the HTTP client for async embedding calls."""
# Setup mock async client
mock_client = MagicMock()
mock_response = MagicMock()
mock_response.status_code = 200
mock_response.headers = {"content-type": "application/json"}
mock_response.json.return_value = {
"object": "list",
"data": [
{
"object": "embedding",
"index": 0,
"embedding": [0.1, 0.2, 0.3, 0.4, 0.5],
}
],
"model": "text-embedding-model",
"usage": {
"prompt_tokens": 5,
"total_tokens": 5,
},
}
mock_response.text = '{"object": "list", "data": [{"object": "embedding", "index": 0, "embedding": [0.1, 0.2, 0.3, 0.4, 0.5]}], "model": "text-embedding-model", "usage": {"prompt_tokens": 5, "total_tokens": 5}}'
async def mock_post(*args, **kwargs):
return mock_response
mock_client.post = mock_post
mock_get_async_httpx_client.return_value = mock_client
try:
await litellm.aembedding(
model="hosted_vllm/text-embedding-model",
input=["hello world"],
api_base="https://test-vllm.example.com/v1",
ssl_verify=False,
)
except Exception:
# Even if the response parsing fails, we just need to verify
# that the mock was called with the correct ssl_verify parameter
pass
# Verify get_async_httpx_client was called with ssl_verify=False
mock_get_async_httpx_client.assert_called()
call_kwargs = mock_get_async_httpx_client.call_args[1]
# Check that params contains ssl_verify=False
params = call_kwargs.get("params", {})
assert (
params.get("ssl_verify") is False
), f"Expected ssl_verify=False in params, got {params}"
if __name__ == "__main__":
pytest.main([__file__, "-v", "-s"])

31
tests/test_litellm/test_budget_ratchet_check.py
View file

@ -9,6 +9,7 @@ import importlib.util
import subprocess
import sys
from pathlib import Path
from typing import Final
_MODULE_PATH = (
Path(__file__).resolve().parents[2] / "scripts" / "budget_ratchet_check.py"
@ -92,6 +93,36 @@ def test_graduation_never_excuses_a_raised_limit():
assert "0 -> 7" in regs[0].detail
def test_dropped_rule_the_checker_retired_is_clean():
base: Final = {"TQ008": _spec_of(10993)}
assert ratchet.regressions_for("b.json", base, {}, retired=frozenset({"TQ008"})) == []
def test_dropped_rule_the_checker_still_emits_is_a_regression():
base: Final = {"TQ001": _spec_of(5), "TQ008": _spec_of(10993)}
regs: Final = ratchet.regressions_for("b.json", base, {}, retired=frozenset({"TQ008"}))
assert [r.rule for r in regs] == ["TQ001"]
assert "dropped" in regs[0].detail
def test_retirement_never_excuses_a_raised_limit():
base: Final = {"TQ008": _spec_of(0)}
regs: Final = ratchet.regressions_for("b.json", base, {"TQ008": _spec_of(7)}, retired=frozenset({"TQ008"}))
assert [r.rule for r in regs] == ["TQ008"]
assert "0 -> 7" in regs[0].detail
def test_retired_rules_come_from_the_paired_checker():
base: Final = {"TQ001": _spec_of(5), "TQ008": _spec_of(10993)}
assert ratchet.retired_rules("test-quality-budget.json", base) == frozenset({"TQ008"})
def test_budgets_without_a_paired_checker_never_retire():
base: Final = {"TQ008": _spec_of(1)}
for rel in ("ruff-strict-budget.json", "type-discipline-budget.json", "basedpyright-code-budget.json"):
assert ratchet.retired_rules(rel, base) == frozenset()
def test_graduated_selectors_come_from_the_paired_ruff_config():
selectors = ratchet.graduated_selectors("ruff-strict-budget.json")
assert "UP006" in selectors

40
tests/test_litellm/test_check_test_quality.py
View file

@ -12,6 +12,8 @@ import os
import subprocess
import sys
from pathlib import Path
from types import MappingProxyType
from typing import Final
import pytest
@ -612,6 +614,44 @@ def test_a_fanned_out_run_reports_each_generated_file_exactly_once(tmp_path):
assert all(" TQ001 " in line for line in reported)
_VIOLATING_SNIPPETS: Final = MappingProxyType(
{
"TQ000": ("test_snippet.py", "def test_broken(:\n pass\n"),
"TQ001": ("test_snippet.py", "def test_nothing():\n compute()\n"),
"TQ002": (
"test_snippet.py",
"from unittest.mock import patch\n"
"\n"
"\n"
"def test_echo():\n"
" with patch('litellm.completion') as mock_completion:\n"
" run()\n"
" mock_completion.assert_called_once()\n",
),
"TQ003": ("test_snippet.py", "import sys\n\nsys.path.insert(0, '..')\n"),
"TQ004": ("test_snippet.py", "import os\n\nos.environ['KEY'] = 'v'\n"),
"TQ005": ("test_snippet.py", "import litellm\n\nlitellm.drop_params = True\n"),
"TQ006": ("test_snippet.py", _DIRECT_GATE),
"TQ007": ("conftest.py", _SNAPSHOT_CONFTEST),
"TQ009": (
"test_snippet.py",
'import subprocess, sys\nsubprocess.run([sys.executable, "-c", "pass"])\n',
),
}
)
def test_rule_codes_match_every_code_the_checker_emits(tmp_path):
emitted: Final = frozenset(
v.code
for name, source in _VIOLATING_SNIPPETS.values()
for v in checker.check_file(_written(tmp_path, source, name))
)
for code, (name, source) in _VIOLATING_SNIPPETS.items():
assert code in [v.code for v in checker.check_file(_written(tmp_path, source, name))], code
assert emitted == checker.RULE_CODES
def test_sys_executable_child_without_isolation_flag_is_flagged(tmp_path):
source = 'import subprocess, sys\nsubprocess.run([sys.executable, "-c", "pass"])\n'
assert _codes(tmp_path, source) == ["TQ009"]

0
tests/test_litellm/llms/nvidia_riva/audio_transcription/__init__.py → tests/unit/__init__.py
View file

14
tests/unit/conftest.py
View file

@ -1,9 +1,11 @@
from collections.abc import Iterator
import os
from typing import Final
import pytest
from pytest_socket import enable_socket, socket_allow_hosts
os.environ["LITELLM_LOCAL_MODEL_COST_MAP"] = "True"
LOOPBACK_HOSTS: Final = ["127.0.0.1", "::1"]
@ -11,13 +13,13 @@ def _allow_loopback_only() -> None:
socket_allow_hosts(LOOPBACK_HOSTS, allow_unix_socket=True)
@pytest.fixture(autouse=True, scope="session")
def block_external_sockets() -> Iterator[None]:
_allow_loopback_only()
yield
enable_socket()
_allow_loopback_only()
@pytest.hookimpl(trylast=True)
def pytest_runtest_setup() -> None:
_allow_loopback_only()
def pytest_sessionfinish() -> None:
enable_socket()

0
tests/test_litellm/llms/oci/embed/__init__.py → tests/unit/integrations/levo/__init__.py
View file

42
tests/test_litellm/integrations/levo/test_levo.py → tests/unit/integrations/levo/test_levo.py
View file

@ -151,48 +151,6 @@ class TestLevoConfig(unittest.TestCase):
class TestLevoIntegration(unittest.TestCase):
"""Integration tests for LevoLogger."""
@patch.dict(
"os.environ",
{
"LEVOAI_API_KEY": "test-api-key",
"LEVOAI_ORG_ID": "test-org-id",
"LEVOAI_WORKSPACE_ID": "test-workspace-id",
"LEVOAI_COLLECTOR_URL": "https://collector.levo.ai",
},
)
@pytest.mark.skipif(
not OPENTELEMETRY_AVAILABLE, reason="OpenTelemetry packages not installed"
)
@patch(
"litellm.integrations.opentelemetry.OpenTelemetry._init_otel_logger_on_litellm_proxy"
)
@pytest.mark.asyncio
async def test_levo_logger_health_check_healthy(self, mock_init_proxy):
"""Test health check returns healthy status when config is valid."""
# Mock the proxy initialization to avoid importing proxy code
mock_init_proxy.return_value = None
config = LevoLogger.get_levo_config()
otel_config = OpenTelemetryConfig(
exporter=config.protocol,
endpoint=config.endpoint,
headers=config.otlp_auth_headers,
)
# Create tracer provider with in-memory exporter
tracer_provider = TracerProvider()
tracer_provider.add_span_processor(SimpleSpanProcessor(InMemorySpanExporter()))
levo_logger = LevoLogger(
config=otel_config, callback_name="levo", tracer_provider=tracer_provider
)
# Run health check
result = await levo_logger.async_health_check()
self.assertEqual(result["status"], "healthy")
self.assertIn("message", result)
@patch.dict("os.environ", {}, clear=True)
def test_levo_logger_health_check_unhealthy(self):
"""Test health check returns unhealthy status when required vars are missing."""

0
tests/test_litellm/llms/ocr/guardrail_translation/__init__.py → tests/unit/integrations/litellm_agent/__init__.py
View file

0
tests/test_litellm/integrations/litellm_agent/test_litellm_agent_model_resolver.py → tests/unit/integrations/litellm_agent/test_litellm_agent_model_resolver.py
View file

0
tests/test_litellm/llms/openai/chat/__init__.py → tests/unit/integrations/mavvrik_focus/__init__.py
View file

0
tests/test_litellm/integrations/mavvrik_focus/test_mavvrik_focus_logger.py → tests/unit/integrations/mavvrik_focus/test_mavvrik_focus_logger.py
View file

0
tests/test_litellm/llms/openai/chat/guardrail_translation/__init__.py → tests/unit/integrations/opik/__init__.py
View file

0
tests/test_litellm/integrations/opik/test_opik_extractors.py → tests/unit/integrations/opik/test_opik_extractors.py
View file

0
tests/unit/integrations/pointfive/__init__.py Normal file
View file

0
tests/test_litellm/integrations/pointfive/test_logger.py → tests/unit/integrations/pointfive/test_logger.py
View file

0
tests/test_litellm/integrations/pointfive/test_payload.py → tests/unit/integrations/pointfive/test_payload.py
View file

0
tests/test_litellm/integrations/pointfive/test_upload_client.py → tests/unit/integrations/pointfive/test_upload_client.py
View file

0
tests/unit/integrations/vector_store_integrations/__init__.py Normal file
View file

0
tests/test_litellm/integrations/vector_store_integrations/test_vector_store_pre_call_hook.py → tests/unit/integrations/vector_store_integrations/test_vector_store_pre_call_hook.py
View file

0
tests/unit/litellm_core_utils/__init__.py Normal file
View file

0
tests/unit/litellm_core_utils/audio_utils/__init__.py Normal file
View file

0
tests/test_litellm/litellm_core_utils/audio_utils/test_subtitle_utils.py → tests/unit/litellm_core_utils/audio_utils/test_subtitle_utils.py
View file

0
tests/unit/litellm_core_utils/llm_response_utils/__init__.py Normal file
View file

0
tests/test_litellm/litellm_core_utils/llm_response_utils/test_convert_dict_to_response.py → tests/unit/litellm_core_utils/llm_response_utils/test_convert_dict_to_response.py
View file

0
tests/test_litellm/litellm_core_utils/llm_response_utils/test_convert_to_streaming_response.py → tests/unit/litellm_core_utils/llm_response_utils/test_convert_to_streaming_response.py
View file

0
tests/test_litellm/litellm_core_utils/llm_response_utils/test_get_formatted_prompt.py → tests/unit/litellm_core_utils/llm_response_utils/test_get_formatted_prompt.py
View file

0
tests/test_litellm/litellm_core_utils/llm_response_utils/test_response_metadata.py → tests/unit/litellm_core_utils/llm_response_utils/test_response_metadata.py
View file

0
tests/unit/llms/__init__.py Normal file
View file

0
tests/unit/llms/a2a/__init__.py Normal file
View file

0
tests/unit/llms/a2a/chat/__init__.py Normal file
View file

0
tests/unit/llms/a2a/chat/guardrail_translation/__init__.py Normal file
View file

0
tests/test_litellm/llms/a2a/chat/guardrail_translation/test_a2a_guardrail_handler.py → tests/unit/llms/a2a/chat/guardrail_translation/test_a2a_guardrail_handler.py
View file

0
tests/test_litellm/llms/a2a/chat/test_a2a_chat_streaming_iterator.py → tests/unit/llms/a2a/chat/test_a2a_chat_streaming_iterator.py
View file

0
tests/test_litellm/llms/a2a/chat/test_a2a_chat_transformation.py → tests/unit/llms/a2a/chat/test_a2a_chat_transformation.py
View file

0
tests/test_litellm/llms/a2a/test_common_utils.py → tests/unit/llms/a2a/test_common_utils.py
View file

0
tests/unit/llms/anthropic/__init__.py Normal file
View file

0
tests/unit/llms/anthropic/batches/__init__.py Normal file
View file

0
tests/test_litellm/llms/anthropic/batches/test_handler.py → tests/unit/llms/anthropic/batches/test_handler.py
View file

0
tests/test_litellm/llms/anthropic/batches/test_transformation.py → tests/unit/llms/anthropic/batches/test_transformation.py
View file

0
tests/unit/llms/base_llm/__init__.py Normal file
View file

0
tests/unit/llms/base_llm/batches/__init__.py Normal file
View file

16
tests/test_litellm/llms/base_llm/batches/test_transformation.py → tests/unit/llms/base_llm/batches/test_transformation.py
View file

@ -129,22 +129,6 @@ def test_subclass_missing_any_abstract_member_cannot_instantiate(missing_member)
Incomplete()
def test_concrete_instance_methods_run():
"""Sanity: the trivial overrides actually execute through the base contract."""
instance = _ConcreteBatchesConfig()
assert instance.custom_llm_provider == LlmProviders.OPENAI
assert instance.validate_environment(
headers={"x": "1"},
model="m",
messages=[],
optional_params={},
litellm_params={},
) == {"x": "1"}
assert instance.transform_retrieve_batch_request(
batch_id="b-1", optional_params={}, litellm_params={}
) == {"batch_id": "b-1"}
# =========================================================================== #
# get_config()
# =========================================================================== #

0
tests/unit/llms/base_llm/realtime/__init__.py Normal file
View file

0
tests/test_litellm/llms/base_llm/realtime/test_transcription_protocol.py → tests/unit/llms/base_llm/realtime/test_transcription_protocol.py
View file

0
tests/unit/llms/baseten/__init__.py Normal file
View file

0
tests/unit/llms/baseten/chat/__init__.py Normal file
View file

0
tests/test_litellm/llms/baseten/chat/test_baseten_completions.py → tests/unit/llms/baseten/chat/test_baseten_completions.py
View file

0
tests/unit/llms/bedrock/__init__.py Normal file
View file

0
tests/unit/llms/bedrock/chat/__init__.py Normal file
View file

0
tests/unit/llms/bedrock/chat/agentcore/__init__.py Normal file
View file

0
tests/test_litellm/llms/bedrock/chat/agentcore/test_agentcore_transformation.py → tests/unit/llms/bedrock/chat/agentcore/test_agentcore_transformation.py
View file

0
tests/unit/llms/bedrock/chat/invoke_transformations/__init__.py Normal file
View file

0
tests/test_litellm/llms/bedrock/chat/invoke_transformations/test_amazon_moonshot_transformation.py → tests/unit/llms/bedrock/chat/invoke_transformations/test_amazon_moonshot_transformation.py
View file

22
tests/test_litellm/llms/bedrock/chat/invoke_transformations/test_amazon_nova_transformation.py → tests/unit/llms/bedrock/chat/invoke_transformations/test_amazon_nova_transformation.py
View file

@ -1,5 +1,8 @@
import json
import pytest
import litellm
from litellm.llms.bedrock.chat.invoke_transformations.amazon_nova_transformation import (
AmazonInvokeNovaConfig,
)
@ -13,6 +16,25 @@ TOOL_CALL = {"id": "call_1", "type": "function", "function": {"name": "f", "argu
PNG_DATA_URL = "data:image/png;base64,iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJAAAADUlEQVR42mNkYPhfDwAChwGA60e6kgAAAABJRU5ErkJggg=="
@pytest.fixture
def local_model_cost_map(monkeypatch):
"""Force the bundled in-repo cost map so capability and pricing assertions do not
depend on the network-fetched ``main`` copy, which lags this branch until merge.
``get_model_info`` is lru_cached, so swapping ``model_cost`` is not enough on its
own; clear on the way in and out so entries warmed against either map never leak
across tests."""
original_model_cost = litellm.model_cost
monkeypatch.setenv("LITELLM_LOCAL_MODEL_COST_MAP", "True")
litellm.model_cost = litellm.get_model_cost_map(url="")
litellm.get_model_info.cache_clear()
try:
yield
finally:
litellm.model_cost = original_model_cost
litellm.get_model_info.cache_clear()
def _transform_request(messages, optional_params, litellm_params=None):
return AmazonInvokeNovaConfig().transform_request(
model=MODEL,

0
tests/test_litellm/llms/bedrock/chat/invoke_transformations/test_amazon_qwen2_transformation.py → tests/unit/llms/bedrock/chat/invoke_transformations/test_amazon_qwen2_transformation.py
View file

0
tests/test_litellm/llms/bedrock/chat/invoke_transformations/test_amazon_qwen3_transformation.py → tests/unit/llms/bedrock/chat/invoke_transformations/test_amazon_qwen3_transformation.py
View file

0
tests/test_litellm/llms/bedrock/chat/invoke_transformations/test_base_invoke_transformation.py → tests/unit/llms/bedrock/chat/invoke_transformations/test_base_invoke_transformation.py
View file

73
tests/test_litellm/llms/bedrock/chat/invoke_transformations/test_bedrock_chat_invoke_transformations_anthropic_claude3_transformation.py → tests/unit/llms/bedrock/chat/invoke_transformations/test_bedrock_chat_invoke_transformations_anthropic_claude3_transformation.py
View file

@ -1,6 +1,8 @@
import asyncio
import base64
import json
import uuid
from types import SimpleNamespace
from typing import Final
from unittest.mock import patch
@ -17,6 +19,77 @@ from litellm.llms.bedrock.chat.invoke_transformations.anthropic_claude3_transfor
from litellm.llms.custom_httpx.http_handler import AsyncHTTPHandler
ONE_PIXEL_PNG = base64.b64decode(
"iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJAAAADUlEQVR42mNkYPhfDwAChwGA60e6kgAAAABJRU5ErkJggg=="
)
@pytest.fixture
def async_only_image_fetch(monkeypatch):
from litellm.litellm_core_utils.prompt_templates import factory, image_handling
from litellm.llms.gemini.chat import transformation as gemini_chat_transformation
fetch = SimpleNamespace(
fetched=[],
base64_png=base64.b64encode(ONE_PIXEL_PNG).decode(),
data_url="data:image/png;base64," + base64.b64encode(ONE_PIXEL_PNG).decode(),
)
def forbid_sync_fetch(client, url, **kwargs):
raise litellm.ImageFetchError(f"sync image fetch ran on the event loop: {url}")
async def serve_png(client, url, **kwargs):
fetch.fetched.append(url)
return httpx.Response(
200,
content=ONE_PIXEL_PNG,
headers={"content-type": "image/png"},
request=httpx.Request("GET", url),
)
def forbid_sync_convert(url, *args, **kwargs):
if url.startswith(("http://", "https://")):
raise litellm.ImageFetchError(f"sync convert_url_to_base64 ran on the request path: {url}")
return url
monkeypatch.setattr(image_handling, "safe_get", forbid_sync_fetch)
monkeypatch.setattr(image_handling, "async_safe_get", serve_png)
for module in (image_handling, factory, gemini_chat_transformation):
monkeypatch.setattr(module, "convert_url_to_base64", forbid_sync_convert)
return fetch
@pytest.fixture
def local_model_cost_map(monkeypatch):
"""Force the bundled in-repo cost map so capability and pricing assertions do not
depend on the network-fetched ``main`` copy, which lags this branch until merge.
``get_model_info`` is lru_cached, so swapping ``model_cost`` is not enough on its
own; clear on the way in and out so entries warmed against either map never leak
across tests."""
original_model_cost = litellm.model_cost
monkeypatch.setenv("LITELLM_LOCAL_MODEL_COST_MAP", "True")
litellm.model_cost = litellm.get_model_cost_map(url="")
litellm.get_model_info.cache_clear()
try:
yield
finally:
litellm.model_cost = original_model_cost
litellm.get_model_info.cache_clear()
@pytest.fixture
def local_beta_headers_config(monkeypatch):
"""Pin the bundled ``anthropic_beta_headers_config.json`` so beta header assertions
do not depend on the network-fetched copy or on what earlier tests left cached."""
from litellm.anthropic_beta_headers_manager import reload_beta_headers_config
monkeypatch.setenv("LITELLM_LOCAL_ANTHROPIC_BETA_HEADERS", "True")
reload_beta_headers_config()
yield
reload_beta_headers_config()
def test_get_supported_params_thinking():
config = AmazonAnthropicClaudeConfig()
params = config.get_supported_openai_params(

0
tests/test_litellm/llms/bedrock/chat/invoke_transformations/test_twelvelabs_pegasus_transformation.py → tests/unit/llms/bedrock/chat/invoke_transformations/test_twelvelabs_pegasus_transformation.py
View file

0
tests/unit/llms/bedrock/chat/mantle/__init__.py Normal file
View file

43
tests/test_litellm/llms/bedrock/chat/mantle/test_bedrock_chat_mantle_transformation.py → tests/unit/llms/bedrock/chat/mantle/test_bedrock_chat_mantle_transformation.py
View file

@ -1,12 +1,55 @@
import base64
import json
import uuid
from types import SimpleNamespace
import httpx
import pytest
import litellm
from litellm.llms.custom_httpx.http_handler import AsyncHTTPHandler
ONE_PIXEL_PNG = base64.b64decode(
"iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJAAAADUlEQVR42mNkYPhfDwAChwGA60e6kgAAAABJRU5ErkJggg=="
)
@pytest.fixture
def async_only_image_fetch(monkeypatch):
from litellm.litellm_core_utils.prompt_templates import factory, image_handling
from litellm.llms.gemini.chat import transformation as gemini_chat_transformation
fetch = SimpleNamespace(
fetched=[],
base64_png=base64.b64encode(ONE_PIXEL_PNG).decode(),
data_url="data:image/png;base64," + base64.b64encode(ONE_PIXEL_PNG).decode(),
)
def forbid_sync_fetch(client, url, **kwargs):
raise litellm.ImageFetchError(f"sync image fetch ran on the event loop: {url}")
async def serve_png(client, url, **kwargs):
fetch.fetched.append(url)
return httpx.Response(
200,
content=ONE_PIXEL_PNG,
headers={"content-type": "image/png"},
request=httpx.Request("GET", url),
)
def forbid_sync_convert(url, *args, **kwargs):
if url.startswith(("http://", "https://")):
raise litellm.ImageFetchError(f"sync convert_url_to_base64 ran on the request path: {url}")
return url
monkeypatch.setattr(image_handling, "safe_get", forbid_sync_fetch)
monkeypatch.setattr(image_handling, "async_safe_get", serve_png)
for module in (image_handling, factory, gemini_chat_transformation):
monkeypatch.setattr(module, "convert_url_to_base64", forbid_sync_convert)
return fetch
async def test_bedrock_mantle_claude_async_completion_inlines_remote_images_off_the_event_loop(async_only_image_fetch):
image_url = f"http://img.example/{uuid.uuid4()}.png"
captured = {}

0
tests/unit/llms/bedrock/count_tokens/__init__.py Normal file
View file

0
tests/test_litellm/llms/bedrock/count_tokens/test_bedrock_count_tokens_handler.py → tests/unit/llms/bedrock/count_tokens/test_bedrock_count_tokens_handler.py
View file

Some files were not shown because too many files have changed in this diff Show more