test: serve test media from our own fixtures instead of third-party hosts

A dozen tests downloaded their image, audio, or document from someone else's
web server: pexels, pinterest, gratisography, squarespace, dummyimage, arxiv,
wikimedia, w3.org, cmu.edu, and a placeholder service that has since shut down.
That makes a run depend on hosts we do not control, and on assets we have no
right to redistribute. It also fails for reasons that have nothing to do with
litellm: the Bedrock vision e2e went red last week on a Wikimedia 429.

tests/fixtures now holds media we generated ourselves, with the drawing code in
its README so anyone can regenerate it: a 100x100 PNG and JPEG, a one-page PDF,
a markdown file, a csv, and a short spoken wav. Tests that only need bytes read
the file. Tests that have to exercise a real URL fetch (convert_url_to_base64,
Bedrock's document and image embedding, Gemini's tool-result media) take the new
asset_base_url fixture, which serves tests/fixtures over loopback and adds its
own host to user_url_allowed_hosts so litellm's SSRF guard lets it through.

Two of these tests got stronger on the way past. test_convert_tool_response_with_url_image
wrapped its whole body in `except Exception: pytest.skip(...)`, which swallowed
assertion failures as well as download failures, so it could not fail; it now
runs for real and checks the inlined bytes and mime type. test_gpt_vision_token_counting
asserted nothing at all and now checks that the image adds tokens over the same
prompt without it. test_vision_with_custom_model compares against the image the
test actually sent rather than a pasted base64 literal.

test_bedrock_document_understanding traded its .xls case for a .csv, since we
can author a csv and cannot author a legacy .xls; both are Bedrock document
formats, so the coverage is the same shape.

What is left, and why: the tests where the provider fetches the URL rather than
litellm (base_llm_unit_tests' image_url and pdf file_id cases, the Vertex pdf
pass-through, the AssemblyAI passthrough) cannot point at loopback. Those
already reference content we own, through jsdelivr or our own S3 buckets.
Moving them off third-party CDNs needs a public host we control, which is a
separate decision.
This commit is contained in:
Yuneng Jiang 2026-08-27 15:53:20 -07:00
parent 628c9d1a74
commit dbf608bfc5
No known key found for this signature in database
24 changed files with 211 additions and 111 deletions

23
tests/fixtures/README.md vendored Normal file
View file

@ -0,0 +1,23 @@
# tests/fixtures
Media the test suite feeds to litellm, generated by us and committed here so no
test depends on a third-party host staying up or on someone else's copyright.
- `test_image.png` / `test_image.jpg`: a 100x100 square with two diagonal lines,
drawn in code (see the snippet below). Any test that just needs valid image
bytes should use one of these.
```python
from PIL import Image, ImageDraw
img = Image.new("RGB", (100, 100), (255, 255, 255))
d = ImageDraw.Draw(img)
d.rectangle([10, 10, 89, 89], outline=(17, 17, 17), width=4)
d.line([10, 10, 89, 89], fill=(200, 30, 30), width=4)
d.line([89, 10, 10, 89], fill=(30, 80, 200), width=4)
img.save("test_image.png", "PNG", optimize=True)
img.save("test_image.jpg", "JPEG", quality=85, optimize=True)
```
`asset_server.py` serves this directory over HTTP on loopback, for the tests that
have to exercise a real URL fetch rather than a `data:` URL

52
tests/fixtures/asset_server.py vendored Normal file
View file

