scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-09-19 00:01:29 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 88

test: remove fully commented-out test files that collect no tests

Browse source 
Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
This commit is contained in:
mateo 2026-09-17 20:05:27 +00:00
parent decbb96382
commit 349223fd8b
14 changed files with 0 additions and 2370 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

87
tests/image_gen_tests/test_image_variation.py
View file

@ -1,87 +0,0 @@
# What this tests?
## This tests the litellm support for the openai /generations endpoint
import logging
import traceback
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
import logging
from litellm._logging import verbose_logger
from io import BytesIO
from PIL import Image as PILImage
verbose_logger.setLevel(logging.DEBUG)
@pytest.fixture
def image_url():
# DALL-E 2 image variations require a square PNG (less than 4MB)
# Generate a 1024x1024 square PNG programmatically to avoid network dependency
# and the non-square aspect ratio of the old LiteLLM logo URL
img = PILImage.new("RGBA", (1024, 1024), color=(128, 128, 128, 255))
image_file = BytesIO()
img.save(image_file, format="PNG")
image_file.seek(0)
# openai>=2.24.0 requires BytesIO to have .name for MIME type detection in multipart uploads
image_file.name = "litellm_logo.png"
return image_file
# Commented out: OpenAI /images/variations endpoint deprecated (DALL-E 2 shutdown May 12, 2026)
# def test_openai_image_variation_openai_sdk(image_url):
# from openai import OpenAI
#
# client = OpenAI()
# response = client.images.create_variation(image=image_url, n=2, size="1024x1024")
# print(response)
#
#
# @pytest.mark.parametrize("sync_mode", [True, False])
# @pytest.mark.asyncio
# async def test_openai_image_variation_litellm_sdk(image_url, sync_mode):
# from litellm import image_variation, aimage_variation
#
# if sync_mode:
# image_variation(image=image_url, n=2, size="1024x1024")
# else:
# await aimage_variation(image=image_url, n=2, size="1024x1024")
#
#
# def test_topaz_image_variation(image_url):
# from litellm import image_variation, aimage_variation
# from litellm.llms.custom_httpx.http_handler import HTTPHandler
# from unittest.mock import patch
#
# client = HTTPHandler()
# with patch.object(client, "post") as mock_post:
# try:
# image_variation(
# model="topaz/Standard V2",
# image=image_url,
# n=2,
# size="1024x1024",
# client=client,
# )
# except Exception as e:
# print(e)
# mock_post.assert_called_once()
def test_image_variation_placeholder():
"""Placeholder: variation tests commented out - OpenAI /images/variations deprecated (DALL-E 2 shutdown May 12, 2026)."""
pass

128
tests/local_testing/test_azure_perf.py
View file

@ -1,128 +0,0 @@
# #### What this tests ####
# # This adds perf testing to the router, to ensure it's never > 50ms slower than the azure-openai sdk.
# import sys, os, time, inspect, asyncio, traceback
# from datetime import datetime
# import pytest
# sys.path.insert(0, os.path.abspath("../.."))
# import openai, litellm, uuid
# from openai import AsyncAzureOpenAI
# client = AsyncAzureOpenAI(
# api_key=os.getenv("AZURE_AI_API_KEY"),
# azure_endpoint=os.getenv("AZURE_AI_API_BASE"), # type: ignore
# api_version=os.getenv("AZURE_API_VERSION"),
# )
# model_list = [
# {
# "model_name": "azure-test",
# "litellm_params": {
# "model": "azure/gpt-4.1-mini",
# "api_key": os.getenv("AZURE_AI_API_KEY"),
# "api_base": os.getenv("AZURE_AI_API_BASE"),
# "api_version": os.getenv("AZURE_API_VERSION"),
# },
# }
# ]
# router = litellm.Router(model_list=model_list) # type: ignore
# async def _openai_completion():
# try:
# start_time = time.time()
# response = await client.chat.completions.create(
# model="chatgpt-v-3",
# messages=[{"role": "user", "content": f"This is a test: {uuid.uuid4()}"}],
# stream=True,
# )
# time_to_first_token = None
# first_token_ts = None
# init_chunk = None
# async for chunk in response:
# if (
# time_to_first_token is None
# and len(chunk.choices) > 0
# and chunk.choices[0].delta.content is not None
# ):
# first_token_ts = time.time()
# time_to_first_token = first_token_ts - start_time
# init_chunk = chunk
# end_time = time.time()
# print(
# "OpenAI Call: ",
# init_chunk,
# start_time,
# first_token_ts,
# time_to_first_token,
# end_time,
# )
# return time_to_first_token
# except Exception as e:
# print(e)
# return None
# async def _router_completion():
# try:
# start_time = time.time()
# response = await router.acompletion(
# model="azure-test",
# messages=[{"role": "user", "content": f"This is a test: {uuid.uuid4()}"}],
# stream=True,
# )
# time_to_first_token = None
# first_token_ts = None
# init_chunk = None
# async for chunk in response:
# if (
# time_to_first_token is None
# and len(chunk.choices) > 0
# and chunk.choices[0].delta.content is not None
# ):
# first_token_ts = time.time()
# time_to_first_token = first_token_ts - start_time
# init_chunk = chunk
# end_time = time.time()
# print(
# "Router Call: ",
# init_chunk,
# start_time,
# first_token_ts,
# time_to_first_token,
# end_time - first_token_ts,
# )
# return time_to_first_token
# except Exception as e:
# print(e)
# return None
# async def test_azure_completion_streaming():
# """
# Test azure streaming call - measure on time to first (non-null) token.
# """
# n = 3 # Number of concurrent tasks
# ## OPENAI AVG. TIME
# tasks = [_openai_completion() for _ in range(n)]
# chat_completions = await asyncio.gather(*tasks)
# successful_completions = [c for c in chat_completions if c is not None]
# total_time = 0
# for item in successful_completions:
# total_time += item
# avg_openai_time = total_time / 3
# ## ROUTER AVG. TIME
# tasks = [_router_completion() for _ in range(n)]
# chat_completions = await asyncio.gather(*tasks)
# successful_completions = [c for c in chat_completions if c is not None]
# total_time = 0
# for item in successful_completions:
# total_time += item
# avg_router_time = total_time / 3
# ## COMPARE
# print(f"avg_router_time: {avg_router_time}; avg_openai_time: {avg_openai_time}")
# assert avg_router_time < avg_openai_time + 0.5
# # asyncio.run(test_azure_completion_streaming())

130
tests/local_testing/test_budget_manager.py
View file

@ -1,130 +0,0 @@
# #### What this tests ####
# # This tests calling batch_completions by running 100 messages together
# import sys, os, json
# import traceback
# import pytest
# sys.path.insert(
# 0, os.path.abspath("../..")
# ) # Adds the parent directory to the system path
# import litellm
# litellm.set_verbose = True
# from litellm import completion, BudgetManager
# budget_manager = BudgetManager(project_name="test_project", client_type="hosted")
# ## Scenario 1: User budget enough to make call
# def test_user_budget_enough():
# try:
# user = "1234"
# # create a budget for a user
# budget_manager.create_budget(total_budget=10, user=user, duration="daily")
# # check if a given call can be made
# data = {
# "model": "gpt-3.5-turbo",
# "messages": [{"role": "user", "content": "Hey, how's it going?"}]
# }
# if budget_manager.get_current_cost(user=user) <= budget_manager.get_total_budget(user):
# response = completion(**data)
# print(budget_manager.update_cost(completion_obj=response, user=user))
# else:
# response = "Sorry - no budget!"
# print(f"response: {response}")
# except Exception as e:
# pytest.fail(f"An error occurred - {str(e)}")
# ## Scenario 2: User budget not enough to make call
# def test_user_budget_not_enough():
# try:
# user = "12345"
# # create a budget for a user
# budget_manager.create_budget(total_budget=0, user=user, duration="daily")
# # check if a given call can be made
# data = {
# "model": "gpt-3.5-turbo",
# "messages": [{"role": "user", "content": "Hey, how's it going?"}]
# }
# model = data["model"]
# messages = data["messages"]
# if budget_manager.get_current_cost(user=user) < budget_manager.get_total_budget(user=user):
# response = completion(**data)
# print(budget_manager.update_cost(completion_obj=response, user=user))
# else:
# response = "Sorry - no budget!"
# print(f"response: {response}")
# except Exception:
# pytest.fail(f"An error occurred")
# ## Scenario 3: Saving budget to client
# def test_save_user_budget():
# try:
# response = budget_manager.save_data()
# if response["status"] == "error":
# raise Exception(f"An error occurred - {json.dumps(response)}")
# print(response)
# except Exception as e:
# pytest.fail(f"An error occurred: {str(e)}")
# test_save_user_budget()
# ## Scenario 4: Getting list of users
# def test_get_users():
# try:
# response = budget_manager.get_users()
# print(response)
# except Exception:
# pytest.fail(f"An error occurred")
# ## Scenario 5: Reset budget at the end of duration
# def test_reset_on_duration():
# try:
# # First, set a short duration budget for a user
# user = "123456"
# budget_manager.create_budget(total_budget=10, user=user, duration="daily")
# # Use some of the budget
# data = {
# "model": "gpt-3.5-turbo",
# "messages": [{"role": "user", "content": "Hello!"}]
# }
# if budget_manager.get_current_cost(user=user) <= budget_manager.get_total_budget(user=user):
# response = litellm.completion(**data)
# print(budget_manager.update_cost(completion_obj=response, user=user))
# assert budget_manager.get_current_cost(user) > 0, f"Test setup failed: Budget did not decrease after completion"
# # Now, we need to simulate the passing of time. Since we don't want our tests to actually take days, we're going
# # to cheat a little -- we'll manually adjust the "created_at" time so it seems like a day has passed.
# # In a real-world testing scenario, we might instead use something like the `freezegun` library to mock the system time.
# one_day_in_seconds = 24 * 60 * 60
# budget_manager.user_dict[user]["last_updated_at"] -= one_day_in_seconds
# # Now the duration should have expired, so our budget should reset
# budget_manager.update_budget_all_users()
# # Make sure the budget was actually reset
# assert budget_manager.get_current_cost(user) == 0, "Budget didn't reset after duration expired"
# except Exception as e:
# pytest.fail(f"An error occurred - {str(e)}")
# ## Scenario 6: passing in text:
# def test_input_text_on_completion():
# try:
# user = "12345"
# budget_manager.create_budget(total_budget=10, user=user, duration="daily")
# input_text = "hello world"
# output_text = "it's a sunny day in san francisco"
# model = "gpt-3.5-turbo"
# budget_manager.update_cost(user=user, model=model, input_text=input_text, output_text=output_text)
# print(budget_manager.get_current_cost(user))
# except Exception as e:
# pytest.fail(f"An error occurred - {str(e)}")
# test_input_text_on_completion()

124
tests/local_testing/test_class.py
View file

