mirror of
https://github.com/BerriAI/litellm.git
synced 2026-09-27 01:22:18 +00:00
test: stop CI tests from downloading tokenizer files and images (#43257)
* test: load the embedding base image from a committed 100x100 PNG instead of downloading it * test: move the volcengine embedding test into tests/unit * test: check gpt2 and r50k_base tokenizer parity against committed tiktoken reference files * test: check hub tokenizer selection against an in-memory Hugging Face hub * test: serve image URLs from respx in the gemini tool-result and format-param tests * ci: drop the emptied legacy core-utils test path * test: cover the cohere and anthropic tokenizer paths in the hub tokenizer test * test: fetch every format-param image through respx and check its bytes reach the request * test: drop the gpt2 and r50k_base parity tests, which no litellm path uses * test: drop comments that restate assertions in the format-param test
This commit is contained in:
yuneng-jiang
GitHub
parent
2ef3250ec3
commit
2530255624
15 changed files with 239 additions and 306 deletions
2
.github/workflows/test-unit.yml
vendored
2
.github/workflows/test-unit.yml
vendored
|
|
@ -61,7 +61,7 @@ jobs:
|
|||
|
||||
- shard: core-utils
|
||||
artifact-name: core-utils
|
||||
test-path: "tests/test_litellm/litellm_core_utils"
|
||||
test-path: ""
|
||||
unit-flag: core-utils
|
||||
workers: 2
|
||||
reruns: 1
|
||||
|
|
|
|||
|
|
@ -16,15 +16,12 @@ from litellm.utils import (
|
|||
get_optional_params,
|
||||
get_optional_params_embeddings,
|
||||
)
|
||||
import requests
|
||||
import base64
|
||||
from pathlib import Path
|
||||
|
||||
# test_example.py
|
||||
from abc import ABC, abstractmethod
|
||||
|
||||
url = "https://dummyimage.com/100/100/fff&text=Test+image"
|
||||
response = requests.get(url)
|
||||
file_data = response.content
|
||||
file_data = (Path(__file__).parent.parent / "white_100x100.png").read_bytes()
|
||||
|
||||
encoded_file = base64.b64encode(file_data).decode("utf-8")
|
||||
base64_image = f"data:image/png;base64,{encoded_file}"
|
||||
|
|
|
|||
|
|
@ -1,51 +0,0 @@
|
|||
import pytest
|
||||
from litellm import create_pretrained_tokenizer
|
||||
from tests.unit.litellm_core_utils.test_token_counter import token_counter
|
||||
|
||||
|
||||
def test_tokenizers():
|
||||
try:
|
||||
### test the openai, claude, cohere and llama2 tokenizers.
|
||||
### The tokenizer value should be different for all
|
||||
sample_text = "Hellö World, this is my input string! My name is ishaan CTO"
|
||||
|
||||
# openai tokenizer
|
||||
openai_tokens = token_counter(model="gpt-3.5-turbo", text=sample_text)
|
||||
|
||||
# claude tokenizer
|
||||
claude_tokens = token_counter(model="claude-3-5-haiku-20241022", text=sample_text)
|
||||
|
||||
# cohere tokenizer
|
||||
cohere_tokens = token_counter(model="command-nightly", text=sample_text)
|
||||
|
||||
# llama2 tokenizer
|
||||
llama2_tokens = token_counter(model="meta-llama/Llama-2-7b-chat", text=sample_text)
|
||||
|
||||
# llama3 tokenizer (also testing custom tokenizer)
|
||||
llama3_tokens_1 = token_counter(model="meta-llama/llama-3-70b-instruct", text=sample_text)
|
||||
|
||||
try:
|
||||
llama3_tokenizer = create_pretrained_tokenizer("Xenova/llama-3-tokenizer")
|
||||
except Exception as e:
|
||||
pytest.skip(f"custom tokenizer download failed (HF hub unreachable): {e}")
|
||||
llama3_tokens_2 = token_counter(custom_tokenizer=llama3_tokenizer, text=sample_text)
|
||||
|
||||
print(
|
||||
f"openai tokens: {openai_tokens}; claude tokens: {claude_tokens}; cohere tokens: {cohere_tokens}; llama2 tokens: {llama2_tokens}; llama3 tokens: {llama3_tokens_1}"
|
||||
)
|
||||
|
||||
# assert that all token values are different
|
||||
# llama2 may fall back to the tiktoken tokenizer when the HuggingFace
|
||||
# model hub is unreachable (e.g. in CI). In that case the count will
|
||||
# equal the openai count and the differentiation assertion is skipped.
|
||||
if openai_tokens == llama2_tokens:
|
||||
pytest.skip("llama2 fell back to tiktoken (HF hub unreachable); skipping differentiation assertion")
|
||||
assert llama2_tokens != llama3_tokens_1, "Token values are not different."
|
||||
|
||||
assert llama3_tokens_1 == llama3_tokens_2, (
|
||||
"Custom tokenizer is not being used! It has been configured to use the same tokenizer as the built in llama3 tokenizer and the results should be the same."
|
||||
)
|
||||
|
||||
print("test tokenizer: It worked!")
|
||||
except Exception as e:
|
||||
pytest.fail(f"An exception occured: {e}")
|
||||
|
|
@ -1,20 +0,0 @@
|
|||
import pytest
|
||||
|
||||
from tests.unit.litellm_core_utils.test_tokenizer import (
|
||||
UNICODE_TEXTS,
|
||||
assert_openai_encoding_exposes_the_tiktoken_vocabulary_surface,
|
||||
assert_openai_encoding_matches_python,
|
||||
)
|
||||
|
||||
NETWORK_ENCODINGS = ("r50k_base", "gpt2")
|
||||
|
||||
|
||||
@pytest.mark.parametrize("name", NETWORK_ENCODINGS)
|
||||
@pytest.mark.parametrize("text", UNICODE_TEXTS)
|
||||
def test_openai_encoding_matches_python_unicode_and_batches(name: str, text: str) -> None:
|
||||
assert_openai_encoding_matches_python(name, text)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("name", ("gpt2",))
|
||||
def test_openai_encoding_exposes_the_tiktoken_vocabulary_surface(name: str) -> None:
|
||||
assert_openai_encoding_exposes_the_tiktoken_vocabulary_surface(name)
|
||||
|
|
@ -1,54 +0,0 @@
|
|||
import pytest
|
||||
|
||||
from litellm.litellm_core_utils.prompt_templates.factory import (
|
||||
convert_to_gemini_tool_call_result,
|
||||
)
|
||||
from litellm.types.llms.vertex_ai import BlobType
|
||||
|
||||
|
||||
def test_convert_tool_response_with_url_image():
|
||||
"""Test tool response with HTTP URL image (will download and convert)."""
|
||||
# Use a publicly accessible test image URL
|
||||
test_image_url = "https://via.placeholder.com/1x1.png"
|
||||
|
||||
tool_message = {
|
||||
"role": "tool",
|
||||
"tool_call_id": "call_test456",
|
||||
"content": [
|
||||
{"type": "text", "text": '{"url": "https://example.com"}'},
|
||||
{"type": "input_image", "image_url": test_image_url},
|
||||
],
|
||||
}
|
||||
|
||||
last_message_with_tool_calls = {
|
||||
"tool_calls": [
|
||||
{
|
||||
"id": "call_test456",
|
||||
"function": {
|
||||
"name": "type_text_at",
|
||||
"arguments": '{"x": 300, "y": 400, "text": "hello"}',
|
||||
},
|
||||
}
|
||||
]
|
||||
}
|
||||
|
||||
try:
|
||||
result = convert_to_gemini_tool_call_result(tool_message, last_message_with_tool_calls)
|
||||
|
||||
assert isinstance(result, list), "Should return a parts list when media is present"
|
||||
assert len(result) == 1, "Should return one function_response part"
|
||||
result_part = result[0]
|
||||
assert "function_response" in result_part
|
||||
assert "inline_data" not in result_part
|
||||
function_response = result_part["function_response"]
|
||||
assert function_response["name"] == "type_text_at"
|
||||
|
||||
# Check inline_data is nested under functionResponse.parts.
|
||||
assert "parts" in function_response
|
||||
assert len(function_response["parts"]) == 1
|
||||
inline_data: BlobType = function_response["parts"][0]["inline_data"]
|
||||
assert "data" in inline_data
|
||||
assert "mime_type" in inline_data
|
||||
except Exception as e:
|
||||
# Skip test if URL download fails (no internet connection, etc.)
|
||||
pytest.skip(f"Failed to download image from URL: {e}")
|
||||
|
|
@ -1 +0,0 @@
|
|||
# Volcengine tests
|
||||
|
|
@ -1,164 +0,0 @@
|
|||
import json
|
||||
import os
|
||||
|
||||
import pytest
|
||||
|
||||
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
import litellm
|
||||
|
||||
|
||||
async def _async_fake_bedrock_image_details(image_url):
|
||||
return "ZmFrZS1pbWFnZQ==", "image/png"
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def clear_client_cache():
|
||||
"""
|
||||
Clear the HTTP client cache before each test to ensure mocks are used.
|
||||
This prevents cached real clients from being reused across tests.
|
||||
"""
|
||||
cache = getattr(litellm, "in_memory_llm_clients_cache", None)
|
||||
if cache is not None:
|
||||
cache.flush_cache()
|
||||
yield
|
||||
if cache is not None:
|
||||
cache.flush_cache()
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def add_api_keys_to_env(monkeypatch):
|
||||
monkeypatch.setenv("ANTHROPIC_API_KEY", "sk-ant-api03-1234567890")
|
||||
monkeypatch.setenv("OPENAI_API_KEY", "sk-openai-api03-1234567890")
|
||||
monkeypatch.setenv("AWS_ACCESS_KEY_ID", "my-fake-aws-access-key-id")
|
||||
monkeypatch.setenv("AWS_SECRET_ACCESS_KEY", "my-fake-aws-secret-access-key")
|
||||
monkeypatch.setenv("AWS_REGION", "us-east-1")
|
||||
# Keep these transformation tests on the simple access-key path. A leaked
|
||||
# session token or role/web-identity env var pushes Bedrock auth down a
|
||||
# different branch and fails before the mocked HTTP client is exercised.
|
||||
monkeypatch.delenv("AWS_SESSION_TOKEN", raising=False)
|
||||
monkeypatch.delenv("AWS_ROLE_ARN", raising=False)
|
||||
monkeypatch.delenv("AWS_WEB_IDENTITY_TOKEN_FILE", raising=False)
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"model",
|
||||
[
|
||||
"gemini/gemini-1.5-flash",
|
||||
"bedrock/us.anthropic.claude-haiku-4-5-20251001-v1:0",
|
||||
"bedrock/invoke/anthropic.claude-haiku-4-5-20251001-v1:0",
|
||||
"anthropic/claude-3-5-sonnet",
|
||||
],
|
||||
)
|
||||
@pytest.mark.parametrize("sync_mode", [True, False])
|
||||
@pytest.mark.asyncio
|
||||
async def test_url_with_format_param(model, sync_mode, monkeypatch):
|
||||
from litellm import acompletion, completion
|
||||
from litellm.llms.custom_httpx.http_handler import AsyncHTTPHandler, HTTPHandler
|
||||
from litellm.litellm_core_utils.prompt_templates import factory as prompt_factory
|
||||
|
||||
if sync_mode:
|
||||
client = HTTPHandler()
|
||||
else:
|
||||
client = AsyncHTTPHandler()
|
||||
|
||||
# This test is about request shaping, not live image downloads. Stub the
|
||||
# URL->image conversion helpers so suite-level network/client state from
|
||||
# earlier tests cannot prevent the mocked provider client from being hit.
|
||||
fake_base64_image = "data:image/png;base64,ZmFrZS1pbWFnZQ=="
|
||||
monkeypatch.setattr(
|
||||
prompt_factory, "convert_url_to_base64", lambda url: fake_base64_image
|
||||
)
|
||||
monkeypatch.setattr(
|
||||
prompt_factory.BedrockImageProcessor,
|
||||
"get_image_details",
|
||||
staticmethod(lambda image_url: ("ZmFrZS1pbWFnZQ==", "image/png")),
|
||||
)
|
||||
monkeypatch.setattr(
|
||||
prompt_factory.BedrockImageProcessor,
|
||||
"get_image_details_async",
|
||||
staticmethod(_async_fake_bedrock_image_details),
|
||||
)
|
||||
|
||||
args = {
|
||||
"model": model,
|
||||
"messages": [
|
||||
{
|
||||
"role": "user",
|
||||
"content": [
|
||||
{
|
||||
"type": "image_url",
|
||||
"image_url": {
|
||||
"url": "https://awsmp-logos.s3.amazonaws.com/seller-xw5kijmvmzasy/c233c9ade2ccb5491072ae232c814942.png",
|
||||
"format": "image/png",
|
||||
},
|
||||
},
|
||||
{"type": "text", "text": "Describe this image"},
|
||||
],
|
||||
}
|
||||
],
|
||||
}
|
||||
if model.startswith("gemini/"):
|
||||
args["api_key"] = "test-api-key"
|
||||
with patch.object(client, "post", new=MagicMock()) as mock_client:
|
||||
try:
|
||||
if sync_mode:
|
||||
response = completion(**args, client=client)
|
||||
else:
|
||||
response = await acompletion(**args, client=client)
|
||||
print(response)
|
||||
except Exception as e:
|
||||
pass
|
||||
|
||||
mock_client.assert_called()
|
||||
|
||||
print(mock_client.call_args.kwargs)
|
||||
|
||||
if "data" in mock_client.call_args.kwargs:
|
||||
json_str = mock_client.call_args.kwargs["data"]
|
||||
else:
|
||||
json_str = json.dumps(mock_client.call_args.kwargs["json"])
|
||||
|
||||
if isinstance(json_str, bytes):
|
||||
json_str = json_str.decode("utf-8")
|
||||
|
||||
print(f"type of json_str: {type(json_str)}")
|
||||
|
||||
# Bedrock models convert URLs to base64, while direct Anthropic models support URLs
|
||||
# bedrock/invoke models use Anthropic messages API which supports URLs
|
||||
if model.startswith("bedrock/invoke/"):
|
||||
# bedrock/invoke should convert URLs to base64 (doesn't support URL references)
|
||||
# URL should NOT be in the JSON (it should be converted to base64)
|
||||
assert "https://awsmp-logos.s3.amazonaws.com" not in json_str
|
||||
# Should have base64 data in the source (type="base64", not type="url")
|
||||
assert '"type":"base64"' in json_str or '"type": "base64"' in json_str
|
||||
# Should have "data" field containing base64 content
|
||||
assert '"data"' in json_str
|
||||
elif model.startswith("bedrock/"):
|
||||
# Regular Bedrock models should convert URLs to base64 (uses "bytes" field)
|
||||
# URL should NOT be in the JSON (it should be converted to base64)
|
||||
assert "https://awsmp-logos.s3.amazonaws.com" not in json_str
|
||||
# Should have "bytes" field (Bedrock uses "bytes" not "base64" in the field name)
|
||||
assert '"bytes"' in json_str or '"bytes":' in json_str
|
||||
elif model.startswith("anthropic/"):
|
||||
# Direct Anthropic models should pass HTTPS URLs directly (HTTP URLs are converted to base64)
|
||||
# Since we're using HTTPS URL, it should be passed as-is
|
||||
assert "https://awsmp-logos.s3.amazonaws.com" in json_str
|
||||
# For Anthropic, URL references use "url" type, not base64
|
||||
assert '"type":"url"' in json_str or '"type": "url"' in json_str
|
||||
else:
|
||||
# For other models, check format parameter is respected
|
||||
assert "png" in json_str
|
||||
assert "jpeg" not in json_str
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def set_openrouter_api_key():
|
||||
original_api_key = os.environ.get("OPENROUTER_API_KEY")
|
||||
os.environ["OPENROUTER_API_KEY"] = "fake-key-for-testing"
|
||||
yield
|
||||
if original_api_key is not None:
|
||||
os.environ["OPENROUTER_API_KEY"] = original_api_key
|
||||
else:
|
||||
del os.environ["OPENROUTER_API_KEY"]
|
||||
|
|
@ -3,16 +3,22 @@
|
|||
import asyncio
|
||||
import base64
|
||||
import importlib
|
||||
import json
|
||||
import os
|
||||
import subprocess
|
||||
import sys
|
||||
import threading
|
||||
import time
|
||||
import traceback
|
||||
from concurrent.futures import Future, wait
|
||||
from pathlib import Path
|
||||
from typing import Final
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
import anyio.to_thread
|
||||
import pytest
|
||||
import tiktoken
|
||||
from tokenizers import Regex, Tokenizer, models, pre_tokenizers
|
||||
|
||||
from unittest.mock import AsyncMock, patch
|
||||
|
||||
|
|
@ -1439,3 +1445,89 @@ def test_high_detail_image_token_upper_bound_covers_every_image_size(width: int,
|
|||
def test_high_detail_image_token_upper_bound_is_reached_by_the_largest_high_res_image() -> None:
|
||||
assert calculate_img_tokens(_png_data_url(2000, 768), mode="high") == high_detail_image_token_upper_bound()
|
||||
assert calculate_img_tokens(_png_data_url(1, 1), mode="high") < high_detail_image_token_upper_bound()
|
||||
|
||||
|
||||
HUB_TOKENIZER_SCRIPT: Final = """
|
||||
import json
|
||||
import sys
|
||||
sys.path.insert(0, sys.argv[1])
|
||||
import httpx
|
||||
import huggingface_hub
|
||||
import litellm
|
||||
served = json.loads(sys.argv[2])
|
||||
text = sys.argv[3]
|
||||
requested = []
|
||||
def handle(request):
|
||||
repo = request.url.path.lstrip("/").split("/resolve/")[0]
|
||||
if repo not in served or not request.url.path.endswith("/tokenizer.json"):
|
||||
return httpx.Response(404)
|
||||
requested.append(repo)
|
||||
payload = served[repo].encode()
|
||||
headers = {"content-length": str(len(payload)), "etag": '"fixture"', "x-repo-commit": "a" * 40}
|
||||
return httpx.Response(200, headers=headers, content=payload if request.method == "GET" else b"")
|
||||
huggingface_hub.set_client_factory(lambda: httpx.Client(transport=httpx.MockTransport(handle)))
|
||||
litellm.cohere_models = {"command-r-v1"}
|
||||
litellm.anthropic_models = {"claude-2"}
|
||||
custom = litellm.create_pretrained_tokenizer("Xenova/llama-3-tokenizer")
|
||||
print(json.dumps({
|
||||
"llama2": litellm.token_counter(model="meta-llama/Llama-2-7b-chat", text=text),
|
||||
"llama3": litellm.token_counter(model="meta-llama/llama-3-70b-instruct", text=text),
|
||||
"cohere": litellm.token_counter(model="command-r-v1", text=text),
|
||||
"anthropic": litellm.token_counter(model="claude-2", text=text),
|
||||
"custom": litellm.token_counter(custom_tokenizer=custom, text=text),
|
||||
"requested": sorted(set(requested)),
|
||||
}))
|
||||
"""
|
||||
|
||||
|
||||
def _word_level_tokenizer_json(pre_tokenizer: pre_tokenizers.PreTokenizer) -> str:
|
||||
tokenizer: Final = Tokenizer(models.WordLevel(vocab={"[UNK]": 0}, unk_token="[UNK]"))
|
||||
tokenizer.pre_tokenizer = pre_tokenizer
|
||||
return tokenizer.to_str()
|
||||
|
||||
|
||||
def test_token_counter_uses_the_tokenizer_of_each_model_family_and_of_a_custom_tokenizer(tmp_path: Path) -> None:
|
||||
sample: Final = "Tokenizers disagree: anthropic, tiktoken; llama-2 & llama-3!"
|
||||
served: Final = {
|
||||
"hf-internal-testing/llama-tokenizer": _word_level_tokenizer_json(pre_tokenizers.WhitespaceSplit()),
|
||||
"Xenova/llama-3-tokenizer": _word_level_tokenizer_json(pre_tokenizers.Split(Regex("."), "isolated")),
|
||||
"Xenova/c4ai-command-r-v01-tokenizer": _word_level_tokenizer_json(pre_tokenizers.Whitespace()),
|
||||
}
|
||||
expected: Final = {repo: len(Tokenizer.from_str(payload).encode(sample).ids) for repo, payload in served.items()}
|
||||
anthropic_count: Final = len(Tokenizer.from_str(claude_json_str).encode(sample).ids)
|
||||
tiktoken_count: Final = litellm.token_counter(model="gpt-3.5-turbo", text=sample)
|
||||
assert len({*expected.values(), anthropic_count, tiktoken_count}) == len(expected) + 2
|
||||
|
||||
result: Final = subprocess.run(
|
||||
[
|
||||
sys.executable,
|
||||
"-I",
|
||||
"-c",
|
||||
HUB_TOKENIZER_SCRIPT,
|
||||
str(Path(litellm.__file__).parent.parent),
|
||||
json.dumps(served),
|
||||
sample,
|
||||
],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
timeout=60,
|
||||
env={
|
||||
**os.environ,
|
||||
"HF_HOME": str(tmp_path / "home"),
|
||||
"HF_HUB_CACHE": str(tmp_path / "cache"),
|
||||
"HF_ENDPOINT": "http://127.0.0.1:9",
|
||||
"HF_HUB_OFFLINE": "0",
|
||||
"LITELLM_LOCAL_MODEL_COST_MAP": "True",
|
||||
},
|
||||
)
|
||||
|
||||
assert result.returncode == 0, result.stdout + result.stderr
|
||||
counts: Final = json.loads(result.stdout.strip().splitlines()[-1])
|
||||
assert counts == {
|
||||
"llama2": expected["hf-internal-testing/llama-tokenizer"],
|
||||
"llama3": expected["Xenova/llama-3-tokenizer"],
|
||||
"cohere": expected["Xenova/c4ai-command-r-v01-tokenizer"],
|
||||
"anthropic": anthropic_count,
|
||||
"custom": expected["Xenova/llama-3-tokenizer"],
|
||||
"requested": sorted(served),
|
||||
}
|
||||
|
|
|
|||
|
|
@ -17,17 +17,13 @@ from litellm.utils import claude_json_str
|
|||
from tests.unit.litellm_core_utils.test_decode_special_tokens import TOKENIZER_JSON
|
||||
|
||||
|
||||
OFFLINE_ENCODINGS: Final = ("cl100k_base", "o200k_base", "p50k_base", "p50k_edit", "o200k_harmony")
|
||||
ENCODINGS: Final = ("cl100k_base", "o200k_base", "p50k_base", "p50k_edit", "o200k_harmony")
|
||||
UNICODE_TEXTS: Final = ("hello world", "café 漢字 🙂", "", "a\ud800b", "\ud83d\ude42", "🙂\ud83d\ude42\udfff", " " * 64)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("name", OFFLINE_ENCODINGS)
|
||||
@pytest.mark.parametrize("name", ENCODINGS)
|
||||
@pytest.mark.parametrize("text", UNICODE_TEXTS)
|
||||
def test_openai_encoding_matches_python_unicode_and_batches(name: str, text: str) -> None:
|
||||
assert_openai_encoding_matches_python(name, text)
|
||||
|
||||
|
||||
def assert_openai_encoding_matches_python(name: str, text: str) -> None:
|
||||
reference: Final = tiktoken.get_encoding(name)
|
||||
encoding: Final = OpenAIEncoding.from_tiktoken(name)
|
||||
expected: Final = reference.encode(text)
|
||||
|
|
@ -309,10 +305,6 @@ def test_huggingface_batch_sequence_containers_match_python(is_pretokenized: boo
|
|||
|
||||
@pytest.mark.parametrize("name", ("cl100k_base", "o200k_base", "p50k_edit"))
|
||||
def test_openai_encoding_exposes_the_tiktoken_vocabulary_surface(name: str) -> None:
|
||||
assert_openai_encoding_exposes_the_tiktoken_vocabulary_surface(name)
|
||||
|
||||
|
||||
def assert_openai_encoding_exposes_the_tiktoken_vocabulary_surface(name: str) -> None:
|
||||
reference: Final = tiktoken.get_encoding(name)
|
||||
encoding: Final = OpenAIEncoding.from_tiktoken(name)
|
||||
text: Final = "hello fanta"
|
||||
|
|
|
|||
|
|
@ -1,7 +1,12 @@
|
|||
import base64
|
||||
from pathlib import Path
|
||||
from typing import Final
|
||||
|
||||
import httpx
|
||||
import pytest
|
||||
import respx
|
||||
|
||||
import litellm
|
||||
from litellm.litellm_core_utils.prompt_templates.factory import (
|
||||
convert_to_gemini_tool_call_result,
|
||||
)
|
||||
|
|
@ -2727,3 +2732,40 @@ def test_gemini_server_side_tool_signature_not_duplicated_on_text():
|
|||
assert "thoughtSignature" not in text_part
|
||||
tool_call_part = next(p for p in parts if "toolCall" in p)
|
||||
assert tool_call_part["thoughtSignature"] == "server_side_signature"
|
||||
|
||||
|
||||
WHITE_PNG: Final = (Path(__file__).parents[4] / "white_100x100.png").read_bytes()
|
||||
|
||||
|
||||
@respx.mock
|
||||
def test_convert_tool_response_with_url_image(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
monkeypatch.setattr(litellm, "user_url_validation", False)
|
||||
image_url: Final = "https://tool-result-images.test/gemini-tool-response.png"
|
||||
respx.get(image_url).mock(return_value=httpx.Response(200, content=WHITE_PNG, headers={"content-type": "image/png"}))
|
||||
tool_message: Final = {
|
||||
"role": "tool",
|
||||
"tool_call_id": "call_test456",
|
||||
"content": [
|
||||
{"type": "text", "text": '{"url": "https://example.com"}'},
|
||||
{"type": "input_image", "image_url": image_url},
|
||||
],
|
||||
}
|
||||
last_message_with_tool_calls: Final = {
|
||||
"tool_calls": [
|
||||
{
|
||||
"id": "call_test456",
|
||||
"function": {"name": "type_text_at", "arguments": '{"x": 300, "y": 400, "text": "hello"}'},
|
||||
}
|
||||
]
|
||||
}
|
||||
|
||||
result: Final = convert_to_gemini_tool_call_result(tool_message, last_message_with_tool_calls)
|
||||
|
||||
assert isinstance(result, list)
|
||||
assert len(result) == 1
|
||||
assert "inline_data" not in result[0]
|
||||
function_response: Final = result[0]["function_response"]
|
||||
assert function_response["name"] == "type_text_at"
|
||||
assert len(function_response["parts"]) == 1
|
||||
inline_data: Final[BlobType] = function_response["parts"][0]["inline_data"]
|
||||
assert inline_data == {"data": base64.b64encode(WHITE_PNG).decode(), "mime_type": "image/png"}
|
||||
|
|
|
|||
|
|
@ -17,6 +17,7 @@ import respx
|
|||
|
||||
import urllib.parse
|
||||
from importlib import import_module
|
||||
from pathlib import Path
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
import litellm
|
||||
|
|
@ -56,6 +57,9 @@ def add_api_keys_to_env(monkeypatch):
|
|||
monkeypatch.delenv("AWS_WEB_IDENTITY_TOKEN_FILE", raising=False)
|
||||
|
||||
|
||||
WHITE_PNG: Final = (Path(__file__).parents[1] / "white_100x100.png").read_bytes()
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def openai_api_response():
|
||||
mock_response_data = {
|
||||
|
|
@ -213,6 +217,102 @@ async def test_url_with_format_param_openai(model, sync_mode):
|
|||
assert "format" not in json_str
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"model",
|
||||
[
|
||||
"gemini/gemini-1.5-flash",
|
||||
"bedrock/us.anthropic.claude-haiku-4-5-20251001-v1:0",
|
||||
"bedrock/invoke/anthropic.claude-haiku-4-5-20251001-v1:0",
|
||||
"anthropic/claude-3-5-sonnet",
|
||||
],
|
||||
)
|
||||
@pytest.mark.parametrize("sync_mode", [True, False])
|
||||
@pytest.mark.asyncio
|
||||
async def test_url_with_format_param(model, sync_mode, monkeypatch):
|
||||
from litellm import acompletion, completion
|
||||
from litellm.llms.custom_httpx.http_handler import AsyncHTTPHandler, HTTPHandler
|
||||
|
||||
if sync_mode:
|
||||
client = HTTPHandler()
|
||||
else:
|
||||
client = AsyncHTTPHandler()
|
||||
|
||||
image_url: Final = (
|
||||
"https://awsmp-logos.s3.amazonaws.com/seller-xw5kijmvmzasy/c233c9ade2ccb5491072ae232c814942.png"
|
||||
f"?case={sync_mode}-{model}"
|
||||
)
|
||||
args = {
|
||||
"model": model,
|
||||
"messages": [
|
||||
{
|
||||
"role": "user",
|
||||
"content": [
|
||||
{
|
||||
"type": "image_url",
|
||||
"image_url": {
|
||||
"url": image_url,
|
||||
"format": "image/png",
|
||||
},
|
||||
},
|
||||
{"type": "text", "text": "Describe this image"},
|
||||
],
|
||||
}
|
||||
],
|
||||
}
|
||||
if model.startswith("gemini/"):
|
||||
args["api_key"] = "test-api-key"
|
||||
monkeypatch.setattr(litellm, "user_url_validation", False)
|
||||
monkeypatch.setattr(litellm, "disable_aiohttp_transport", True)
|
||||
monkeypatch.setattr(litellm, "module_level_aclient", AsyncHTTPHandler(transport=httpx.AsyncHTTPTransport()))
|
||||
with (
|
||||
respx.mock(assert_all_called=False) as image_host,
|
||||
patch.object(client, "post", new=MagicMock()) as mock_client,
|
||||
):
|
||||
image_route = image_host.get(image_url).mock(
|
||||
return_value=httpx.Response(200, content=WHITE_PNG, headers={"content-type": "image/png"})
|
||||
)
|
||||
try:
|
||||
if sync_mode:
|
||||
response = completion(**args, client=client)
|
||||
else:
|
||||
response = await acompletion(**args, client=client)
|
||||
print(response)
|
||||
except Exception as e:
|
||||
pass
|
||||
|
||||
mock_client.assert_called()
|
||||
|
||||
print(mock_client.call_args.kwargs)
|
||||
|
||||
if "data" in mock_client.call_args.kwargs:
|
||||
json_str = mock_client.call_args.kwargs["data"]
|
||||
else:
|
||||
json_str = json.dumps(mock_client.call_args.kwargs["json"])
|
||||
|
||||
if isinstance(json_str, bytes):
|
||||
json_str = json_str.decode("utf-8")
|
||||
|
||||
print(f"type of json_str: {type(json_str)}")
|
||||
|
||||
if model.startswith("bedrock/invoke/"):
|
||||
assert "https://awsmp-logos.s3.amazonaws.com" not in json_str
|
||||
assert '"type":"base64"' in json_str or '"type": "base64"' in json_str
|
||||
assert '"data"' in json_str
|
||||
elif model.startswith("bedrock/"):
|
||||
assert "https://awsmp-logos.s3.amazonaws.com" not in json_str
|
||||
assert '"bytes"' in json_str or '"bytes":' in json_str
|
||||
elif model.startswith("anthropic/"):
|
||||
assert "https://awsmp-logos.s3.amazonaws.com" in json_str
|
||||
assert '"type":"url"' in json_str or '"type": "url"' in json_str
|
||||
else:
|
||||
assert "png" in json_str
|
||||
assert "jpeg" not in json_str
|
||||
|
||||
fetches_image: Final = not model.startswith("anthropic/")
|
||||
assert image_route.called is fetches_image
|
||||
assert (base64.b64encode(WHITE_PNG).decode() in json_str) is fetches_image
|
||||
|
||||
|
||||
def test_bedrock_latency_optimized_inference():
|
||||
from litellm.llms.custom_httpx.http_handler import HTTPHandler
|
||||
|
||||
|
|
|
|||
BIN
tests/white_100x100.png
Normal file
BIN
tests/white_100x100.png
Normal file
Binary file not shown.
|
After Width: | Height: | Size: 214 B |
Loading…
Add table
Reference in a new issue