@ -0,0 +1,52 @@
"""Serves tests/fixtures over loopback HTTP.
Some code paths under test (``convert_url_to_base64``, Bedrock's image
embedding, Gemini's tool-result media handling) only run when the image arrives
as a URL, so a ``data:`` URL would skip the very branch being tested. Those
tests point at this server instead of a third-party image host.
litellm's SSRF guard rejects loopback by default, so the server also adds its
own host to ``litellm.user_url_allowed_hosts`` for as long as it is up: the
same setting an operator uses to reach an internal image host.
"""
from __future__ import annotations
import threading
from functools import partial
from http.server import SimpleHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path
from typing import Iterator
import litellm
ASSETS_DIR = Path(__file__).parent
TEST_IMAGE_PNG = ASSETS_DIR / "test_image.png"
TEST_IMAGE_JPG = ASSETS_DIR / "test_image.jpg"
TEST_SPEECH_WAV = ASSETS_DIR / "test_speech.wav"
TEST_DOCUMENT_PDF = ASSETS_DIR / "test_document.pdf"
TEST_DOCUMENT_MD = ASSETS_DIR / "test_document.md"
TEST_TABLE_CSV = ASSETS_DIR / "test_table.csv"
class _QuietHandler(SimpleHTTPRequestHandler):
def log_message(self, *args: object) -> None:
pass
def serve_assets() -> Iterator[str]:
server = ThreadingHTTPServer(
("127.0.0.1", 0), partial(_QuietHandler, directory=str(ASSETS_DIR))
)
host = f"127.0.0.1:{server.server_address[1]}"
thread = threading.Thread(target=server.serve_forever, daemon=True)
thread.start()
previously_allowed = litellm.user_url_allowed_hosts
litellm.user_url_allowed_hosts = [*previously_allowed, host]
try:
yield f"http://{host}"
finally:
litellm.user_url_allowed_hosts = previously_allowed
server.shutdown()
server.server_close()
thread.join(timeout=5)

6
tests/fixtures/test_document.md vendored Normal file
View file

@ -0,0 +1,6 @@
# LiteLLM test document
This file exists so the test suite can send a Markdown document without
downloading one from someone else's web server.
The quick brown fox jumps over the lazy dog.

42
tests/fixtures/test_document.pdf vendored Normal file
View file

