phoenix/litellm-mirror
0
0
Fork 1
mirror of https://github.com/BerriAI/litellm.git synced 2025-04-24 18:24:20 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions
litellm-mirror/litellm/tests/test_logging.py

382 lines
No EOL
13 KiB
Python
Raw Blame History

# #### 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-instant-1", 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-instant-1", 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-instant-1", 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-instant-1", 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}")
Reference in a new issue View git blame Copy permalink