scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-08-28 05:25:59 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 36
litellm/tests/image_gen_tests/test_image_generation.py
yuneng-jiang 6a0d03914c
test: drop the cwd-relative sys.path.insert calls from the test suite (#37802) 
* test: drop the cwd-relative sys.path.insert calls from the test suite

TQ003 stands at 1,077 across 1,058 files, and 1,015 of them are the same shape:
sys.path.insert(0, os.path.abspath("../..")) and its deeper siblings. The
argument resolves against the working directory rather than the file, so from
the repo root, where every job runs pytest, it inserts the directory two levels
above the checkout. It has never pointed at litellm. The package is installed
into the environment anyway, which is what actually makes the import work, and
what the rule's message has said all along.

Removing them leaves 1,634 imports of sys and os with no remaining reference,
and those go too, except where another test module imports the name back out of
the file. The rest of TQ003 is 62 call sites that resolve against __file__ or a
variable, which are a different question and are left alone.

Collection is identical either way: 45,871 tests and the same 51 pre-existing
collection errors before and after, and ruff reports no new undefined name.

* test: drop the duplicate imports the sys.path sweep exposed to F811

* test(pre-call-utils): restore the os import the new bedrock tests need
2026-08-22 09:25:58 -07:00

454 lines
17 KiB
Python
Raw Permalink Blame History

# What this tests?
## This tests the litellm support for the openai /generations endpoint
import logging
import os
import traceback
from unittest.mock import AsyncMock, MagicMock, patch
from dotenv import load_dotenv
from openai.types.image import Image
from litellm.caching import InMemoryCache
logging.basicConfig(level=logging.DEBUG)
load_dotenv()
import asyncio
import pytest
import litellm
import json
import tempfile
from base_image_generation_test import BaseImageGenTest, TestCustomLogger
import logging
from litellm._logging import verbose_logger
verbose_logger.setLevel(logging.DEBUG)
def get_vertex_ai_creds_json() -> dict:
# Define the path to the vertex_key.json file
print("loading vertex ai credentials")
filepath = os.path.dirname(os.path.abspath(__file__))
vertex_key_path = filepath + "/vertex_key.json"
# Read the existing content of the file or create an empty dictionary
try:
with open(vertex_key_path, "r") as file:
# Read the file content
print("Read vertexai file path")
content = file.read()
# If the file is empty or not valid JSON, create an empty dictionary
if not content or not content.strip():
service_account_key_data = {}
else:
# Attempt to load the existing JSON content
file.seek(0)
service_account_key_data = json.load(file)
except FileNotFoundError:
# If the file doesn't exist, create an empty dictionary
service_account_key_data = {}
# Update the service_account_key_data with environment variables
private_key_id = os.environ.get("VERTEX_AI_PRIVATE_KEY_ID", "")
private_key = os.environ.get("VERTEX_AI_PRIVATE_KEY", "")
private_key = private_key.replace("\\n", "\n")
service_account_key_data["private_key_id"] = private_key_id
service_account_key_data["private_key"] = private_key
return service_account_key_data
def load_vertex_ai_credentials():
# Define the path to the vertex_key.json file
print("loading vertex ai credentials")
filepath = os.path.dirname(os.path.abspath(__file__))
vertex_key_path = filepath + "/vertex_key.json"
# Read the existing content of the file or create an empty dictionary
try:
with open(vertex_key_path, "r") as file:
# Read the file content
print("Read vertexai file path")
content = file.read()
# If the file is empty or not valid JSON, create an empty dictionary
if not content or not content.strip():
service_account_key_data = {}
else:
# Attempt to load the existing JSON content
file.seek(0)
service_account_key_data = json.load(file)
except FileNotFoundError:
# If the file doesn't exist, create an empty dictionary
service_account_key_data = {}
# Update the service_account_key_data with environment variables
private_key_id = os.environ.get("VERTEX_AI_PRIVATE_KEY_ID", "")
private_key = os.environ.get("VERTEX_AI_PRIVATE_KEY", "")
private_key = private_key.replace("\\n", "\n")
service_account_key_data["private_key_id"] = private_key_id
service_account_key_data["private_key"] = private_key
# Create a temporary file
with tempfile.NamedTemporaryFile(mode="w+", delete=False) as temp_file:
# Write the updated content to the temporary files
json.dump(service_account_key_data, temp_file, indent=2)
# Export the temporary file as GOOGLE_APPLICATION_CREDENTIALS
os.environ["GOOGLE_APPLICATION_CREDENTIALS"] = os.path.abspath(temp_file.name)
class TestVertexAIGeminiImageGeneration(BaseImageGenTest):
"""Test Gemini image generation models (Nano Banana)"""
def get_base_image_generation_call_args(self) -> dict:
# comment this when running locally
load_vertex_ai_credentials()
litellm.in_memory_llm_clients_cache = InMemoryCache()
return {
"model": "vertex_ai/gemini-2.5-flash-image",
"vertex_ai_project": "litellm-ci-cd",
"vertex_ai_location": "us-central1",
"n": 1,
"size": "1024x1024",
}
class TestBedrockNovaCanvasTextToImage(BaseImageGenTest):
def get_base_image_generation_call_args(self) -> dict:
litellm.in_memory_llm_clients_cache = InMemoryCache()
return {
"model": "bedrock/amazon.nova-canvas-v1:0",
"n": 1,
"size": "320x320",
"imageGenerationConfig": {"cfgScale": 6.5, "seed": 12},
"taskType": "TEXT_IMAGE",
"aws_region_name": "us-east-1",
}
class TestBedrockNovaCanvasColorGuidedGeneration(BaseImageGenTest):
def get_base_image_generation_call_args(self) -> dict:
litellm.in_memory_llm_clients_cache = InMemoryCache()
return {
"model": "bedrock/amazon.nova-canvas-v1:0",
"n": 1,
"size": "320x320",
"imageGenerationConfig": {"cfgScale": 6.5, "seed": 12},
"taskType": "COLOR_GUIDED_GENERATION",
"colorGuidedGenerationParams": {"colors": ["#FFFFFF"]},
"aws_region_name": "us-east-1",
}
class TestOpenAIGPTImage1(BaseImageGenTest):
def get_base_image_generation_call_args(self) -> dict:
return {"model": "gpt-image-1"}
@pytest.mark.skip(reason="Recraft image generation API only tested locally")
class TestRecraftImageGeneration(BaseImageGenTest):
def get_base_image_generation_call_args(self) -> dict:
return {"model": "recraft/recraftv3"}
class TestAimlImageGeneration(BaseImageGenTest):
def get_base_image_generation_call_args(self) -> dict:
return {"model": "aiml/flux-pro/v1.1"}
@pytest.mark.asyncio(scope="module")
@pytest.mark.flaky(retries=0)
async def test_basic_image_generation(self):
"""Test basic image generation"""
from unittest.mock import AsyncMock, patch
mock_aiml_response = {
"created": 1703658209,
"data": [{"url": "https://example.com/generated_image.png"}],
}
mock_response = MagicMock()
mock_response.status_code = 200
mock_response.json.return_value = mock_aiml_response
mock_response.text = json.dumps(mock_aiml_response)
mock_response.headers = {}
with (
patch(
"litellm.llms.custom_httpx.http_handler.AsyncHTTPHandler.post",
new_callable=AsyncMock,
) as mock_async_post,
patch(
"litellm.llms.custom_httpx.http_handler.HTTPHandler.post",
) as mock_sync_post,
):
mock_async_post.return_value = mock_response
mock_sync_post.return_value = mock_response
try:
litellm._turn_on_debug()
custom_logger = TestCustomLogger()
litellm.logging_callback_manager._reset_all_callbacks()
litellm.callbacks = [custom_logger]
base_image_generation_call_args = (
self.get_base_image_generation_call_args()
)
litellm.set_verbose = True
# Pass dummy api_key so validate_environment passes; HTTP is mocked
response = await litellm.aimage_generation(
**base_image_generation_call_args,
prompt="A image of a otter",
api_key="test-key-mocked-no-credits-needed",
)
print("FAL AI RESPONSE: ", response)
await asyncio.sleep(1)
# assert response._hidden_params["response_cost"] is not None
# assert response._hidden_params["response_cost"] > 0
# print("response_cost", response._hidden_params["response_cost"])
logged_standard_logging_payload = custom_logger.standard_logging_payload
print(
"logged_standard_logging_payload", logged_standard_logging_payload
)
assert logged_standard_logging_payload is not None
assert logged_standard_logging_payload["response_cost"] is not None
assert logged_standard_logging_payload["response_cost"] > 0
import openai
from openai.types.images_response import ImagesResponse
# print openai version
print("openai version=", openai.__version__)
response_dict = dict(response)
if "usage" in response_dict:
response_dict["usage"] = dict(response_dict["usage"])
print("response usage=", response_dict.get("usage"))
assert (
response.data is not None
) # type guard for iteration (base fails here if None)
for d in response.data:
assert isinstance(d, Image)
print("data in response.data", d)
assert d.b64_json is not None or d.url is not None
except litellm.RateLimitError as e:
pass
except litellm.ContentPolicyViolationError:
pass # Azure randomly raises these errors - skip when they occur
except litellm.InternalServerError:
pass
except Exception as e:
if "Your task failed as a result of our safety system." in str(e):
pass
else:
pytest.fail(f"An exception occurred - {str(e)}")
class TestGoogleImageGen(BaseImageGenTest):
def get_base_image_generation_call_args(self) -> dict:
return {"model": "gemini/gemini-3.1-flash-image"}
@pytest.mark.skip(reason="Runwayml image generation API only tested locally")
class TestRunwaymlImageGeneration(BaseImageGenTest):
def get_base_image_generation_call_args(self) -> dict:
return {"model": "runwayml/gen4_image"}
## AZURE AI DALL-E 3 is deprecated and new deployments cannot be made
# class TestAzureOpenAIDalle3(BaseImageGenTest):
# def get_base_image_generation_call_args(self) -> dict:
# return {
# "model": "azure/dall-e-3",
# "api_version": "2024-02-01",
# "api_base": os.getenv("AZURE_AI_API_BASE"),
# "api_key": os.getenv("AZURE_AI_API_KEY"),
# "metadata": {
# "model_info": {
# "base_model": "azure/dall-e-3",
# }
# },
# }
@pytest.mark.skip(reason="model EOL")
@pytest.mark.asyncio
async def test_aimage_generation_bedrock_with_optional_params():
try:
litellm.in_memory_llm_clients_cache = InMemoryCache()
response = await litellm.aimage_generation(
prompt="A cute baby sea otter",
model="bedrock/stability.stable-diffusion-xl-v1",
size="256x256",
)
print(f"response: {response}")
except litellm.RateLimitError as e:
pass
except litellm.ContentPolicyViolationError:
pass # Azure randomly raises these errors skip when they occur
except Exception as e:
if "Your task failed as a result of our safety system." in str(e):
pass
else:
pytest.fail(f"An exception occurred - {str(e)}")
@pytest.mark.asyncio
async def test_aiml_image_generation_with_dynamic_api_key():
"""
Test that when api_key is passed as a dynamic parameter to aimage_generation,
it gets properly used for AIML provider authentication instead of falling back
to environment variables.
This test validates the fix for ensuring dynamic API keys are respected
when making image generation requests to the AIML provider.
"""
from unittest.mock import AsyncMock, patch, MagicMock
import httpx
# Mock AIML response
mock_aiml_response = {
"created": 1703658209,
"data": [{"url": "https://example.com/generated_image.png"}],
}
# Track captured arguments
captured_headers = None
captured_url = None
captured_json_data = None
def capture_post_call(*args, **kwargs):
nonlocal captured_headers, captured_url, captured_json_data
captured_url = kwargs.get("url") or (args[0] if args else None)
captured_headers = kwargs.get("headers", {})
captured_json_data = kwargs.get("json", {})
# Create a mock response
mock_response = MagicMock()
mock_response.status_code = 200
mock_response.json.return_value = mock_aiml_response
mock_response.text = json.dumps(mock_aiml_response)
return mock_response
# Mock the HTTP client that actually makes the request (sync version for image generation)
with patch("litellm.llms.custom_httpx.http_handler.HTTPHandler.post") as mock_post:
mock_post.side_effect = capture_post_call
# Test with dynamic api_key
test_api_key = "test-dynamic-api-key-12345"
response = await litellm.aimage_generation(
prompt="A cute baby sea otter",
model="aiml/flux-pro/v1.1",
api_key=test_api_key, # This should be used instead of env vars
)
# Validate the response (mocked response processing might not populate data correctly)
assert response is not None
# The most important validations: API key and endpoint usage
# These prove that the dynamic API key was properly used
assert captured_headers is not None
assert "Authorization" in captured_headers
assert captured_headers["Authorization"] == f"Bearer {test_api_key}"
print("TESTCAPTURED HEADERS", captured_headers)
# Validate the correct AIML endpoint was called
assert captured_url is not None
assert "api.aimlapi.com" in captured_url
assert "/v1/images/generations" in captured_url
# Validate the request data
assert captured_json_data is not None
assert captured_json_data["prompt"] == "A cute baby sea otter"
assert captured_json_data["model"] == "flux-pro/v1.1"
@pytest.mark.asyncio
async def test_aiml_openai_gpt_image_2_request_uses_openai_param_shape():
"""End-to-end check that ``aiml/openai/gpt-image-2`` keeps the upstream
OpenAI request shape (``size``/``n``/``response_format``) instead of
being remapped to the AI/ML flux schema (``image_size``/``num_images``/
``output_format``), and hits the correct upstream model name.
"""
from unittest.mock import MagicMock, patch
import json as _json
mock_aiml_response = {
"created": 1703658209,
"data": [{"url": "https://example.com/gpt-image-2.png"}],
}
captured = {}
def capture_post_call(*args, **kwargs):
captured["url"] = kwargs.get("url") or (args[0] if args else None)
captured["headers"] = kwargs.get("headers", {})
captured["json"] = kwargs.get("json", {})
mock_response = MagicMock()
mock_response.status_code = 200
mock_response.json.return_value = mock_aiml_response
mock_response.text = _json.dumps(mock_aiml_response)
return mock_response
with patch("litellm.llms.custom_httpx.http_handler.HTTPHandler.post") as mock_post:
mock_post.side_effect = capture_post_call
await litellm.aimage_generation(
prompt="A T-Rex relaxing on a beach",
model="aiml/openai/gpt-image-2",
api_key="test-key-mocked-no-credits-needed",
size="1024x1536",
quality="high",
response_format="b64_json",
n=1,
)
assert captured["url"] is not None
assert "api.aimlapi.com" in captured["url"]
assert "/v1/images/generations" in captured["url"]
body = captured["json"]
assert body["model"] == "openai/gpt-image-2"
assert body["prompt"] == "A T-Rex relaxing on a beach"
assert body["size"] == "1024x1536"
assert body["quality"] == "high"
assert body["response_format"] == "b64_json"
assert body["n"] == 1
assert "image_size" not in body
assert "num_images" not in body
assert "output_format" not in body
@pytest.mark.asyncio
async def test_azure_image_generation_request_body():
"""Azure deployment URL selects the model; JSON body omits ``model`` (#26316)."""
from litellm import aimage_generation
test_dir = os.path.dirname(__file__)
expected_path = os.path.join(test_dir, "request_payloads", "azure_gpt_image_1.json")
with open(expected_path, "r") as f:
expected_body = json.load(f)
with patch(
"litellm.llms.custom_httpx.http_handler.AsyncHTTPHandler.post",
new_callable=AsyncMock,
) as mock_post:
mock_post.side_effect = Exception("test")
with pytest.raises(litellm.APIConnectionError):
await aimage_generation(
model="azure/gpt-image-1",
prompt="test prompt",
api_base="https://example.azure.com",
api_key="test-key",
api_version="2025-04-01-preview",
)
mock_post.assert_called_once()
call_args = mock_post.call_args
request_json = call_args.kwargs.get("json", {})
assert request_json == expected_body
Reference in a new issue View git blame Copy permalink