@ -0,0 +1,42 @@
%PDF-1.4
1 0 obj
<< /Type /Catalog /Pages 2 0 R >>
endobj
2 0 obj
<< /Type /Pages /Kids [3 0 R] /Count 1 >>
endobj
3 0 obj
<< /Type /Page /Parent 2 0 R /MediaBox [0 0 612 792] /Resources << /Font << /F1 4 0 R >> >> /Contents 5 0 R >>
endobj
4 0 obj
<< /Type /Font /Subtype /Type1 /BaseFont /Helvetica >>
endobj
5 0 obj
<< /Length 255 >>
stream
BT
/F1 14 Tf
72 720 Td
18 TL
(LiteLLM test document) Tj T*
() Tj T*
(This file exists so the test suite can send a PDF without) Tj T*
(downloading one from someone else's web server.) Tj T*
() Tj T*
(The quick brown fox jumps over the lazy dog.) Tj T*
ET
endstream
endobj
xref
0 6
0000000000 65535 f
0000000009 00000 n
0000000058 00000 n
0000000115 00000 n
0000000241 00000 n
0000000311 00000 n
trailer
<< /Size 6 /Root 1 0 R >>
startxref
616
%%EOF

BIN
tests/fixtures/test_image.jpg vendored Normal file

Binary file not shown.

After

Width:  |  Height:  |  Size: 2.9 KiB

BIN
tests/fixtures/test_image.png vendored Normal file

Binary file not shown.

After

Width:  |  Height:  |  Size: 752 B

BIN
tests/fixtures/test_speech.wav vendored Normal file

Binary file not shown.

5
tests/fixtures/test_table.csv vendored Normal file
View file

@ -0,0 +1,5 @@
region,quarter,revenue
north,Q1,1200
north,Q2,1450
south,Q1,980
south,Q2,1130
1 region quarter revenue
2 north Q1 1200
3 north Q2 1450
4 south Q1 980
5 south Q2 1130

View file

@ -1124,14 +1124,10 @@ def test_usage_object_null_tokens():
def test_is_base64_encoded():
import base64
import requests
from tests.fixtures.asset_server import TEST_IMAGE_PNG
litellm.set_verbose = True
url = "https://dummyimage.com/100/100/fff&text=Test+image"
response = requests.get(url)
file_data = response.content
encoded_file = base64.b64encode(file_data).decode("utf-8")
encoded_file = base64.b64encode(TEST_IMAGE_PNG.read_bytes()).decode("utf-8")
base64_image = f"data:image/png;base64,{encoded_file}"
from litellm.utils import is_base64_encoded

View file

@ -16,17 +16,13 @@ from litellm.utils import (
get_optional_params,
get_optional_params_embeddings,
)
import requests
import base64
# 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
from tests.fixtures.asset_server import TEST_IMAGE_PNG
encoded_file = base64.b64encode(file_data).decode("utf-8")
encoded_file = base64.b64encode(TEST_IMAGE_PNG.read_bytes()).decode("utf-8")
base64_image = f"data:image/png;base64,{encoded_file}"

View file

@ -30,6 +30,7 @@ from openai import OpenAI
sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..")))
from tests._live_test_helpers import _skip_live_prompt_caching_test # noqa: E402
from tests.fixtures.asset_server import TEST_SPEECH_WAV # noqa: E402
def _usage_format_tests(usage: litellm.Usage):
@ -758,11 +759,7 @@ class BaseLLMChatTest(ABC):
f"Model={base_completion_call_args['model']} does not support audio input"
)
url = "https://openaiassets.blob.core.windows.net/$web/API/docs/audio/alloy.wav"
response = httpx.get(url)
response.raise_for_status()
wav_data = response.content
encoded_string = base64.b64encode(wav_data).decode("utf-8")
encoded_string = base64.b64encode(TEST_SPEECH_WAV.read_bytes()).decode("utf-8")
completion = self.completion_function(
**base_completion_call_args,
@ -1274,12 +1271,8 @@ class BaseLLMChatTest(ABC):
print("Model does not support audio input")
pytest.skip("Model does not support audio input")
url = "https://openaiassets.blob.core.windows.net/$web/API/docs/audio/alloy.wav"
response = httpx.get(url)
response.raise_for_status()
wav_data = response.content
audio_format = "wav"
encoded_string = base64.b64encode(wav_data).decode("utf-8")
encoded_string = base64.b64encode(TEST_SPEECH_WAV.read_bytes()).decode("utf-8")
audio_content = [{"type": "text", "text": "What is in this recording?"}]

View file

@ -27,6 +27,12 @@ from tests._vcr_conftest_common import ( # noqa: E402,F401
vcr_config_dict,
)
from tests.fake_openai_endpoint import ensure_fake_openai_endpoint # noqa: E402
from tests.fixtures.asset_server import serve_assets # noqa: E402
@pytest.fixture
def asset_base_url():
yield from serve_assets()
@pytest.fixture(scope="session", autouse=True)

View file

@ -2441,7 +2441,7 @@ class TestBedrockEmbedding(BaseLLMEmbeddingTest):
@pytest.mark.asyncio
async def test_bedrock_image_url_sync_client():
async def test_bedrock_image_url_sync_client(asset_base_url):
from litellm.llms.custom_httpx.http_handler import AsyncHTTPHandler
import logging
from litellm import verbose_logger
@ -2459,7 +2459,7 @@ async def test_bedrock_image_url_sync_client():
{
"type": "image_url",
"image_url": {
"url": "https://awsmp-logos.s3.amazonaws.com/seller-xw5kijmvmzasy/c233c9ade2ccb5491072ae232c814942.png"
"url": f"{asset_base_url}/test_image.png"
},
},
],
@ -2523,18 +2523,12 @@ def test_bedrock_error_handling_streaming(exception_type, expected_status_code):
@pytest.mark.parametrize(
"image_url",
[
"https://www.w3.org/WAI/ER/tests/xhtml/testfiles/resources/pdf/dummy.pdf",
# "https://raw.githubusercontent.com/datasets/gdp/master/data/gdp.csv",
"https://www.cmu.edu/blackboard/files/evaluate/tests-example.xls",
# "https://raw.githubusercontent.com/datasets/sample-data/master/README.txt", # invalid url
"https://raw.githubusercontent.com/mdn/content/main/README.md",
],
"asset_name",
["test_document.pdf", "test_table.csv", "test_document.md"],
)
@pytest.mark.flaky(retries=6, delay=2)
@pytest.mark.asyncio
async def test_bedrock_document_understanding(image_url):
async def test_bedrock_document_understanding(asset_name, asset_base_url):
from litellm import acompletion
litellm._turn_on_debug()
@ -2544,7 +2538,7 @@ async def test_bedrock_document_understanding(image_url):
{"type": "text", "text": f"What's this file about?"},
{
"type": "image_url",
"image_url": image_url,
"image_url": f"{asset_base_url}/{asset_name}",
},
]

