litellm-mirror/litellm/tests/test_image_generation.py

# What this tests?
## This tests the litellm support for the openai /generations endpoint

import logging
import os
import sys
import traceback

from dotenv import load_dotenv
from openai.types.image import Image

logging.basicConfig(level=logging.DEBUG)
load_dotenv()
import asyncio
import os

sys.path.insert(
    0, os.path.abspath("../..")
)  # Adds the parent directory to the system path
import pytest

import litellm


def test_image_generation_openai():
    try:
        litellm.set_verbose = True
        response = litellm.image_generation(
            prompt="A cute baby sea otter", model="dall-e-3"
        )
        print(f"response: {response}")
        assert len(response.data) > 0
    except litellm.RateLimitError as e:
        pass
    except litellm.ContentPolicyViolationError:
        pass  # OpenAI randomly raises these errors - skip when they occur
    except Exception as e:
        if "Connection error" in str(e):
            pass
        pytest.fail(f"An exception occurred - {str(e)}")


# test_image_generation_openai()


@pytest.mark.parametrize(
    "sync_mode",
    [
        True,
    ],  # False
)  #
@pytest.mark.asyncio
@pytest.mark.flaky(retries=3, delay=1)
async def test_image_generation_azure(sync_mode):
    try:
        if sync_mode:
            response = litellm.image_generation(
                prompt="A cute baby sea otter",
                model="azure/",
                api_version="2023-06-01-preview",
            )
        else:
            response = await litellm.aimage_generation(
                prompt="A cute baby sea otter",
                model="azure/",
                api_version="2023-06-01-preview",
            )
        print(f"response: {response}")
        assert len(response.data) > 0
    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
        if "Connection error" in str(e):
            pass
        else:
            pytest.fail(f"An exception occurred - {str(e)}")


# test_image_generation_azure()


@pytest.mark.flaky(retries=3, delay=1)
def test_image_generation_azure_dall_e_3():
    try:
        litellm.set_verbose = True
        response = litellm.image_generation(
            prompt="A cute baby sea otter",
            model="azure/dall-e-3-test",
            api_version="2023-12-01-preview",
            api_base=os.getenv("AZURE_SWEDEN_API_BASE"),
            api_key=os.getenv("AZURE_SWEDEN_API_KEY"),
        )
        print(f"response: {response}")
        assert len(response.data) > 0
    except litellm.InternalServerError as e:
        pass
    except litellm.ContentPolicyViolationError:
        pass  # OpenAI 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
        if "Connection error" in str(e):
            pass
        else:
            pytest.fail(f"An exception occurred - {str(e)}")


# test_image_generation_azure_dall_e_3()
@pytest.mark.asyncio
async def test_async_image_generation_openai():
    try:
        response = litellm.image_generation(
            prompt="A cute baby sea otter", model="dall-e-3"
        )
        print(f"response: {response}")
        assert len(response.data) > 0
    except litellm.APIError:
        pass
    except litellm.RateLimitError as e:
        pass
    except litellm.ContentPolicyViolationError:
        pass  # openai randomly raises these errors - skip when they occur
    except Exception as e:
        if "Connection error" in str(e):
            pass
        pytest.fail(f"An exception occurred - {str(e)}")


# asyncio.run(test_async_image_generation_openai())


@pytest.mark.asyncio
async def test_async_image_generation_azure():
    try:
        response = await litellm.aimage_generation(
            prompt="A cute baby sea otter",
            model="azure/dall-e-3-test",
            api_version="2023-09-01-preview",
        )
        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
        if "Connection error" in str(e):
            pass
        else:
            pytest.fail(f"An exception occurred - {str(e)}")


def test_image_generation_bedrock():
    try:
        litellm.set_verbose = True
        response = litellm.image_generation(
            prompt="A cute baby sea otter",
            model="bedrock/stability.stable-diffusion-xl-v1",
            aws_region_name="us-west-2",
        )

        print(f"response: {response}")
        from openai.types.images_response import ImagesResponse

        ImagesResponse.model_validate(response.model_dump())
    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_aimage_generation_bedrock_with_optional_params():
    try:
        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)}")


from openai.types.image import Image


@pytest.mark.parametrize("sync_mode", [True, False])
@pytest.mark.asyncio
async def test_aimage_generation_vertex_ai(sync_mode):
    from test_amazing_vertex_completion import load_vertex_ai_credentials

    litellm.set_verbose = True

    load_vertex_ai_credentials()
    data = {
        "prompt": "An olympic size swimming pool",
        "model": "vertex_ai/imagegeneration@006",
        "vertex_ai_project": "adroit-crow-413218",
        "vertex_ai_location": "us-central1",
        "n": 1,
    }
    try:
        if sync_mode:
            response = litellm.image_generation(**data)
        else:
            response = await litellm.aimage_generation(**data)
        assert response.data is not None
        assert len(response.data) > 0

        for d in response.data:
            assert isinstance(d, Image)
            print("data in response.data", d)
            assert d.b64_json is not None
    except litellm.ServiceUnavailableError as e:
        pass
    except litellm.RateLimitError as e:
        pass
    except litellm.InternalServerError 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)}")