@ -1,124 +0,0 @@
# # #### What this tests ####
# # # This tests the LiteLLM Class
# import sys, os
# import traceback
# import pytest
# sys.path.insert(
# 0, os.path.abspath("../..")
# ) # Adds the parent directory to the system path
# import litellm
# import asyncio
# # litellm.set_verbose = True
# # from litellm import Router
# import instructor
# from litellm import completion
# from pydantic import BaseModel
# class User(BaseModel):
# name: str
# age: int
# client = instructor.from_litellm(completion)
# litellm.set_verbose = True
# resp = client.chat.completions.create(
# model="gpt-3.5-turbo",
# max_tokens=1024,
# messages=[
# {
# "role": "user",
# "content": "Extract Jason is 25 years old.",
# }
# ],
# response_model=User,
# num_retries=10,
# )
# assert isinstance(resp, User)
# assert resp.name == "Jason"
# assert resp.age == 25
# # from pydantic import BaseModel
# # # This enables response_model keyword
# # # from client.chat.completions.create
# # client = instructor.patch(
# # Router(
# # model_list=[
# # {
# # "model_name": "gpt-3.5-turbo", # openai model name
# # "litellm_params": { # params for litellm completion/embedding call
# # "model": "azure/gpt-4.1-mini",
# # "api_key": os.getenv("AZURE_AI_API_KEY"),
# # "api_version": os.getenv("AZURE_API_VERSION"),
# # "api_base": os.getenv("AZURE_AI_API_BASE"),
# # },
# # }
# # ]
# # )
# # )
# # class UserDetail(BaseModel):
# # name: str
# # age: int
# # user = client.chat.completions.create(
# # model="gpt-3.5-turbo",
# # response_model=UserDetail,
# # messages=[
# # {"role": "user", "content": "Extract Jason is 25 years old"},
# # ],
# # )
# # assert isinstance(user, UserDetail)
# # assert user.name == "Jason"
# # assert user.age == 25
# # print(f"user: {user}")
# # # import instructor
# # # from openai import AsyncOpenAI
# # aclient = instructor.apatch(
# # Router(
# # model_list=[
# # {
# # "model_name": "gpt-3.5-turbo", # openai model name
# # "litellm_params": { # params for litellm completion/embedding call
# # "model": "azure/gpt-4.1-mini",
# # "api_key": os.getenv("AZURE_AI_API_KEY"),
# # "api_version": os.getenv("AZURE_API_VERSION"),
# # "api_base": os.getenv("AZURE_AI_API_BASE"),
# # },
# # }
# # ],
# # default_litellm_params={"acompletion": True},
# # )
# # )
# # class UserExtract(BaseModel):
# # name: str
# # age: int
# # async def main():
# # model = await aclient.chat.completions.create(
# # model="gpt-3.5-turbo",
# # response_model=UserExtract,
# # messages=[
# # {"role": "user", "content": "Extract jason is 25 years old"},
# # ],
# # )
# # print(f"model: {model}")
# # asyncio.run(main())

90
tests/local_testing/test_langchain_ChatLiteLLM.py
View file

@ -1,90 +0,0 @@
# import os
# import sys, os
# import traceback
# from dotenv import load_dotenv
# load_dotenv()
# import os, io
# sys.path.insert(
# 0, os.path.abspath("../..")
# ) # Adds the parent directory to the system path
# import pytest
# import litellm
# from litellm import embedding, completion, text_completion, completion_cost
# from langchain.chat_models import ChatLiteLLM
# from langchain.prompts.chat import (
# ChatPromptTemplate,
# SystemMessagePromptTemplate,
# AIMessagePromptTemplate,
# HumanMessagePromptTemplate,
# )
# from langchain.schema import AIMessage, HumanMessage, SystemMessage
# def test_chat_gpt():
# try:
# chat = ChatLiteLLM(model="gpt-3.5-turbo", max_tokens=10)
# messages = [
# HumanMessage(
# content="what model are you"
# )
# ]
# resp = chat(messages)
# print(resp)
# except Exception as e:
# pytest.fail(f"Error occurred: {e}")
# # test_chat_gpt()
# def test_claude():
# try:
# chat = ChatLiteLLM(model="claude-2", max_tokens=10)
# messages = [
# HumanMessage(
# content="what model are you"
# )
# ]
# resp = chat(messages)
# print(resp)
# except Exception as e:
# pytest.fail(f"Error occurred: {e}")
# # test_claude()
# # def test_openai_with_params():
# # try:
# # api_key = os.environ["OPENAI_API_KEY"]
# # os.environ.pop("OPENAI_API_KEY")
# # print("testing openai with params")
# # llm = ChatLiteLLM(
# # model="gpt-3.5-turbo",
# # openai_api_key=api_key,
# # # Prefer using None which is the default value, endpoint could be empty string
# # openai_api_base= None,
# # max_tokens=20,
# # temperature=0.5,
# # request_timeout=10,
# # model_kwargs={
# # "frequency_penalty": 0,
# # "presence_penalty": 0,
# # },
# # verbose=True,
# # max_retries=0,
# # )
# # messages = [
# # HumanMessage(
# # content="what model are you"
# # )
# # ]
# # resp = llm(messages)
# # print(resp)
# # except Exception as e:
# # pytest.fail(f"Error occurred: {e}")
# # test_openai_with_params()

94
tests/local_testing/test_load_test_router_s3.py
View file

@ -1,94 +0,0 @@
# import sys, os
# import traceback
# from dotenv import load_dotenv
# import copy
# load_dotenv()
# sys.path.insert(
# 0, os.path.abspath("../..")
# ) # Adds the parent directory to the system path
# import asyncio
# from litellm import Router, Timeout
# import time
# from litellm.caching.caching import Cache
# import litellm
# litellm.cache = Cache(
# type="s3", s3_bucket_name="litellm-my-test-bucket-2", s3_region_name="us-west-2"
# )
# ### Test calling router with s3 Cache
# async def call_acompletion(semaphore, router: Router, input_data):
# async with semaphore:
# try:
# # Use asyncio.wait_for to set a timeout for the task
# response = await router.acompletion(**input_data)
# # Handle the response as needed
# print(response)
# return response
# except Timeout:
# print(f"Task timed out: {input_data}")
# return None # You may choose to return something else or raise an exception
# async def main():
# # Initialize the Router
# model_list = [
# {
# "model_name": "gpt-3.5-turbo",
# "litellm_params": {
# "model": "gpt-3.5-turbo",
# "api_key": os.getenv("OPENAI_API_KEY"),
# },
# },
# {
# "model_name": "gpt-3.5-turbo",
# "litellm_params": {
# "model": "azure/gpt-4.1-mini",
# "api_key": os.getenv("AZURE_API_KEY"),
# "api_base": os.getenv("AZURE_API_BASE"),
# "api_version": os.getenv("AZURE_API_VERSION"),
# },
# },
# ]
# router = Router(model_list=model_list, num_retries=3, timeout=10)
# # Create a semaphore with a capacity of 100
# semaphore = asyncio.Semaphore(100)
# # List to hold all task references
# tasks = []
# start_time_all_tasks = time.time()
# # Launch 1000 tasks
# for _ in range(500):
# task = asyncio.create_task(
# call_acompletion(
# semaphore,
# router,
# {
# "model": "gpt-3.5-turbo",
# "messages": [{"role": "user", "content": "Hey, how's it going?"}],
# },
# )
# )
# tasks.append(task)
# # Wait for all tasks to complete
# responses = await asyncio.gather(*tasks)
# # Process responses as needed
# # Record the end time for all tasks
# end_time_all_tasks = time.time()
# # Calculate the total time for all tasks
# total_time_all_tasks = end_time_all_tasks - start_time_all_tasks
# print(f"Total time for all tasks: {total_time_all_tasks} seconds")
# # Calculate the average time per response
# average_time_per_response = total_time_all_tasks / len(responses)
# print(f"Average time per response: {average_time_per_response} seconds")
# print(f"NUMBER OF COMPLETED TASKS: {len(responses)}")
# # Run the main function
# asyncio.run(main())

86
tests/local_testing/test_loadtest_router.py
View file

@ -1,86 +0,0 @@
# import sys, os
# import traceback
# from dotenv import load_dotenv
# import copy
# load_dotenv()
# sys.path.insert(
# 0, os.path.abspath("../..")
# ) # Adds the parent directory to the system path
# import asyncio
# from litellm import Router, Timeout
# import time
# async def call_acompletion(semaphore, router: Router, input_data):
# async with semaphore:
# try:
# # Use asyncio.wait_for to set a timeout for the task
# response = await router.acompletion(**input_data)
# # Handle the response as needed
# print(response)
# return response
# except Timeout:
# print(f"Task timed out: {input_data}")
# return None # You may choose to return something else or raise an exception
# async def main():
# # Initialize the Router
# model_list = [
# {
# "model_name": "gpt-3.5-turbo",
# "litellm_params": {
# "model": "gpt-3.5-turbo",
# "api_key": os.getenv("OPENAI_API_KEY"),
# },
# },
# {
# "model_name": "gpt-3.5-turbo",
# "litellm_params": {
# "model": "azure/gpt-4.1-mini",
# "api_key": os.getenv("AZURE_AI_API_KEY"),
# "api_base": os.getenv("AZURE_AI_API_BASE"),
# "api_version": os.getenv("AZURE_API_VERSION"),
# },
# },
# ]
# router = Router(model_list=model_list, num_retries=3, timeout=10)
# # Create a semaphore with a capacity of 100
# semaphore = asyncio.Semaphore(100)
# # List to hold all task references
# tasks = []
# start_time_all_tasks = time.time()
# # Launch 1000 tasks
# for _ in range(500):
# task = asyncio.create_task(
# call_acompletion(
# semaphore,
# router,
# {
# "model": "gpt-3.5-turbo",
# "messages": [{"role": "user", "content": "Hey, how's it going?"}],
# },
# )
# )
# tasks.append(task)
# # Wait for all tasks to complete
# responses = await asyncio.gather(*tasks)
# # Process responses as needed
# # Record the end time for all tasks
# end_time_all_tasks = time.time()
# # Calculate the total time for all tasks
# total_time_all_tasks = end_time_all_tasks - start_time_all_tasks
# print(f"Total time for all tasks: {total_time_all_tasks} seconds")
# # Calculate the average time per response
# average_time_per_response = total_time_all_tasks / len(responses)
# print(f"Average time per response: {average_time_per_response} seconds")
# print(f"NUMBER OF COMPLETED TASKS: {len(responses)}")
# # Run the main function
# asyncio.run(main())

382
tests/local_testing/test_logging.py
View file