View file

@ -11,7 +11,8 @@ import litellm
from litellm import Choices, Message, ModelResponse
from litellm.types.utils import StreamingChoices, ChatCompletionAudioResponse
import base64
import requests
from tests.fixtures.asset_server import TEST_SPEECH_WAV
def check_non_streaming_response(completion):
@ -88,17 +89,12 @@ async def test_audio_output_from_model(stream):
@pytest.mark.parametrize("stream", [True, False])
@pytest.mark.parametrize("model", ["gpt-audio-1.5"])
async def test_audio_input_to_model(stream, model):
# Fetch the audio file and convert it to a base64 encoded string
audio_format = "pcm16"
if stream is False:
audio_format = "wav"
litellm._turn_on_debug()
litellm.drop_params = True
url = "https://openaiassets.blob.core.windows.net/$web/API/docs/audio/alloy.wav"
response = requests.get(url)
response.raise_for_status()
wav_data = response.content
encoded_string = base64.b64encode(wav_data).decode("utf-8")
encoded_string = base64.b64encode(TEST_SPEECH_WAV.read_bytes()).decode("utf-8")
try:
completion = await litellm.acompletion(
model=model,

View file

@ -209,19 +209,16 @@ async def test_vision_with_custom_model():
"""
import base64
import requests
from openai import AsyncOpenAI
from tests.fixtures.asset_server import TEST_IMAGE_PNG
client = AsyncOpenAI(api_key="fake-api-key")
litellm.set_verbose = True
api_base = "https://my-custom.api.openai.com"
# Fetch and encode a test image
url = "https://dummyimage.com/100/100/fff&text=Test+image"
response = requests.get(url)
file_data = response.content
encoded_file = base64.b64encode(file_data).decode("utf-8")
encoded_file = base64.b64encode(TEST_IMAGE_PNG.read_bytes()).decode("utf-8")
base64_image = f"data:image/png;base64,{encoded_file}"
with patch.object(
@ -261,9 +258,7 @@ async def test_vision_with_custom_model():
{"type": "text", "text": "What's in this image?"},
{
"type": "image_url",
"image_url": {
"url": "data:image/png;base64,iVBORw0KGgoAAAANSUhEUgAAAGQAAABkBAMAAACCzIhnAAAAG1BMVEURAAD///+ln5/h39/Dv79qX18uHx+If39MPz9oMSdmAAAACXBIWXMAAA7EAAAOxAGVKw4bAAABDElEQVRYhe2SzWqEMBRGPyQTfQxJsc5jBKGzFmlslyFIZxsCQ7sUaWd87EanpdpIrbtC71mE/NyTm9wEIAiCIAiC+N/otQBxU2Sf/aeh4enqptHXri+/yxIq63jlKCw6cXssnr3ObdzdGYFYCJ2IzHKXLygHXCB98Gm4DE+ZZemu5EisQSyZTmyg+AuzQbkezCuIy7EI0k9Ig3FtruwydY+qniqtV5yQyo8qpUIl2fc90KVzJWohWf2qu75vlw52rdfjVDHg8vLWwixW7PChqLkSyUadwfSS0uQZhEvRuIkS53uJvrK8cGWYaPwpGt8efvw+vlo8TPMzcmP8w7lrNypc1RsNgiAIgiD+Iu/RyDYhCaWrgQAAAABJRU5ErkJggg=="
},
"image_url": {"url": base64_image},
},
],
},
@ -473,7 +468,7 @@ class TestOpenAIGPT4OAudioTranscription(BaseLLMAudioTranscriptionTest):
@pytest.mark.asyncio
@pytest.mark.parametrize("model", ["gpt-4o"])
async def test_openai_pdf_url(model):
async def test_openai_pdf_url(model, asset_base_url):
from litellm.utils import return_raw_request, CallTypes
request = return_raw_request(
@ -487,7 +482,9 @@ async def test_openai_pdf_url(model):
{"type": "text", "text": "What is the first page of the PDF?"},
{
"type": "file",
"file": {"file_id": "https://arxiv.org/pdf/2303.08774"},
"file": {
"file_id": f"{asset_base_url}/test_document.pdf"
},
},
],
}

View file

@ -181,10 +181,8 @@ def test_bedrock_tool_calling_pt():
print(converted_tools)
def test_convert_url_to_img():
response_url = convert_url_to_base64(
url="https://images.pexels.com/photos/1319515/pexels-photo-1319515.jpeg?auto=compress&cs=tinysrgb&w=1260&h=750&dpr=1"
)
def test_convert_url_to_img(asset_base_url):
response_url = convert_url_to_base64(url=f"{asset_base_url}/test_image.jpg")
assert "image/jpeg" in response_url
@ -1289,17 +1287,16 @@ def test_just_system_message():
assert "bedrock requires at least one non-system message" in str(e.value)
def test_convert_generic_image_chunk_to_openai_image_obj():
def test_convert_generic_image_chunk_to_openai_image_obj(asset_base_url):
from litellm.litellm_core_utils.prompt_templates.factory import (
convert_generic_image_chunk_to_openai_image_obj,
convert_to_anthropic_image_obj,
)
url = "https://i.pinimg.com/736x/b4/b1/be/b4b1becad04d03a9071db2817fc9fe77.jpg"
url = f"{asset_base_url}/test_image.jpg"
image_obj = convert_to_anthropic_image_obj(url, format=None)
url_str = convert_generic_image_chunk_to_openai_image_obj(image_obj)
image_obj = convert_to_anthropic_image_obj(url_str, format=None)
print(image_obj)
assert convert_to_anthropic_image_obj(url_str, format=None) == image_obj
def test_hf_chat_template():

View file

@ -49,6 +49,12 @@ from tests._vcr_conftest_common import ( # noqa: E402,F401
vcr_config_dict,
)
from tests.fake_openai_endpoint import ensure_fake_openai_endpoint # noqa: E402
from tests.fixtures.asset_server import serve_assets # noqa: E402
@pytest.fixture
def asset_base_url():
yield from serve_assets()
@pytest.fixture(scope="session", autouse=True)

View file

@ -2000,16 +2000,12 @@ async def test_vertexai_multimodal_embedding_image_in_input():
async def test_vertexai_multimodal_embedding_base64image_in_input():
import base64
import requests
from tests.fixtures.asset_server import TEST_IMAGE_PNG
load_vertex_ai_credentials()
mock_response = AsyncMock()
url = "https://dummyimage.com/100/100/fff&text=Test+image"
response = requests.get(url)
file_data = response.content
encoded_file = base64.b64encode(file_data).decode("utf-8")
encoded_file = base64.b64encode(TEST_IMAGE_PNG.read_bytes()).decode("utf-8")
base64_image = f"data:image/png;base64,{encoded_file}"
def return_val():
@ -4061,7 +4057,7 @@ def test_vertex_ai_gemini_audio_ogg():
"content": [
{
"file": {
"file_id": "https://upload.wikimedia.org/wikipedia/commons/5/5f/En-us-public.ogg"
"file_id": "https://example.com/audio.ogg"
},
"type": "file",
}

View file

@ -616,14 +616,10 @@ def test_completion_base64(model):
try:
import base64
import requests
from tests.fixtures.asset_server import TEST_IMAGE_PNG
litellm.set_verbose = True
url = "https://dummyimage.com/100/100/fff&text=Test+image"
response = requests.get(url)
file_data = response.content
encoded_file = base64.b64encode(file_data).decode("utf-8")
encoded_file = base64.b64encode(TEST_IMAGE_PNG.read_bytes()).decode("utf-8")
base64_image = f"data:image/png;base64,{encoded_file}"
resp = litellm.completion(
model=model,

View file

@ -162,14 +162,10 @@ async def test_together_ai_embedding(model, api_base, api_key, sync_mode):
# test_openai_azure_embedding_simple()
import base64
import requests
from tests.fixtures.asset_server import TEST_IMAGE_PNG
litellm.set_verbose = True
url = "https://dummyimage.com/100/100/fff&text=Test+image"
response = requests.get(url)
file_data = response.content
encoded_file = base64.b64encode(file_data).decode("utf-8")
encoded_file = base64.b64encode(TEST_IMAGE_PNG.read_bytes()).decode("utf-8")
base64_image = f"data:image/png;base64,{encoded_file}"

View file

@ -75,7 +75,7 @@ def test_ollama_json_mode():
# test_ollama_json_mode()
def test_ollama_vision_model():
def test_ollama_vision_model(asset_base_url):
from litellm.llms.custom_httpx.http_handler import HTTPHandler
client = HTTPHandler()
@ -93,7 +93,7 @@ def test_ollama_vision_model():
{
"type": "image_url",
"image_url": {
"url": "https://dummyimage.com/100/100/fff&text=Test+image"
"url": f"{asset_base_url}/test_image.png"
},
},
],

View file

@ -39,6 +39,13 @@ from litellm.llms.custom_httpx.async_client_cleanup import (
)
from litellm.proxy.db import tool_registry_writer as tool_registry_writer_module
from tests.fixtures.asset_server import serve_assets
@pytest.fixture
def asset_base_url():
yield from serve_assets()
def _reset_module_level_aws_auth_caches():
"""

View file

@ -380,7 +380,7 @@ def test_encoding_and_decoding():
# test_encoding_and_decoding()
def test_gpt_vision_token_counting():
def test_gpt_vision_token_counting(asset_base_url):
messages = [
{
"role": "user",
@ -388,13 +388,17 @@ def test_gpt_vision_token_counting():
{"type": "text", "text": "Whats in this image?"},
{
"type": "image_url",
"image_url": "https://awsmp-logos.s3.amazonaws.com/seller-xw5kijmvmzasy/c233c9ade2ccb5491072ae232c814942.png",
"image_url": f"{asset_base_url}/test_image.png",
},
],
}
]
text_only_tokens = token_counter(
model="gpt-4-vision-preview",
messages=[{"role": "user", "content": "Whats in this image?"}],
)
tokens = token_counter(model="gpt-4-vision-preview", messages=messages)
print(f"tokens: {tokens}")
assert tokens > text_only_tokens
# test_gpt_vision_token_counting()

