fix CI and add monkeypatch test

This commit is contained in:
lioZ129 2026-03-26 15:46:22 +08:00
parent f484b2ca6d
commit d4b8fd4a54
5 changed files with 97 additions and 36 deletions

View file

@ -411,7 +411,7 @@ class Cache:
Returns:
str: The hashed cache key.
"""
hash_object = hashlib.sha256(cache_key.encode())
hash_object = hashlib.sha256(cache_key.encode(), usedforsecurity=False)
# Hexadecimal representation of the hash
hash_hex = hash_object.hexdigest()
verbose_logger.debug("Hashed cache key (SHA-256): %s", hash_hex)

View file

@ -1393,10 +1393,10 @@ def convert_to_gemini_tool_call_invoke(
if tool_calls is not None:
for idx, tool in enumerate(tool_calls):
if "function" in tool:
gemini_function_call: Optional[
VertexFunctionCall
] = _gemini_tool_call_invoke_helper(
function_call_params=tool["function"]
gemini_function_call: Optional[VertexFunctionCall] = (
_gemini_tool_call_invoke_helper(
function_call_params=tool["function"]
)
)
if gemini_function_call is not None:
part_dict: VertexPartType = {
@ -1574,9 +1574,7 @@ def convert_to_gemini_tool_call_result( # noqa: PLR0915
file_data = (
file_content.get("file_data", "")
if isinstance(file_content, dict)
else file_content
if isinstance(file_content, str)
else ""
else file_content if isinstance(file_content, str) else ""
)
if file_data:
@ -2081,9 +2079,9 @@ def _sanitize_empty_text_content(
if isinstance(content, str):
if not content or not content.strip():
message = cast(AllMessageValues, dict(message)) # Make a copy
message[
"content"
] = "[System: Empty message content sanitised to satisfy protocol]"
message["content"] = (
"[System: Empty message content sanitised to satisfy protocol]"
)
verbose_logger.debug(
f"_sanitize_empty_text_content: Replaced empty text content in {message.get('role')} message"
)
@ -2423,9 +2421,9 @@ def anthropic_messages_pt( # noqa: PLR0915
# Convert ChatCompletionImageUrlObject to dict if needed
image_url_value = m["image_url"]
if isinstance(image_url_value, str):
image_url_input: Union[
str, dict[str, Any]
] = image_url_value
image_url_input: Union[str, dict[str, Any]] = (
image_url_value
)
else:
# ChatCompletionImageUrlObject or dict case - convert to dict
image_url_input = {
@ -2452,9 +2450,9 @@ def anthropic_messages_pt( # noqa: PLR0915
)
if "cache_control" in _content_element:
_anthropic_content_element[
"cache_control"
] = _content_element["cache_control"]
_anthropic_content_element["cache_control"] = (
_content_element["cache_control"]
)
user_content.append(_anthropic_content_element)
elif m.get("type", "") == "text":
m = cast(ChatCompletionTextObject, m)
@ -2514,9 +2512,9 @@ def anthropic_messages_pt( # noqa: PLR0915
)
if "cache_control" in _content_element:
_anthropic_content_text_element[
"cache_control"
] = _content_element["cache_control"]
_anthropic_content_text_element["cache_control"] = (
_content_element["cache_control"]
)
user_content.append(_anthropic_content_text_element)
@ -2649,9 +2647,9 @@ def anthropic_messages_pt( # noqa: PLR0915
original_content_element=dict(assistant_content_block),
)
if "cache_control" in _content_element:
_anthropic_text_content_element[
"cache_control"
] = _content_element["cache_control"]
_anthropic_text_content_element["cache_control"] = (
_content_element["cache_control"]
)
text_element = _anthropic_text_content_element
# Interleave: each thinking block precedes its server tool group.
@ -2811,9 +2809,9 @@ def anthropic_messages_pt( # noqa: PLR0915
)
if "cache_control" in _content_element:
_anthropic_text_content_element[
"cache_control"
] = _content_element["cache_control"]
_anthropic_text_content_element["cache_control"] = (
_content_element["cache_control"]
)
assistant_content.append(_anthropic_text_content_element)
@ -3725,7 +3723,7 @@ class BedrockImageProcessor:
sample = normalized[:HASH_SAMPLE_BYTES]
# --- Compute deterministic hash (sample + total length) ---
hasher = hashlib.sha256()
hasher = hashlib.sha256(usedforsecurity=False)
hasher.update(sample)
hasher.update(
str(len(normalized)).encode("utf-8")
@ -5255,9 +5253,7 @@ def default_response_schema_prompt(response_schema: dict) -> str:
prompt_str = """Use this JSON schema:
```json
{}
```""".format(
response_schema
)
```""".format(response_schema)
return prompt_str