@ -1,382 +0,0 @@
# #### What this tests ####
# # This tests error logging (with custom user functions) for the raw `completion` + `embedding` endpoints
# # Test Scenarios (test across completion, streaming, embedding)
# ## 1: Pre-API-Call
# ## 2: Post-API-Call
# ## 3: On LiteLLM Call success
# ## 4: On LiteLLM Call failure
# import sys, os, io
# import traceback, logging
# import pytest
# import dotenv
# dotenv.load_dotenv()
# # Create logger
# logger = logging.getLogger(__name__)
# logger.setLevel(logging.DEBUG)
# # Create a stream handler
# stream_handler = logging.StreamHandler(sys.stdout)
# logger.addHandler(stream_handler)
# # Create a function to log information
# def logger_fn(message):
# logger.info(message)
# sys.path.insert(
# 0, os.path.abspath("../..")
# ) # Adds the parent directory to the system path
# import litellm
# from litellm import embedding, completion
# from openai.error import AuthenticationError
# litellm.set_verbose = True
# score = 0
# user_message = "Hello, how are you?"
# messages = [{"content": user_message, "role": "user"}]
# # 1. On Call Success
# # normal completion
# # test on openai completion call
# def test_logging_success_completion():
# global score
# try:
# # Redirect stdout
# old_stdout = sys.stdout
# sys.stdout = new_stdout = io.StringIO()
# response = completion(model="gpt-3.5-turbo", messages=messages)
# # Restore stdout
# sys.stdout = old_stdout
# output = new_stdout.getvalue().strip()
# if "Logging Details Pre-API Call" not in output:
# raise Exception("Required log message not found!")
# elif "Logging Details Post-API Call" not in output:
# raise Exception("Required log message not found!")
# elif "Logging Details LiteLLM-Success Call" not in output:
# raise Exception("Required log message not found!")
# score += 1
# except Exception as e:
# pytest.fail(f"Error occurred: {e}")
# pass
# # ## test on non-openai completion call
# # def test_logging_success_completion_non_openai():
# # global score
# # try:
# # # Redirect stdout
# # old_stdout = sys.stdout
# # sys.stdout = new_stdout = io.StringIO()
# # response = completion(model="claude-3-5-haiku-20241022", messages=messages)
# # # Restore stdout
# # sys.stdout = old_stdout
# # output = new_stdout.getvalue().strip()
# # if "Logging Details Pre-API Call" not in output:
# # raise Exception("Required log message not found!")
# # elif "Logging Details Post-API Call" not in output:
# # raise Exception("Required log message not found!")
# # elif "Logging Details LiteLLM-Success Call" not in output:
# # raise Exception("Required log message not found!")
# # score += 1
# # except Exception as e:
# # pytest.fail(f"Error occurred: {e}")
# # pass
# # streaming completion
# ## test on openai completion call
# def test_logging_success_streaming_openai():
# global score
# try:
# # litellm.set_verbose = False
# def custom_callback(
# kwargs, # kwargs to completion
# completion_response, # response from completion
# start_time, end_time # start/end time
# ):
# if "complete_streaming_response" in kwargs:
# print(f"Complete Streaming Response: {kwargs['complete_streaming_response']}")
# # Assign the custom callback function
# litellm.success_callback = [custom_callback]
# # Redirect stdout
# old_stdout = sys.stdout
# sys.stdout = new_stdout = io.StringIO()
# response = completion(model="gpt-3.5-turbo", messages=messages, stream=True)
# for chunk in response:
# pass
# # Restore stdout
# sys.stdout = old_stdout
# output = new_stdout.getvalue().strip()
# if "Logging Details Pre-API Call" not in output:
# raise Exception("Required log message not found!")
# elif "Logging Details Post-API Call" not in output:
# raise Exception("Required log message not found!")
# elif "Logging Details LiteLLM-Success Call" not in output:
# raise Exception("Required log message not found!")
# elif "Complete Streaming Response:" not in output:
# raise Exception("Required log message not found!")
# score += 1
# except Exception as e:
# pytest.fail(f"Error occurred: {e}")
# pass
# # test_logging_success_streaming_openai()
# ## test on non-openai completion call
# def test_logging_success_streaming_non_openai():
# global score
# try:
# # litellm.set_verbose = False
# def custom_callback(
# kwargs, # kwargs to completion
# completion_response, # response from completion
# start_time, end_time # start/end time
# ):
# # print(f"streaming response: {completion_response}")
# if "complete_streaming_response" in kwargs:
# print(f"Complete Streaming Response: {kwargs['complete_streaming_response']}")
# # Assign the custom callback function
# litellm.success_callback = [custom_callback]
# # Redirect stdout
# old_stdout = sys.stdout
# sys.stdout = new_stdout = io.StringIO()
# response = completion(model="claude-3-5-haiku-20241022", messages=messages, stream=True)
# for idx, chunk in enumerate(response):
# pass
# # Restore stdout
# sys.stdout = old_stdout
# output = new_stdout.getvalue().strip()
# if "Logging Details Pre-API Call" not in output:
# raise Exception("Required log message not found!")
# elif "Logging Details Post-API Call" not in output:
# raise Exception("Required log message not found!")
# elif "Logging Details LiteLLM-Success Call" not in output:
# raise Exception("Required log message not found!")
# elif "Complete Streaming Response:" not in output:
# raise Exception(f"Required log message not found! {output}")
# score += 1
# except Exception as e:
# pytest.fail(f"Error occurred: {e}")
# pass
# # test_logging_success_streaming_non_openai()
# # embedding
# def test_logging_success_embedding_openai():
# try:
# # Redirect stdout
# old_stdout = sys.stdout
# sys.stdout = new_stdout = io.StringIO()
# response = embedding(model="text-embedding-ada-002", input=["good morning from litellm"])
# # Restore stdout
# sys.stdout = old_stdout
# output = new_stdout.getvalue().strip()
# if "Logging Details Pre-API Call" not in output:
# raise Exception("Required log message not found!")
# elif "Logging Details Post-API Call" not in output:
# raise Exception("Required log message not found!")
# elif "Logging Details LiteLLM-Success Call" not in output:
# raise Exception("Required log message not found!")
# except Exception as e:
# pytest.fail(f"Error occurred: {e}")
# # ## 2. On LiteLLM Call failure
# # ## TEST BAD KEY
# # # normal completion
# # ## test on openai completion call
# # try:
# # temporary_oai_key = os.environ["OPENAI_API_KEY"]
# # os.environ["OPENAI_API_KEY"] = "bad-key"
# # temporary_anthropic_key = os.environ["ANTHROPIC_API_KEY"]
# # os.environ["ANTHROPIC_API_KEY"] = "bad-key"
# # # Redirect stdout
# # old_stdout = sys.stdout
# # sys.stdout = new_stdout = io.StringIO()
# # try:
# # response = completion(model="gpt-3.5-turbo", messages=messages)
# # except AuthenticationError:
# # print(f"raised auth error")
# # pass
# # # Restore stdout
# # sys.stdout = old_stdout
# # output = new_stdout.getvalue().strip()
# # print(output)
# # if "Logging Details Pre-API Call" not in output:
# # raise Exception("Required log message not found!")
# # elif "Logging Details Post-API Call" not in output:
# # raise Exception("Required log message not found!")
# # elif "Logging Details LiteLLM-Failure Call" not in output:
# # raise Exception("Required log message not found!")
# # os.environ["OPENAI_API_KEY"] = temporary_oai_key
# # os.environ["ANTHROPIC_API_KEY"] = temporary_anthropic_key
# # score += 1
# # except Exception as e:
# # print(f"exception type: {type(e).__name__}")
# # pytest.fail(f"Error occurred: {e}")
# # pass
# # ## test on non-openai completion call
# # try:
# # temporary_oai_key = os.environ["OPENAI_API_KEY"]
# # os.environ["OPENAI_API_KEY"] = "bad-key"
# # temporary_anthropic_key = os.environ["ANTHROPIC_API_KEY"]
# # os.environ["ANTHROPIC_API_KEY"] = "bad-key"
# # # Redirect stdout
# # old_stdout = sys.stdout
# # sys.stdout = new_stdout = io.StringIO()
# # try:
# # response = completion(model="claude-3-5-haiku-20241022", messages=messages)
# # except AuthenticationError:
# # pass
# # if "Logging Details Pre-API Call" not in output:
# # raise Exception("Required log message not found!")
# # elif "Logging Details Post-API Call" not in output:
# # raise Exception("Required log message not found!")
# # elif "Logging Details LiteLLM-Failure Call" not in output:
# # raise Exception("Required log message not found!")
# # os.environ["OPENAI_API_KEY"] = temporary_oai_key
# # os.environ["ANTHROPIC_API_KEY"] = temporary_anthropic_key
# # score += 1
# # except Exception as e:
# # print(f"exception type: {type(e).__name__}")
# # # Restore stdout
# # sys.stdout = old_stdout
# # output = new_stdout.getvalue().strip()
# # print(output)
# # pytest.fail(f"Error occurred: {e}")
# # # streaming completion
# # ## test on openai completion call
# # try:
# # temporary_oai_key = os.environ["OPENAI_API_KEY"]
# # os.environ["OPENAI_API_KEY"] = "bad-key"
# # temporary_anthropic_key = os.environ["ANTHROPIC_API_KEY"]
# # os.environ["ANTHROPIC_API_KEY"] = "bad-key"
# # # Redirect stdout
# # old_stdout = sys.stdout
# # sys.stdout = new_stdout = io.StringIO()
# # try:
# # response = completion(model="gpt-3.5-turbo", messages=messages)
# # except AuthenticationError:
# # pass
# # # Restore stdout
# # sys.stdout = old_stdout
# # output = new_stdout.getvalue().strip()
# # print(output)
# # if "Logging Details Pre-API Call" not in output:
# # raise Exception("Required log message not found!")
# # elif "Logging Details Post-API Call" not in output:
# # raise Exception("Required log message not found!")
# # elif "Logging Details LiteLLM-Failure Call" not in output:
# # raise Exception("Required log message not found!")
# # os.environ["OPENAI_API_KEY"] = temporary_oai_key
# # os.environ["ANTHROPIC_API_KEY"] = temporary_anthropic_key
# # score += 1
# # except Exception as e:
# # print(f"exception type: {type(e).__name__}")
# # pytest.fail(f"Error occurred: {e}")
# # ## test on non-openai completion call
# # try:
# # temporary_oai_key = os.environ["OPENAI_API_KEY"]
# # os.environ["OPENAI_API_KEY"] = "bad-key"
# # temporary_anthropic_key = os.environ["ANTHROPIC_API_KEY"]
# # os.environ["ANTHROPIC_API_KEY"] = "bad-key"
# # # Redirect stdout
# # old_stdout = sys.stdout
# # sys.stdout = new_stdout = io.StringIO()
# # try:
# # response = completion(model="claude-3-5-haiku-20241022", messages=messages)
# # except AuthenticationError:
# # pass
# # # Restore stdout
# # sys.stdout = old_stdout
# # output = new_stdout.getvalue().strip()
# # print(output)
# # if "Logging Details Pre-API Call" not in output:
# # raise Exception("Required log message not found!")
# # elif "Logging Details Post-API Call" not in output:
# # raise Exception("Required log message not found!")
# # elif "Logging Details LiteLLM-Failure Call" not in output:
# # raise Exception("Required log message not found!")
# # score += 1
# # except Exception as e:
# # print(f"exception type: {type(e).__name__}")
# # pytest.fail(f"Error occurred: {e}")
# # # embedding
# # try:
# # temporary_oai_key = os.environ["OPENAI_API_KEY"]
# # os.environ["OPENAI_API_KEY"] = "bad-key"
# # temporary_anthropic_key = os.environ["ANTHROPIC_API_KEY"]
# # os.environ["ANTHROPIC_API_KEY"] = "bad-key"
# # # Redirect stdout
# # old_stdout = sys.stdout
# # sys.stdout = new_stdout = io.StringIO()
# # try:
# # response = embedding(model="text-embedding-ada-002", input=["good morning from litellm"])
# # except AuthenticationError:
# # pass
# # # Restore stdout
# # sys.stdout = old_stdout
# # output = new_stdout.getvalue().strip()
# # print(output)
# # if "Logging Details Pre-API Call" not in output:
# # raise Exception("Required log message not found!")
# # elif "Logging Details Post-API Call" not in output:
# # raise Exception("Required log message not found!")
# # elif "Logging Details LiteLLM-Failure Call" not in output:
# # raise Exception("Required log message not found!")
# # except Exception as e:
# # print(f"exception type: {type(e).__name__}")
# # pytest.fail(f"Error occurred: {e}")