View file

@ -15,6 +15,8 @@ from litellm.llms.vertex_ai.gemini.transformation import (
from litellm.types.llms.vertex_ai import BlobType
from litellm.types.utils import Message
from tests.fixtures.asset_server import TEST_IMAGE_PNG
def test_check_if_part_exists_in_parts():
parts = [
@ -1669,12 +1671,9 @@ def test_gemini_history_nests_multimodal_tool_response_parts():
]
def test_convert_tool_response_with_url_image():
def test_convert_tool_response_with_url_image(asset_base_url):
"""Test tool response with HTTP URL image (will download and convert)."""
import pytest
# Use a publicly accessible test image URL
test_image_url = "https://via.placeholder.com/1x1.png"
test_image_url = f"{asset_base_url}/test_image.png"
tool_message = {
"role": "tool",
@ -1697,30 +1696,23 @@ def test_convert_tool_response_with_url_image():
]
}
try:
result = convert_to_gemini_tool_call_result(
tool_message, last_message_with_tool_calls
)
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"
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}")
assert "parts" in function_response
assert len(function_response["parts"]) == 1
inline_data: BlobType = function_response["parts"][0]["inline_data"]
assert inline_data["data"] == base64.b64encode(TEST_IMAGE_PNG.read_bytes()).decode()
assert inline_data["mime_type"] == "image/png"
def test_convert_tool_response_text_only():