View file

@ -29,7 +29,7 @@ _file_cache: Dict[str, str] = {}
def _get_url_hash(url: str) -> str:
"""Generate hash for URL to use as cache key."""
return hashlib.sha256(url.encode()).hexdigest()
return hashlib.sha256(url.encode(), usedforsecurity=False).hexdigest()
def _parse_data_url(data_url: str) -> Optional[Tuple[bytes, str, str]]:

View file

@ -51,7 +51,11 @@ def _build_audit_log_payload(
if request_data.updated_at is not None:
updated_at = request_data.updated_at.isoformat()
table_name_str: str = request_data.table_name.value if isinstance(request_data.table_name, LitellmTableNames) else str(request_data.table_name)
table_name_str: str = (
request_data.table_name.value
if isinstance(request_data.table_name, LitellmTableNames)
else str(request_data.table_name)
)
return StandardAuditLogPayload(
id=request_data.id,
@ -89,7 +93,9 @@ async def _dispatch_audit_log_to_callbacks(
for callback in litellm.audit_log_callbacks:
try:
resolved: Optional[CustomLogger] = callback if isinstance(callback, CustomLogger) else None
resolved: Optional[CustomLogger] = (
callback if isinstance(callback, CustomLogger) else None
)
if isinstance(callback, str):
resolved = _resolve_audit_log_callback(callback)
if resolved is None:
@ -138,9 +144,7 @@ async def create_object_audit_log(
return
_changed_by = (
litellm_changed_by
or user_api_key_dict.user_id
or litellm_proxy_admin_name
litellm_changed_by or user_api_key_dict.user_id or litellm_proxy_admin_name
)
await create_audit_log_for_update(

View file

@ -4,6 +4,7 @@ Unit tests for HPC-AI OpenAI-compatible configuration.
import os
import sys
from typing import Any, Optional
sys.path.insert(0, os.path.abspath("../../../../.."))
@ -15,6 +16,66 @@ from litellm.llms.hpc_ai.chat.transformation import HpcAiConfig
class TestHpcAiConfig:
def test_get_openai_compatible_provider_info_default_base_when_env_unset(
self, monkeypatch: pytest.MonkeyPatch
) -> None:
def fake_get_secret_str(
secret_name: str, default_value: Optional[Any] = None
) -> Optional[str]:
return None
monkeypatch.setattr(
"litellm.llms.hpc_ai.chat.transformation.get_secret_str",
fake_get_secret_str,
)
config = HpcAiConfig()
api_base, api_key = config._get_openai_compatible_provider_info(None, None)
assert api_base == "https://api.hpc-ai.com/inference/v1"
assert api_key is None
def test_get_openai_compatible_provider_info_api_key_from_env(
self, monkeypatch: pytest.MonkeyPatch
) -> None:
def fake_get_secret_str(
secret_name: str, default_value: Optional[Any] = None
) -> Optional[str]:
if secret_name == "HPC_AI_API_KEY":
return "env-hpc-ai-key"
return None
monkeypatch.setattr(
"litellm.llms.hpc_ai.chat.transformation.get_secret_str",
fake_get_secret_str,
)
config = HpcAiConfig()
api_base, api_key = config._get_openai_compatible_provider_info(None, None)
assert api_base == "https://api.hpc-ai.com/inference/v1"
assert api_key == "env-hpc-ai-key"
def test_get_openai_compatible_provider_info_explicit_overrides_env(
self, monkeypatch: pytest.MonkeyPatch
) -> None:
def fake_get_secret_str(
secret_name: str, default_value: Optional[Any] = None
) -> Optional[str]:
if secret_name == "HPC_AI_API_BASE":
return "https://from-env.example/v1"
if secret_name == "HPC_AI_API_KEY":
return "from-env-key"
return None
monkeypatch.setattr(
"litellm.llms.hpc_ai.chat.transformation.get_secret_str",
fake_get_secret_str,
)
config = HpcAiConfig()
api_base, api_key = config._get_openai_compatible_provider_info(
"https://explicit.example/v1",
"explicit-key",
)
assert api_base == "https://explicit.example/v1"
assert api_key == "explicit-key"
def test_validate_environment_sets_auth_header(self):
config = HpcAiConfig()
headers = {}