163
tests/local_testing/test_max_tpm_rpm_limiter.py
View file

@ -1,163 +0,0 @@
### REPLACED BY 'test_parallel_request_limiter.py' ###
# What is this?
## Unit tests for the max tpm / rpm limiter hook for proxy
# import sys, os, asyncio, time, random
# from datetime import datetime
# import traceback
# from dotenv import load_dotenv
# from typing import Optional
# load_dotenv()
# import os
# sys.path.insert(
# 0, os.path.abspath("../..")
# ) # Adds the parent directory to the system path
# import pytest
# import litellm
# from litellm import Router
# from litellm.proxy.utils import ProxyLogging, hash_token
# from litellm.proxy._types import UserAPIKeyAuth
# from litellm.caching.caching import DualCache, RedisCache
# from litellm.proxy.hooks.tpm_rpm_limiter import _PROXY_MaxTPMRPMLimiter
# from datetime import datetime
# @pytest.mark.asyncio
# async def test_pre_call_hook_rpm_limits():
# """
# Test if error raised on hitting rpm limits
# """
# litellm.set_verbose = True
# _api_key = hash_token("sk-12345")
# user_api_key_dict = UserAPIKeyAuth(api_key=_api_key, tpm_limit=9, rpm_limit=1)
# local_cache = DualCache()
# # redis_usage_cache = RedisCache()
# local_cache.set_cache(
# key=_api_key, value={"api_key": _api_key, "tpm_limit": 9, "rpm_limit": 1}
# )
# tpm_rpm_limiter = _PROXY_MaxTPMRPMLimiter(internal_cache=DualCache())
# await tpm_rpm_limiter.async_pre_call_hook(
# user_api_key_dict=user_api_key_dict, cache=local_cache, data={}, call_type=""
# )
# kwargs = {"litellm_params": {"metadata": {"user_api_key": _api_key}}}
# await tpm_rpm_limiter.async_log_success_event(
# kwargs=kwargs,
# response_obj="",
# start_time="",
# end_time="",
# )
# ## Expected cache val: {"current_requests": 0, "current_tpm": 0, "current_rpm": 1}
# try:
# await tpm_rpm_limiter.async_pre_call_hook(
# user_api_key_dict=user_api_key_dict,
# cache=local_cache,
# data={},
# call_type="",
# )
# pytest.fail(f"Expected call to fail")
# except Exception as e:
# assert e.status_code == 429
# @pytest.mark.asyncio
# async def test_pre_call_hook_team_rpm_limits(
# _redis_usage_cache: Optional[RedisCache] = None,
# ):
# """
# Test if error raised on hitting team rpm limits
# """
# litellm.set_verbose = True
# _api_key = "sk-12345"
# _team_id = "unique-team-id"
# _user_api_key_dict = {
# "api_key": _api_key,
# "max_parallel_requests": 1,
# "tpm_limit": 9,
# "rpm_limit": 10,
# "team_rpm_limit": 1,
# "team_id": _team_id,
# }
# user_api_key_dict = UserAPIKeyAuth(**_user_api_key_dict) # type: ignore
# _api_key = hash_token(_api_key)
# local_cache = DualCache()
# local_cache.set_cache(key=_api_key, value=_user_api_key_dict)
# internal_cache = DualCache(redis_cache=_redis_usage_cache)
# tpm_rpm_limiter = _PROXY_MaxTPMRPMLimiter(internal_cache=internal_cache)
# await tpm_rpm_limiter.async_pre_call_hook(
# user_api_key_dict=user_api_key_dict, cache=local_cache, data={}, call_type=""
# )
# kwargs = {
# "litellm_params": {
# "metadata": {"user_api_key": _api_key, "user_api_key_team_id": _team_id}
# }
# }
# await tpm_rpm_limiter.async_log_success_event(
# kwargs=kwargs,
# response_obj="",
# start_time="",
# end_time="",
# )
# print(f"local_cache: {local_cache}")
# ## Expected cache val: {"current_requests": 0, "current_tpm": 0, "current_rpm": 1}
# try:
# await tpm_rpm_limiter.async_pre_call_hook(
# user_api_key_dict=user_api_key_dict,
# cache=local_cache,
# data={},
# call_type="",
# )
# pytest.fail(f"Expected call to fail")
# except Exception as e:
# assert e.status_code == 429 # type: ignore
# @pytest.mark.asyncio
# async def test_namespace():
# """
# - test if default namespace set via `proxyconfig._init_cache`
# - respected for tpm/rpm caching
# """
# from litellm.proxy.proxy_server import ProxyConfig
# redis_usage_cache: Optional[RedisCache] = None
# cache_params = {"type": "redis", "namespace": "litellm_default"}
# ## INIT CACHE ##
# proxy_config = ProxyConfig()
# setattr(litellm.proxy.proxy_server, "proxy_config", proxy_config)
# proxy_config._init_cache(cache_params=cache_params)
# redis_cache: Optional[RedisCache] = getattr(
# litellm.proxy.proxy_server, "redis_usage_cache"
# )
# ## CHECK IF NAMESPACE SET ##
# assert redis_cache.namespace == "litellm_default"
# ## CHECK IF TPM/RPM RATE LIMITING WORKS ##
# await test_pre_call_hook_team_rpm_limits(_redis_usage_cache=redis_cache)
# current_date = datetime.now().strftime("%Y-%m-%d")
# current_hour = datetime.now().strftime("%H")
# current_minute = datetime.now().strftime("%M")
# precise_minute = f"{current_date}-{current_hour}-{current_minute}"
# cache_key = "litellm_default:usage:{}".format(precise_minute)
# value = await redis_cache.async_get_cache(key=cache_key)
# assert value is not None

243
tests/local_testing/test_mem_leak.py
View file

@ -1,243 +0,0 @@
# import io
# import os
# import sys
# sys.path.insert(0, os.path.abspath("../.."))
# import litellm
# from memory_profiler import profile
# from litellm.utils import (
# ModelResponseIterator,
# ModelResponseListIterator,
# CustomStreamWrapper,
# )
# from litellm.types.utils import ModelResponse, Choices, Message
# import time
# import pytest
# # @app.post("/debug")
# # async def debug(body: ExampleRequest) -> str:
# # return await main_logic(body.query)
# def model_response_list_factory():
# chunks = [
# {
# "id": "chatcmpl-9SQxdH5hODqkWyJopWlaVOOUnFwlj",
# "choices": [
# {
# "delta": {"content": "", "role": "assistant"},
# "finish_reason": None,
# "index": 0,
# }
# ],
# "created": 1716563849,
# "model": "gpt-4o-2024-05-13",
# "object": "chat.completion.chunk",
# "system_fingerprint": "fp_5f4bad809a",
# },
# {
# "id": "chatcmpl-9SQxdH5hODqkWyJopWlaVOOUnFwlj",
# "choices": [
# {"delta": {"content": "This"}, "finish_reason": None, "index": 0}
# ],
# "created": 1716563849,
# "model": "gpt-4o-2024-05-13",
# "object": "chat.completion.chunk",
# "system_fingerprint": "fp_5f4bad809a",
# },
# {
# "id": "chatcmpl-9SQxdH5hODqkWyJopWlaVOOUnFwlj",
# "choices": [
# {"delta": {"content": " is"}, "finish_reason": None, "index": 0}
# ],
# "created": 1716563849,
# "model": "gpt-4o-2024-05-13",
# "object": "chat.completion.chunk",
# "system_fingerprint": "fp_5f4bad809a",
# },
# {
# "id": "chatcmpl-9SQxdH5hODqkWyJopWlaVOOUnFwlj",
# "choices": [
# {"delta": {"content": " a"}, "finish_reason": None, "index": 0}
# ],
# "created": 1716563849,
# "model": "gpt-4o-2024-05-13",
# "object": "chat.completion.chunk",
# "system_fingerprint": "fp_5f4bad809a",
# },
# {
# "id": "chatcmpl-9SQxdH5hODqkWyJopWlaVOOUnFwlj",
# "choices": [
# {"delta": {"content": " dummy"}, "finish_reason": None, "index": 0}
# ],
# "created": 1716563849,
# "model": "gpt-4o-2024-05-13",
# "object": "chat.completion.chunk",
# "system_fingerprint": "fp_5f4bad809a",
# },
# {
# "id": "chatcmpl-9SQxdH5hODqkWyJopWlaVOOUnFwlj",
# "choices": [
# {
# "delta": {"content": " response"},
# "finish_reason": None,
# "index": 0,
# }
# ],
# "created": 1716563849,
# "model": "gpt-4o-2024-05-13",
# "object": "chat.completion.chunk",
# "system_fingerprint": "fp_5f4bad809a",
# },
# {
# "id": "",
# "choices": [
# {
# "finish_reason": None,
# "index": 0,
# "content_filter_offsets": {
# "check_offset": 35159,
# "start_offset": 35159,
# "end_offset": 36150,
# },
# "content_filter_results": {
# "hate": {"filtered": False, "severity": "safe"},
# "self_harm": {"filtered": False, "severity": "safe"},
# "sexual": {"filtered": False, "severity": "safe"},
# "violence": {"filtered": False, "severity": "safe"},
# },
# }
# ],
# "created": 0,
# "model": "",
# "object": "",
# },
# {
# "id": "chatcmpl-9SQxdH5hODqkWyJopWlaVOOUnFwlj",
# "choices": [{"delta": {"content": "."}, "finish_reason": None, "index": 0}],
# "created": 1716563849,
# "model": "gpt-4o-2024-05-13",
# "object": "chat.completion.chunk",
# "system_fingerprint": "fp_5f4bad809a",
# },
# {
# "id": "chatcmpl-9SQxdH5hODqkWyJopWlaVOOUnFwlj",
# "choices": [{"delta": {}, "finish_reason": "stop", "index": 0}],
# "created": 1716563849,
# "model": "gpt-4o-2024-05-13",
# "object": "chat.completion.chunk",
# "system_fingerprint": "fp_5f4bad809a",
# },
# {
# "id": "",
# "choices": [
# {
# "finish_reason": None,
# "index": 0,
# "content_filter_offsets": {
# "check_offset": 36150,
# "start_offset": 36060,
# "end_offset": 37029,
# },
# "content_filter_results": {
# "hate": {"filtered": False, "severity": "safe"},
# "self_harm": {"filtered": False, "severity": "safe"},
# "sexual": {"filtered": False, "severity": "safe"},
# "violence": {"filtered": False, "severity": "safe"},
# },
# }
# ],
# "created": 0,
# "model": "",
# "object": "",
# },
# ]
# chunk_list = []
# for chunk in chunks:
# new_chunk = litellm.ModelResponse(stream=True, id=chunk["id"])
# if "choices" in chunk and isinstance(chunk["choices"], list):
# new_choices = []
# for choice in chunk["choices"]:
# if isinstance(choice, litellm.utils.StreamingChoices):
# _new_choice = choice
# elif isinstance(choice, dict):
# _new_choice = litellm.utils.StreamingChoices(**choice)
# new_choices.append(_new_choice)
# new_chunk.choices = new_choices
# chunk_list.append(new_chunk)
# return ModelResponseListIterator(model_responses=chunk_list)
# async def mock_completion(*args, **kwargs):
# completion_stream = model_response_list_factory()
# return litellm.CustomStreamWrapper(
# completion_stream=completion_stream,
# model="gpt-4-0613",
# custom_llm_provider="cached_response",
# logging_obj=litellm.Logging(
# model="gpt-4-0613",
# messages=[{"role": "user", "content": "Hey"}],
# stream=True,
# call_type="completion",
# start_time=time.time(),
# litellm_call_id="12345",
# function_id="1245",
# ),
# )
# @profile
# async def main_logic() -> str:
# stream = await mock_completion()
# result = ""
# async for chunk in stream:
# result += chunk.choices[0].delta.content or ""
# return result
# import asyncio
# for _ in range(100):
# asyncio.run(main_logic())
# # @pytest.mark.asyncio
# # def test_memory_profile(capsys):
# # # Run the async function
# # result = asyncio.run(main_logic())
# # # Verify the result
# # assert result == "This is a dummy response."
# # # Capture the output
# # captured = capsys.readouterr()
# # # Print memory output for debugging
# # print("Memory Profiler Output:")
# # print(f"captured out: {captured.out}")
# # # Basic memory leak checks
# # for idx, line in enumerate(captured.out.split("\n")):
# # if idx % 2 == 0 and "MiB" in line:
# # print(f"line: {line}")
# # # mem_lines = [line for line in captured.out.split("\n") if "MiB" in line]
# # print(mem_lines)
# # # Ensure we have some memory lines
# # assert len(mem_lines) > 0, "No memory profiler output found"
# # # Optional: Add more specific memory leak detection
# # for line in mem_lines:
# # # Extract memory increment
# # parts = line.split()
# # if len(parts) >= 3:
# # try:
# # mem_increment = float(parts[2].replace("MiB", ""))
# # # Assert that memory increment is below a reasonable threshold
# # assert mem_increment < 1.0, f"Potential memory leak detected: {line}"
# # except (ValueError, IndexError):
# # pass # Skip lines that don't match expected format

