forked from phoenix/litellm-mirror
27e18358ab
* fix(pattern_match_deployments.py): default to user input if unable to map based on wildcards * test: fix test * test: reset test name * test: update conftest to reload proxy server module between tests * ci(config.yml): move langfuse out of local_testing reduce ci/cd time * ci(config.yml): cleanup langfuse ci/cd tests * fix: update test to not use global proxy_server app module * ci: move caching to a separate test pipeline speed up ci pipeline * test: update conftest to check if proxy_server attr exists before reloading * build(conftest.py): don't block on inability to reload proxy_server * ci(config.yml): update caching unit test filter to work on 'cache' keyword as well * fix(encrypt_decrypt_utils.py): use function to get salt key * test: mark flaky test * test: handle anthropic overloaded errors * refactor: create separate ci/cd pipeline for proxy unit tests make ci/cd faster * ci(config.yml): add litellm_proxy_unit_testing to build_and_test jobs * ci(config.yml): generate prisma binaries for proxy unit tests * test: readd vertex_key.json * ci(config.yml): remove `-s` from proxy_unit_test cmd speed up test * ci: remove any 'debug' logging flag speed up ci pipeline * test: fix test * test(test_braintrust.py): rerun * test: add delay for braintrust test
94 lines
3 KiB
Python
94 lines
3 KiB
Python
# What this tests
|
|
## This tests the proxy server startup
|
|
import sys, os, json
|
|
import traceback
|
|
from dotenv import load_dotenv
|
|
|
|
load_dotenv()
|
|
import os, io
|
|
|
|
# this file is to test litellm/proxy
|
|
|
|
sys.path.insert(
|
|
0, os.path.abspath("../..")
|
|
) # Adds the parent directory to the system path
|
|
import pytest, logging, asyncio
|
|
import litellm
|
|
from litellm.proxy.proxy_server import (
|
|
router,
|
|
save_worker_config,
|
|
initialize,
|
|
startup_event,
|
|
llm_model_list,
|
|
shutdown_event,
|
|
)
|
|
|
|
|
|
def test_proxy_gunicorn_startup_direct_config():
|
|
"""
|
|
gunicorn startup requires the config to be passed in via environment variables
|
|
|
|
We support saving either the config or the dict as an environment variable.
|
|
|
|
Test both approaches
|
|
"""
|
|
try:
|
|
from litellm._logging import verbose_proxy_logger, verbose_router_logger
|
|
import logging
|
|
|
|
# unset set DATABASE_URL in env for this test
|
|
# set prisma client to None
|
|
setattr(litellm.proxy.proxy_server, "prisma_client", None)
|
|
database_url = os.environ.pop("DATABASE_URL", None)
|
|
|
|
verbose_proxy_logger.setLevel(level=logging.DEBUG)
|
|
verbose_router_logger.setLevel(level=logging.DEBUG)
|
|
filepath = os.path.dirname(os.path.abspath(__file__))
|
|
# test with worker_config = config yaml
|
|
config_fp = f"{filepath}/test_configs/test_config_no_auth.yaml"
|
|
os.environ["WORKER_CONFIG"] = config_fp
|
|
asyncio.run(startup_event())
|
|
asyncio.run(shutdown_event())
|
|
except Exception as e:
|
|
if "Already connected to the query engine" in str(e):
|
|
pass
|
|
else:
|
|
pytest.fail(f"An exception occurred - {str(e)}")
|
|
finally:
|
|
# restore DATABASE_URL after the test
|
|
if database_url is not None:
|
|
os.environ["DATABASE_URL"] = database_url
|
|
|
|
|
|
def test_proxy_gunicorn_startup_config_dict():
|
|
try:
|
|
from litellm._logging import verbose_proxy_logger, verbose_router_logger
|
|
import logging
|
|
|
|
verbose_proxy_logger.setLevel(level=logging.DEBUG)
|
|
verbose_router_logger.setLevel(level=logging.DEBUG)
|
|
# unset set DATABASE_URL in env for this test
|
|
# set prisma client to None
|
|
setattr(litellm.proxy.proxy_server, "prisma_client", None)
|
|
database_url = os.environ.pop("DATABASE_URL", None)
|
|
|
|
filepath = os.path.dirname(os.path.abspath(__file__))
|
|
# test with worker_config = config yaml
|
|
config_fp = f"{filepath}/test_configs/test_config_no_auth.yaml"
|
|
# test with worker_config = dict
|
|
worker_config = {"config": config_fp}
|
|
os.environ["WORKER_CONFIG"] = json.dumps(worker_config)
|
|
asyncio.run(startup_event())
|
|
asyncio.run(shutdown_event())
|
|
except Exception as e:
|
|
if "Already connected to the query engine" in str(e):
|
|
pass
|
|
else:
|
|
pytest.fail(f"An exception occurred - {str(e)}")
|
|
finally:
|
|
# restore DATABASE_URL after the test
|
|
if database_url is not None:
|
|
os.environ["DATABASE_URL"] = database_url
|
|
|
|
|
|
# test_proxy_gunicorn_startup()
|