153
tests/local_testing/test_mem_usage.py
View file

@ -1,153 +0,0 @@
# #### What this tests ####
# from memory_profiler import profile, memory_usage
# import sys, os, time
# import traceback, asyncio
# import pytest
# sys.path.insert(
# 0, os.path.abspath("../..")
# ) # Adds the parent directory to the system path
# import litellm
# from litellm import Router
# from concurrent.futures import ThreadPoolExecutor
# from collections import defaultdict
# from dotenv import load_dotenv
# from litellm._uuid import uuid
# import tracemalloc
# import objgraph
# objgraph.growth(shortnames=True)
# objgraph.show_most_common_types(limit=10)
# from mem_top import mem_top
# load_dotenv()
# model_list = [
# {
# "model_name": "gpt-3.5-turbo", # openai model name
# "litellm_params": { # params for litellm completion/embedding call
# "model": "azure/gpt-4.1-mini",
# "api_key": os.getenv("AZURE_API_KEY"),
# "api_version": os.getenv("AZURE_API_VERSION"),
# "api_base": os.getenv("AZURE_API_BASE"),
# },
# "tpm": 240000,
# "rpm": 1800,
# },
# {
# "model_name": "bad-model", # openai model name
# "litellm_params": { # params for litellm completion/embedding call
# "model": "azure/gpt-4.1-mini",
# "api_key": "bad-key",
# "api_version": os.getenv("AZURE_API_VERSION"),
# "api_base": os.getenv("AZURE_API_BASE"),
# },
# "tpm": 240000,
# "rpm": 1800,
# },
# {
# "model_name": "text-embedding-ada-002",
# "litellm_params": {
# "model": "azure/text-embedding-ada-002",
# "api_key": os.environ["AZURE_API_KEY"],
# "api_base": os.environ["AZURE_API_BASE"],
# },
# "tpm": 100000,
# "rpm": 10000,
# },
# ]
# litellm.set_verbose = True
# litellm.cache = litellm.Cache(
# type="s3", s3_bucket_name="litellm-my-test-bucket-2", s3_region_name="us-east-1"
# )
# router = Router(
# model_list=model_list,
# fallbacks=[
# {"bad-model": ["gpt-3.5-turbo"]},
# ],
# ) # type: ignore
# async def router_acompletion():
# # embedding call
# question = f"This is a test: {uuid.uuid4()}" * 1
# response = await router.acompletion(
# model="bad-model", messages=[{"role": "user", "content": question}]
# )
# print("completion-resp", response)
# return response
# async def main():
# for i in range(1):
# start = time.time()
# n = 15 # Number of concurrent tasks
# tasks = [router_acompletion() for _ in range(n)]
# chat_completions = await asyncio.gather(*tasks)
# successful_completions = [c for c in chat_completions if c is not None]
# # Write errors to error_log.txt
# with open("error_log.txt", "a") as error_log:
# for completion in chat_completions:
# if isinstance(completion, str):
# error_log.write(completion + "\n")
# print(n, time.time() - start, len(successful_completions))
# print()
# print(vars(router))
# prev_models = router.previous_models
# print("vars in prev_models")
# print(prev_models[0].keys())
# if __name__ == "__main__":
# # Blank out contents of error_log.txt
# open("error_log.txt", "w").close()
# import tracemalloc
# tracemalloc.start(25)
# # ... run your application ...
# asyncio.run(main())
# print(mem_top())
# snapshot = tracemalloc.take_snapshot()
# # top_stats = snapshot.statistics('lineno')
# # print("[ Top 10 ]")
# # for stat in top_stats[:50]:
# # print(stat)
# top_stats = snapshot.statistics("traceback")
# # pick the biggest memory block
# stat = top_stats[0]
# print("%s memory blocks: %.1f KiB" % (stat.count, stat.size / 1024))
# for line in stat.traceback.format():
# print(line)
# print()
# stat = top_stats[1]
# print("%s memory blocks: %.1f KiB" % (stat.count, stat.size / 1024))
# for line in stat.traceback.format():
# print(line)
# print()
# stat = top_stats[2]
# print("%s memory blocks: %.1f KiB" % (stat.count, stat.size / 1024))
# for line in stat.traceback.format():
# print(line)
# print()
# stat = top_stats[3]
# print("%s memory blocks: %.1f KiB" % (stat.count, stat.size / 1024))
# for line in stat.traceback.format():
# print(line)

336
tests/local_testing/test_ollama_local.py
View file

@ -1,336 +0,0 @@
# ##### THESE TESTS CAN ONLY RUN LOCALLY WITH THE OLLAMA SERVER RUNNING ######
# # https://ollama.ai/
# import sys, os
# import traceback
# from dotenv import load_dotenv
# load_dotenv()
# import os
# sys.path.insert(0, os.path.abspath('../..')) # Adds the parent directory to the system path
# import pytest
# import litellm
# from litellm import embedding, completion
# import asyncio
# user_message = "respond in 20 words. who are you?"
# messages = [{ "content": user_message,"role": "user"}]
# async def test_ollama_aembeddings():
# litellm.set_verbose = True
# input = "The food was delicious and the waiter..."
# response = await litellm.aembedding(model="ollama/mistral", input=input)
# print(response)
# asyncio.run(test_ollama_aembeddings())
# def test_ollama_embeddings():
# litellm.set_verbose = True
# input = "The food was delicious and the waiter..."
# response = litellm.embedding(model="ollama/mistral", input=input)
# print(response)
# test_ollama_embeddings()
# def test_ollama_streaming():
# try:
# litellm.set_verbose = False
# messages = [
# {"role": "user", "content": "What is the weather like in Boston?"}
# ]
# functions = [
# {
# "name": "get_current_weather",
# "description": "Get the current weather in a given location",
# "parameters": {
# "type": "object",
# "properties": {
# "location": {
# "type": "string",
# "description": "The city and state, e.g. San Francisco, CA"
# },
# "unit": {
# "type": "string",
# "enum": ["celsius", "fahrenheit"]
# }
# },
# "required": ["location"]
# }
# }
# ]
# response = litellm.completion(model="ollama/mistral",
# messages=messages,
# functions=functions,
# stream=True)
# for chunk in response:
# print(f"CHUNK: {chunk}")
# except Exception as e:
# print(e)
# # test_ollama_streaming()
# async def test_async_ollama_streaming():
# try:
# litellm.set_verbose = False
# response = await litellm.acompletion(model="ollama/mistral-openorca",
# messages=[{"role": "user", "content": "Hey, how's it going?"}],
# stream=True)
# async for chunk in response:
# print(f"CHUNK: {chunk}")
# except Exception as e:
# print(e)
# # asyncio.run(test_async_ollama_streaming())
# def test_completion_ollama():
# try:
# litellm.set_verbose = True
# response = completion(
# model="ollama/mistral",
# messages=[{"role": "user", "content": "Hey, how's it going?"}],
# max_tokens=200,
# request_timeout = 10,
# stream=True
# )
# for chunk in response:
# print(chunk)
# print(response)
# except Exception as e:
# pytest.fail(f"Error occurred: {e}")
# # test_completion_ollama()
# def test_completion_ollama_function_calling():
# try:
# litellm.set_verbose = True
# messages = [
# {"role": "user", "content": "What is the weather like in Boston?"}
# ]
# functions = [
# {
# "name": "get_current_weather",
# "description": "Get the current weather in a given location",
# "parameters": {
# "type": "object",
# "properties": {
# "location": {
# "type": "string",
# "description": "The city and state, e.g. San Francisco, CA"
# },
# "unit": {
# "type": "string",
# "enum": ["celsius", "fahrenheit"]
# }
# },
# "required": ["location"]
# }
# }
# ]
# response = completion(
# model="ollama/mistral",
# messages=messages,
# functions=functions,
# max_tokens=200,
# request_timeout = 10,
# )
# for chunk in response:
# print(chunk)
# print(response)
# except Exception as e:
# pytest.fail(f"Error occurred: {e}")
# # test_completion_ollama_function_calling()
# async def async_test_completion_ollama_function_calling():
# try:
# litellm.set_verbose = True
# messages = [
# {"role": "user", "content": "What is the weather like in Boston?"}
# ]
# functions = [
# {
# "name": "get_current_weather",
# "description": "Get the current weather in a given location",
# "parameters": {
# "type": "object",
# "properties": {
# "location": {
# "type": "string",
# "description": "The city and state, e.g. San Francisco, CA"
# },
# "unit": {
# "type": "string",
# "enum": ["celsius", "fahrenheit"]
# }
# },
# "required": ["location"]
# }
# }
# ]
# response = await litellm.acompletion(
# model="ollama/mistral",
# messages=messages,
# functions=functions,
# max_tokens=200,
# request_timeout = 10,
# )
# print(response)
# except Exception as e:
# pytest.fail(f"Error occurred: {e}")
# # asyncio.run(async_test_completion_ollama_function_calling())
# def test_completion_ollama_with_api_base():
# try:
# response = completion(
# model="ollama/llama2",
# messages=messages,
# api_base="http://localhost:11434"
# )
# print(response)
# except Exception as e:
# pytest.fail(f"Error occurred: {e}")
# # test_completion_ollama_with_api_base()
# def test_completion_ollama_custom_prompt_template():
# user_message = "what is litellm?"
# litellm.register_prompt_template(
# model="ollama/llama2",
# roles={
# "system": {"pre_message": "System: "},
# "user": {"pre_message": "User: "},
# "assistant": {"pre_message": "Assistant: "}
# }
# )
# messages = [{ "content": user_message,"role": "user"}]
# litellm.set_verbose = True
# try:
# response = completion(
# model="ollama/llama2",
# messages=messages,
# stream=True
# )
# print(response)
# for chunk in response:
# print(chunk)
# # print(chunk['choices'][0]['delta'])
# except Exception as e:
# traceback.print_exc()
# pytest.fail(f"Error occurred: {e}")
# # test_completion_ollama_custom_prompt_template()
# async def test_completion_ollama_async_stream():
# user_message = "what is the weather"
# messages = [{ "content": user_message,"role": "user"}]
# try:
# response = await litellm.acompletion(
# model="ollama/llama2",
# messages=messages,
# api_base="http://localhost:11434",
# stream=True
# )
# async for chunk in response:
# print(chunk['choices'][0]['delta'])
# print("TEST ASYNC NON Stream")
# response = await litellm.acompletion(
# model="ollama/llama2",
# messages=messages,
# api_base="http://localhost:11434",
# )
# print(response)
# except Exception as e:
# pytest.fail(f"Error occurred: {e}")
# # import asyncio
# # asyncio.run(test_completion_ollama_async_stream())
# def prepare_messages_for_chat(text: str) -> list:
# messages = [
# {"role": "user", "content": text},
# ]
# return messages
# async def ask_question():
# params = {
# "messages": prepare_messages_for_chat("What is litellm? tell me 10 things about it who is sihaan.write an essay"),
# "api_base": "http://localhost:11434",
# "model": "ollama/llama2",
# "stream": True,
# }
# response = await litellm.acompletion(**params)
# return response
# async def main():
# response = await ask_question()
# async for chunk in response:
# print(chunk)
# print("test async completion without streaming")
# response = await litellm.acompletion(
# model="ollama/llama2",
# messages=prepare_messages_for_chat("What is litellm? respond in 2 words"),
# )
# print("response", response)
# def test_completion_expect_error():
# # this tests if we can exception map correctly for ollama
# print("making ollama request")
# # litellm.set_verbose=True
# user_message = "what is litellm?"
# messages = [{ "content": user_message,"role": "user"}]
# try:
# response = completion(
# model="ollama/invalid",
# messages=messages,
# stream=True
# )
# print(response)
# for chunk in response:
# print(chunk)
# # print(chunk['choices'][0]['delta'])
# except Exception as e:
# pass
# pytest.fail(f"Error occurred: {e}")
# # test_completion_expect_error()
# def test_ollama_llava():
# litellm.set_verbose=True
# # same params as gpt-4 vision
# response = completion(
# model = "ollama/llava",
# messages=[
# {
# "role": "user",
# "content": [
# {
# "type": "text",
# "text": "What is in this picture"
# },
# {
# "type": "image_url",
# "image_url": {
# "url": "iVBORw0KGgoAAAANSUhEUgAAAG0AAABmCAYAAADBPx+VAAAACXBIWXMAAAsTAAALEwEAmpwYAAAAAXNSR0IArs4c6QAAAARnQU1BAACxjwv8YQUAAA3VSURBVHgB7Z27r0zdG8fX743i1bi1ikMoFMQloXRpKFFIqI7LH4BEQ+NWIkjQuSWCRIEoULk0gsK1kCBI0IhrQVT7tz/7zZo888yz1r7MnDl7z5xvsjkzs2fP3uu71nNfa7lkAsm7d++Sffv2JbNmzUqcc8m0adOSzZs3Z+/XES4ZckAWJEGWPiCxjsQNLWmQsWjRIpMseaxcuTKpG/7HP27I8P79e7dq1ars/yL4/v27S0ejqwv+cUOGEGGpKHR37tzJCEpHV9tnT58+dXXCJDdECBE2Ojrqjh071hpNECjx4cMHVycM1Uhbv359B2F79+51586daxN/+pyRkRFXKyRDAqxEp4yMlDDzXG1NPnnyJKkThoK0VFd1ELZu3TrzXKxKfW7dMBQ6bcuWLW2v0VlHjx41z717927ba22U9APcw7Nnz1oGEPeL3m3p2mTAYYnFmMOMXybPPXv2bNIPpFZr1NHn4HMw0KRBjg9NuRw95s8PEcz/6DZELQd/09C9QGq5RsmSRybqkwHGjh07OsJSsYYm3ijPpyHzoiacg35MLdDSIS/O1yM778jOTwYUkKNHWUzUWaOsylE00MyI0fcnOwIdjvtNdW/HZwNLGg+sR1kMepSNJXmIwxBZiG8tDTpEZzKg0GItNsosY8USkxDhD0Rinuiko2gfL/RbiD2LZAjU9zKQJj8RDR0vJBR1/Phx9+PHj9Z7REF4nTZkxzX4LCXHrV271qXkBAPGfP/atWvu/PnzHe4C97F48eIsRLZ9+3a3f/9+87dwP1JxaF7/3r17ba+5l4EcaVo0lj3SBq5kGTJSQmLWMjgYNei2GPT1MuMqGTDEFHzeQSP2wi/jGnkmPJ/nhccs44jvDAxpVcxnq0F6eT8h4ni/iIWpR5lPyA6ETkNXoSukvpJAD3AsXLiwpZs49+fPn5ke4j10TqYvegSfn0OnafC+Tv9ooA/JPkgQysqQNBzagXY55nO/oa1F7qvIPWkRL12WRpMWUvpVDYmxAPehxWSe8ZEXL20sadYIozfmNch4QJPAfeJgW3rNsnzphBKNJM2KKODo1rVOMRYik5ETy3ix4qWNI81qAAirizgMIc+yhTytx0JWZuNI03qsrgWlGtwjoS9XwgUhWGyhUaRZZQNNIEwCiXD16tXcAHUs79co0vSD8rrJCIW98pzvxpAWyyo3HYwqS0+H0BjStClcZJT5coMm6D2LOF8TolGJtK9fvyZpyiC5ePFi9nc/oJU4eiEP0jVoAnHa9wyJycITMP78+eMeP37sXrx44d6+fdt6f82aNdkx1pg9e3Zb5W+RSRE+n+VjksQWifvVaTKFhn5O8my63K8Qabdv33b379/PiAP//vuvW7BggZszZ072/+TJk91YgkafPn166zXB1rQHFvouAWHq9z3SEevSUerqCn2/dDCeta2jxYbr69evk4MHDyY7d+7MjhMnTiTPnz9Pfv/+nfQT2ggpO2dMF8cghuoM7Ygj5iWCqRlGFml0QC/ftGmTmzt3rmsaKDsgBSPh0/8yPeLLBihLkOKJc0jp8H8vUzcxIA1k6QJ/c78tWEyj5P3o4u9+jywNPdJi5rAH9x0KHcl4Hg570eQp3+vHXGyrmEeigzQsQsjavXt38ujRo44LQuDDhw+TW7duRS1HGgMxhNXHgflaNTOsHyKvHK5Ijo2jbFjJBQK9YwFd6RVMzfgRBmEfP37suBBm/p49e1qjEP2mwTViNRo0VJWH1deMXcNK08uUjVUu7s/zRaL+oLNxz1bpANco4npUgX4G2eFbpDFyQoQxojBCpEGSytmOH8qrH5Q9vuzD6ofQylkCUmh8DBAr+q8JCyVNtWQIidKQE9wNtLSQnS4jDSsxNHogzFuQBw4cyM61UKVsjfr3ooBkPSqqQHesUPWVtzi9/vQi1T+rJj7WiTz4Pt/l3LxUkr5P2VYZaZ4URpsE+st/dujQoaBBYokbrz/8TJNQYLSonrPS9kUaSkPeZyj1AWSj+d+VBoy1pIWVNed8P0Ll/ee5HdGRhrHhR5GGN0r4LGZBaj8oFDJitBTJzIZgFcmU0Y8ytWMZMzJOaXUSrUs5RxKnrxmbb5YXO9VGUhtpXldhEUogFr3IzIsvlpmdosVcGVGXFWp2oU9kLFL3dEkSz6NHEY1sjSRdIuDFWEhd8KxFqsRi1uM/nz9/zpxnwlESONdg6dKlbsaMGS4EHFHtjFIDHwKOo46l4TxSuxgDzi+rE2jg+BaFruOX4HXa0Nnf1lwAPufZeF8/r6zD97WK2qFnGjBxTw5qNGPxT+5T/r7/7RawFC3j4vTp09koCxkeHjqbHJqArmH5UrFKKksnxrK7FuRIs8STfBZv+luugXZ2pR/pP9Ois4z+TiMzUUkUjD0iEi1fzX8GmXyuxUBRcaUfykV0YZnlJGKQpOiGB76x5GeWkWWJc3mOrK6S7xdND+W5N6XyaRgtWJFe13GkaZnKOsYqGdOVVVbGupsyA/l7emTLHi7vwTdirNEt0qxnzAvBFcnQF16xh/TMpUuXHDowhlA9vQVraQhkudRdzOnK+04ZSP3DUhVSP61YsaLtd/ks7ZgtPcXqPqEafHkdqa84X6aCeL7YWlv6edGFHb+ZFICPlljHhg0bKuk0CSvVznWsotRu433alNdFrqG45ejoaPCaUkWERpLXjzFL2Rpllp7PJU2a/v7Ab8N05/9t27Z16KUqoFGsxnI9EosS2niSYg9SpU6B4JgTrvVW1flt1sT+0ADIJU2maXzcUTraGCRaL1Wp9rUMk16PMom8QhruxzvZIegJjFU7LLCePfS8uaQdPny4jTTL0dbee5mYokQsXTIWNY46kuMbnt8Kmec+LGWtOVIl9cT1rCB0V8WqkjAsRwta93TbwNYoGKsUSChN44lgBNCoHLHzquYKrU6qZ8lolCIN0Rh6cP0Q3U6I6IXILYOQI513hJaSKAorFpuHXJNfVlpRtmYBk1Su1obZr5dnKAO+L10Hrj3WZW+E3qh6IszE37F6EB+68mGpvKm4eb9bFrlzrok7fvr0Kfv727dvWRmdVTJHw0qiiCUSZ6wCK+7XL/AcsgNyL74DQQ730sv78Su7+t/A36MdY0sW5o40ahslXr58aZ5HtZB8GH64m9EmMZ7FpYw4T6QnrZfgenrhFxaSiSGXtPnz57e9TkNZLvTjeqhr734CNtrK41L40sUQckmj1lGKQ0rC37x544r8eNXRpnVE3ZZY7zXo8NomiO0ZUCj2uHz58rbXoZ6gc0uA+F6ZeKS/jhRDUq8MKrTho9fEkihMmhxtBI1DxKFY9XLpVcSkfoi8JGnToZO5sU5aiDQIW716ddt7ZLYtMQlhECdBGXZZMWldY5BHm5xgAroWj4C0hbYkSc/jBmggIrXJWlZM6pSETsEPGqZOndr2uuuR5rF169a2HoHPdurUKZM4CO1WTPqaDaAd+GFGKdIQkxAn9RuEWcTRyN2KSUgiSgF5aWzPTeA/lN5rZubMmR2bE4SIC4nJoltgAV/dVefZm72AtctUCJU2CMJ327hxY9t7EHbkyJFseq+EJSY16RPo3Dkq1kkr7+q0bNmyDuLQcZBEPYmHVdOBiJyIlrRDq41YPWfXOxUysi5fvtyaj+2BpcnsUV/oSoEMOk2CQGlr4ckhBwaetBhjCwH0ZHtJROPJkyc7UjcYLDjmrH7ADTEBXFfOYmB0k9oYBOjJ8b4aOYSe7QkKcYhFlq3QYLQhSidNmtS2RATwy8YOM3EQJsUjKiaWZ+vZToUQgzhkHXudb/PW5YMHD9yZM2faPsMwoc7RciYJXbGuBqJ1UIGKKLv915jsvgtJxCZDubdXr165mzdvtr1Hz5LONA8jrUwKPqsmVesKa49S3Q4WxmRPUEYdTjgiUcfUwLx589ySJUva3oMkP6IYddq6HMS4o55xBJBUeRjzfa4Zdeg56QZ43LhxoyPo7Lf1kNt7oO8wWAbNwaYjIv5lhyS7kRf96dvm5Jah8vfvX3flyhX35cuX6HfzFHOToS1H4BenCaHvO8pr8iDuwoUL7tevX+b5ZdbBair0xkFIlFDlW4ZknEClsp/TzXyAKVOmmHWFVSbDNw1l1+4f90U6IY/q4V27dpnE9bJ+v87QEydjqx/UamVVPRG+mwkNTYN+9tjkwzEx+atCm/X9WvWtDtAb68Wy9LXa1UmvCDDIpPkyOQ5ZwSzJ4jMrvFcr0rSjOUh+GcT4LSg5ugkW1Io0/SCDQBojh0hPlaJdah+tkVYrnTZowP8iq1F1TgMBBauufyB33x1v+NWFYmT5KmppgHC+NkAgbmRkpD3yn9QIseXymoTQFGQmIOKTxiZIWpvAatenVqRVXf2nTrAWMsPnKrMZHz6bJq5jvce6QK8J1cQNgKxlJapMPdZSR64/UivS9NztpkVEdKcrs5alhhWP9NeqlfWopzhZScI6QxseegZRGeg5a8C3Re1Mfl1ScP36ddcUaMuv24iOJtz7sbUjTS4qBvKmstYJoUauiuD3k5qhyr7QdUHMeCgLa1Ear9NquemdXgmum4fvJ6w1lqsuDhNrg1qSpleJK7K3TF0Q2jSd94uSZ60kK1e3qyVpQK6PVWXp2/FC3mp6jBhKKOiY2h3gtUV64TWM6wDETRPLDfSakXmH3w8g9Jlug8ZtTt4kVF0kLUYYmCCtD/DrQ5YhMGbA9L3ucdjh0y8kOHW5gU/VEEmJTcL4Pz/f7mgoAbYkAAAAAElFTkSuQmCC"
# }
# }
# ]
# }
# ],
# )
# print("Response from ollama/llava")
# print(response)
# # test_ollama_llava()
# # PROCESSED CHUNK PRE CHUNK CREATOR

334
tests/local_testing/test_ollama_local_chat.py
View file

@ -1,334 +0,0 @@
# ##### THESE TESTS CAN ONLY RUN LOCALLY WITH THE OLLAMA SERVER RUNNING ######
# # https://ollama.ai/
# import sys, os
# import traceback
# from dotenv import load_dotenv
# load_dotenv()
# import os
# sys.path.insert(
# 0, os.path.abspath("../..")
# ) # Adds the parent directory to the system path
# import pytest
# import litellm
# from litellm import embedding, completion
# import asyncio
# user_message = "respond in 20 words. who are you?"
# messages = [{"content": user_message, "role": "user"}]
# def test_ollama_streaming():
# try:
# litellm.set_verbose = False
# messages = [{"role": "user", "content": "What is the weather like in Boston?"}]
# functions = [
# {
# "name": "get_current_weather",
# "description": "Get the current weather in a given location",
# "parameters": {
# "type": "object",
# "properties": {
# "location": {
# "type": "string",
# "description": "The city and state, e.g. San Francisco, CA",
# },
# "unit": {"type": "string", "enum": ["celsius", "fahrenheit"]},
# },
# "required": ["location"],
# },
# }
# ]
# response = litellm.completion(
# model="ollama_chat/mistral",
# messages=messages,
# functions=functions,
# stream=True,
# )
# for chunk in response:
# print(f"CHUNK: {chunk}")
# except Exception as e:
# print(e)
# # test_ollama_streaming()
# async def test_async_ollama_streaming():
# try:
# litellm.set_verbose = True
# response = await litellm.acompletion(
# model="ollama_chat/llama2",
# messages=[{"role": "user", "content": "Hey, how's it going?"}],
# stream=True,
# )
# async for chunk in response:
# print(f"CHUNK: {chunk}")
# except Exception as e:
# print(e)
# # asyncio.run(test_async_ollama_streaming())
# async def test_async_ollama():
# try:
# litellm.set_verbose = True
# response = await litellm.acompletion(
# model="ollama_chat/llama2",
# messages=[{"role": "user", "content": "Hey, how's it going?"}],
# )
# print("\n response", response)
# except Exception as e:
# print(e)
# # asyncio.run(test_async_ollama())
# def test_completion_ollama():
# try:
# litellm.set_verbose = True
# response = completion(
# model="ollama_chat/mistral",
# messages=[{"role": "user", "content": "Hey, how's it going?"}],
# max_tokens=200,
# request_timeout=10,
# stream=True,
# )
# for chunk in response:
# print(chunk)
# print(response)
# except Exception as e:
# pytest.fail(f"Error occurred: {e}")
# # test_completion_ollama()
# def test_completion_ollama_function_calling():
# try:
# litellm.set_verbose = True
# messages = [{"role": "user", "content": "What is the weather like in Boston?"}]
# functions = [
# {
# "name": "get_current_weather",
# "description": "Get the current weather in a given location",
# "parameters": {
# "type": "object",
# "properties": {
# "location": {
# "type": "string",
# "description": "The city and state, e.g. San Francisco, CA",
# },
# "unit": {"type": "string", "enum": ["celsius", "fahrenheit"]},
# },
# "required": ["location"],
# },
# }
# ]
# response = completion(
# model="ollama_chat/mistral",
# messages=messages,
# functions=functions,
# max_tokens=200,
# request_timeout=10,
# )
# for chunk in response:
# print(chunk)
# print(response)
# except Exception as e:
# pytest.fail(f"Error occurred: {e}")
# test_completion_ollama_function_calling()
# async def async_test_completion_ollama_function_calling():
# try:
# litellm.set_verbose = True
# messages = [{"role": "user", "content": "What is the weather like in Boston?"}]
# functions = [
# {
# "name": "get_current_weather",
# "description": "Get the current weather in a given location",
# "parameters": {
# "type": "object",
# "properties": {
# "location": {
# "type": "string",
# "description": "The city and state, e.g. San Francisco, CA",
# },
# "unit": {"type": "string", "enum": ["celsius", "fahrenheit"]},
# },
# "required": ["location"],
# },
# }
# ]
# response = await litellm.acompletion(
# model="ollama/mistral",
# messages=messages,
# functions=functions,
# max_tokens=200,
# request_timeout=10,
# )
# print(response)
# except Exception as e:
# pytest.fail(f"Error occurred: {e}")
# # asyncio.run(async_test_completion_ollama_function_calling())
# def test_completion_ollama_with_api_base():
# try:
# response = completion(
# model="ollama/llama2", messages=messages, api_base="http://localhost:11434"
# )
# print(response)
# except Exception as e:
# pytest.fail(f"Error occurred: {e}")
# # test_completion_ollama_with_api_base()
# def test_completion_ollama_custom_prompt_template():
# user_message = "what is litellm?"
# litellm.register_prompt_template(
# model="ollama/llama2",
# roles={
# "system": {"pre_message": "System: "},
# "user": {"pre_message": "User: "},
# "assistant": {"pre_message": "Assistant: "},
# },
# )
# messages = [{"content": user_message, "role": "user"}]
# litellm.set_verbose = True
# try:
# response = completion(model="ollama/llama2", messages=messages, stream=True)
# print(response)
# for chunk in response:
# print(chunk)
# # print(chunk['choices'][0]['delta'])
# except Exception as e:
# traceback.print_exc()
# pytest.fail(f"Error occurred: {e}")
# # test_completion_ollama_custom_prompt_template()
# async def test_completion_ollama_async_stream():
# user_message = "what is the weather"
# messages = [{"content": user_message, "role": "user"}]
# try:
# response = await litellm.acompletion(
# model="ollama/llama2",
# messages=messages,
# api_base="http://localhost:11434",
# stream=True,
# )
# async for chunk in response:
# print(chunk["choices"][0]["delta"])
# print("TEST ASYNC NON Stream")
# response = await litellm.acompletion(
# model="ollama/llama2",
# messages=messages,
# api_base="http://localhost:11434",
# )
# print(response)
# except Exception as e:
# pytest.fail(f"Error occurred: {e}")
# # import asyncio
# # asyncio.run(test_completion_ollama_async_stream())
# def prepare_messages_for_chat(text: str) -> list:
# messages = [
# {"role": "user", "content": text},
# ]
# return messages
# async def ask_question():
# params = {
# "messages": prepare_messages_for_chat(
# "What is litellm? tell me 10 things about it who is sihaan.write an essay"
# ),
# "api_base": "http://localhost:11434",
# "model": "ollama/llama2",
# "stream": True,
# }
# response = await litellm.acompletion(**params)
# return response
# async def main():
# response = await ask_question()
# async for chunk in response:
# print(chunk)
# print("test async completion without streaming")
# response = await litellm.acompletion(
# model="ollama/llama2",
# messages=prepare_messages_for_chat("What is litellm? respond in 2 words"),
# )
# print("response", response)
# def test_completion_expect_error():
# # this tests if we can exception map correctly for ollama
# print("making ollama request")
# # litellm.set_verbose=True
# user_message = "what is litellm?"
# messages = [{"content": user_message, "role": "user"}]
# try:
# response = completion(model="ollama/invalid", messages=messages, stream=True)
# print(response)
# for chunk in response:
# print(chunk)
# # print(chunk['choices'][0]['delta'])
# except Exception as e:
# pass
# pytest.fail(f"Error occurred: {e}")
# # test_completion_expect_error()
# def test_ollama_llava():
# litellm.set_verbose = True
# # same params as gpt-4 vision
# response = completion(
# model="ollama/llava",
# messages=[
# {
# "role": "user",
# "content": [
# {"type": "text", "text": "What is in this picture"},
# {
# "type": "image_url",
# "image_url": {
# "url": "iVBORw0KGgoAAAANSUhEUgAAAG0AAABmCAYAAADBPx+VAAAACXBIWXMAAAsTAAALEwEAmpwYAAAAAXNSR0IArs4c6QAAAARnQU1BAACxjwv8YQUAAA3VSURBVHgB7Z27r0zdG8fX743i1bi1ikMoFMQloXRpKFFIqI7LH4BEQ+NWIkjQuSWCRIEoULk0gsK1kCBI0IhrQVT7tz/7zZo888yz1r7MnDl7z5xvsjkzs2fP3uu71nNfa7lkAsm7d++Sffv2JbNmzUqcc8m0adOSzZs3Z+/XES4ZckAWJEGWPiCxjsQNLWmQsWjRIpMseaxcuTKpG/7HP27I8P79e7dq1ars/yL4/v27S0ejqwv+cUOGEGGpKHR37tzJCEpHV9tnT58+dXXCJDdECBE2Ojrqjh071hpNECjx4cMHVycM1Uhbv359B2F79+51586daxN/+pyRkRFXKyRDAqxEp4yMlDDzXG1NPnnyJKkThoK0VFd1ELZu3TrzXKxKfW7dMBQ6bcuWLW2v0VlHjx41z717927ba22U9APcw7Nnz1oGEPeL3m3p2mTAYYnFmMOMXybPPXv2bNIPpFZr1NHn4HMw0KRBjg9NuRw95s8PEcz/6DZELQd/09C9QGq5RsmSRybqkwHGjh07OsJSsYYm3ijPpyHzoiacg35MLdDSIS/O1yM778jOTwYUkKNHWUzUWaOsylE00MyI0fcnOwIdjvtNdW/HZwNLGg+sR1kMepSNJXmIwxBZiG8tDTpEZzKg0GItNsosY8USkxDhD0Rinuiko2gfL/RbiD2LZAjU9zKQJj8RDR0vJBR1/Phx9+PHj9Z7REF4nTZkxzX4LCXHrV271qXkBAPGfP/atWvu/PnzHe4C97F48eIsRLZ9+3a3f/9+87dwP1JxaF7/3r17ba+5l4EcaVo0lj3SBq5kGTJSQmLWMjgYNei2GPT1MuMqGTDEFHzeQSP2wi/jGnkmPJ/nhccs44jvDAxpVcxnq0F6eT8h4ni/iIWpR5lPyA6ETkNXoSukvpJAD3AsXLiwpZs49+fPn5ke4j10TqYvegSfn0OnafC+Tv9ooA/JPkgQysqQNBzagXY55nO/oa1F7qvIPWkRL12WRpMWUvpVDYmxAPehxWSe8ZEXL20sadYIozfmNch4QJPAfeJgW3rNsnzphBKNJM2KKODo1rVOMRYik5ETy3ix4qWNI81qAAirizgMIc+yhTytx0JWZuNI03qsrgWlGtwjoS9XwgUhWGyhUaRZZQNNIEwCiXD16tXcAHUs79co0vSD8rrJCIW98pzvxpAWyyo3HYwqS0+H0BjStClcZJT5coMm6D2LOF8TolGJtK9fvyZpyiC5ePFi9nc/oJU4eiEP0jVoAnHa9wyJycITMP78+eMeP37sXrx44d6+fdt6f82aNdkx1pg9e3Zb5W+RSRE+n+VjksQWifvVaTKFhn5O8my63K8Qabdv33b379/PiAP//vuvW7BggZszZ072/+TJk91YgkafPn166zXB1rQHFvouAWHq9z3SEevSUerqCn2/dDCeta2jxYbr69evk4MHDyY7d+7MjhMnTiTPnz9Pfv/+nfQT2ggpO2dMF8cghuoM7Ygj5iWCqRlGFml0QC/ftGmTmzt3rmsaKDsgBSPh0/8yPeLLBihLkOKJc0jp8H8vUzcxIA1k6QJ/c78tWEyj5P3o4u9+jywNPdJi5rAH9x0KHcl4Hg570eQp3+vHXGyrmEeigzQsQsjavXt38ujRo44LQuDDhw+TW7duRS1HGgMxhNXHgflaNTOsHyKvHK5Ijo2jbFjJBQK9YwFd6RVMzfgRBmEfP37suBBm/p49e1qjEP2mwTViNRo0VJWH1deMXcNK08uUjVUu7s/zRaL+oLNxz1bpANco4npUgX4G2eFbpDFyQoQxojBCpEGSytmOH8qrH5Q9vuzD6ofQylkCUmh8DBAr+q8JCyVNtWQIidKQE9wNtLSQnS4jDSsxNHogzFuQBw4cyM61UKVsjfr3ooBkPSqqQHesUPWVtzi9/vQi1T+rJj7WiTz4Pt/l3LxUkr5P2VYZaZ4URpsE+st/dujQoaBBYokbrz/8TJNQYLSonrPS9kUaSkPeZyj1AWSj+d+VBoy1pIWVNed8P0Ll/ee5HdGRhrHhR5GGN0r4LGZBaj8oFDJitBTJzIZgFcmU0Y8ytWMZMzJOaXUSrUs5RxKnrxmbb5YXO9VGUhtpXldhEUogFr3IzIsvlpmdosVcGVGXFWp2oU9kLFL3dEkSz6NHEY1sjSRdIuDFWEhd8KxFqsRi1uM/nz9/zpxnwlESONdg6dKlbsaMGS4EHFHtjFIDHwKOo46l4TxSuxgDzi+rE2jg+BaFruOX4HXa0Nnf1lwAPufZeF8/r6zD97WK2qFnGjBxTw5qNGPxT+5T/r7/7RawFC3j4vTp09koCxkeHjqbHJqArmH5UrFKKksnxrK7FuRIs8STfBZv+luugXZ2pR/pP9Ois4z+TiMzUUkUjD0iEi1fzX8GmXyuxUBRcaUfykV0YZnlJGKQpOiGB76x5GeWkWWJc3mOrK6S7xdND+W5N6XyaRgtWJFe13GkaZnKOsYqGdOVVVbGupsyA/l7emTLHi7vwTdirNEt0qxnzAvBFcnQF16xh/TMpUuXHDowhlA9vQVraQhkudRdzOnK+04ZSP3DUhVSP61YsaLtd/ks7ZgtPcXqPqEafHkdqa84X6aCeL7YWlv6edGFHb+ZFICPlljHhg0bKuk0CSvVznWsotRu433alNdFrqG45ejoaPCaUkWERpLXjzFL2Rpllp7PJU2a/v7Ab8N05/9t27Z16KUqoFGsxnI9EosS2niSYg9SpU6B4JgTrvVW1flt1sT+0ADIJU2maXzcUTraGCRaL1Wp9rUMk16PMom8QhruxzvZIegJjFU7LLCePfS8uaQdPny4jTTL0dbee5mYokQsXTIWNY46kuMbnt8Kmec+LGWtOVIl9cT1rCB0V8WqkjAsRwta93TbwNYoGKsUSChN44lgBNCoHLHzquYKrU6qZ8lolCIN0Rh6cP0Q3U6I6IXILYOQI513hJaSKAorFpuHXJNfVlpRtmYBk1Su1obZr5dnKAO+L10Hrj3WZW+E3qh6IszE37F6EB+68mGpvKm4eb9bFrlzrok7fvr0Kfv727dvWRmdVTJHw0qiiCUSZ6wCK+7XL/AcsgNyL74DQQ730sv78Su7+t/A36MdY0sW5o40ahslXr58aZ5HtZB8GH64m9EmMZ7FpYw4T6QnrZfgenrhFxaSiSGXtPnz57e9TkNZLvTjeqhr734CNtrK41L40sUQckmj1lGKQ0rC37x544r8eNXRpnVE3ZZY7zXo8NomiO0ZUCj2uHz58rbXoZ6gc0uA+F6ZeKS/jhRDUq8MKrTho9fEkihMmhxtBI1DxKFY9XLpVcSkfoi8JGnToZO5sU5aiDQIW716ddt7ZLYtMQlhECdBGXZZMWldY5BHm5xgAroWj4C0hbYkSc/jBmggIrXJWlZM6pSETsEPGqZOndr2uuuR5rF169a2HoHPdurUKZM4CO1WTPqaDaAd+GFGKdIQkxAn9RuEWcTRyN2KSUgiSgF5aWzPTeA/lN5rZubMmR2bE4SIC4nJoltgAV/dVefZm72AtctUCJU2CMJ327hxY9t7EHbkyJFseq+EJSY16RPo3Dkq1kkr7+q0bNmyDuLQcZBEPYmHVdOBiJyIlrRDq41YPWfXOxUysi5fvtyaj+2BpcnsUV/oSoEMOk2CQGlr4ckhBwaetBhjCwH0ZHtJROPJkyc7UjcYLDjmrH7ADTEBXFfOYmB0k9oYBOjJ8b4aOYSe7QkKcYhFlq3QYLQhSidNmtS2RATwy8YOM3EQJsUjKiaWZ+vZToUQgzhkHXudb/PW5YMHD9yZM2faPsMwoc7RciYJXbGuBqJ1UIGKKLv915jsvgtJxCZDubdXr165mzdvtr1Hz5LONA8jrUwKPqsmVesKa49S3Q4WxmRPUEYdTjgiUcfUwLx589ySJUva3oMkP6IYddq6HMS4o55xBJBUeRjzfa4Zdeg56QZ43LhxoyPo7Lf1kNt7oO8wWAbNwaYjIv5lhyS7kRf96dvm5Jah8vfvX3flyhX35cuX6HfzFHOToS1H4BenCaHvO8pr8iDuwoUL7tevX+b5ZdbBair0xkFIlFDlW4ZknEClsp/TzXyAKVOmmHWFVSbDNw1l1+4f90U6IY/q4V27dpnE9bJ+v87QEydjqx/UamVVPRG+mwkNTYN+9tjkwzEx+atCm/X9WvWtDtAb68Wy9LXa1UmvCDDIpPkyOQ5ZwSzJ4jMrvFcr0rSjOUh+GcT4LSg5ugkW1Io0/SCDQBojh0hPlaJdah+tkVYrnTZowP8iq1F1TgMBBauufyB33x1v+NWFYmT5KmppgHC+NkAgbmRkpD3yn9QIseXymoTQFGQmIOKTxiZIWpvAatenVqRVXf2nTrAWMsPnKrMZHz6bJq5jvce6QK8J1cQNgKxlJapMPdZSR64/UivS9NztpkVEdKcrs5alhhWP9NeqlfWopzhZScI6QxseegZRGeg5a8C3Re1Mfl1ScP36ddcUaMuv24iOJtz7sbUjTS4qBvKmstYJoUauiuD3k5qhyr7QdUHMeCgLa1Ear9NquemdXgmum4fvJ6w1lqsuDhNrg1qSpleJK7K3TF0Q2jSd94uSZ60kK1e3qyVpQK6PVWXp2/FC3mp6jBhKKOiY2h3gtUV64TWM6wDETRPLDfSakXmH3w8g9Jlug8ZtTt4kVF0kLUYYmCCtD/DrQ5YhMGbA9L3ucdjh0y8kOHW5gU/VEEmJTcL4Pz/f7mgoAbYkAAAAAElFTkSuQmCC"
# },
# },
# ],
# }
# ],
# )
# print("Response from ollama/llava")
# print(response)
# # test_ollama_llava()
# # PROCESSED CHUNK PRE CHUNK CREATOR

20
tests/search_tests/test_google_pse_search.py
View file

@ -1,20 +0,0 @@
"""
Tests for Google Programmable Search Engine (PSE) API integration.
"""
import pytest
from tests.search_tests.base_search_unit_tests import BaseSearchTest
# class TestGooglePSESearch(BaseSearchTest):
# """
# Tests for Google PSE Search functionality.
# """
# def get_search_provider(self) -> str:
# """
# Return search_provider for Google PSE Search.
# """
# return "google_pse"