scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-11-11 08:38:24 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 359

Merge ea2ac4a136 into 2e27a02a9f

Browse source
This commit is contained in:
yucheng-berri 2026-10-26 18:48:55 -04:00 • committed by GitHub
commit b8db50f6d1
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
13 changed files with 1174 additions and 105 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

56
litellm/_logging.py
View file

@ -1,6 +1,7 @@
import ast
import contextvars
import functools
import inspect
import itertools
import logging
import os
@ -897,6 +898,61 @@ verbose_proxy_logger.addFilter(_diagnostic_filter)
verbose_proxy_stdout_logger.addFilter(_diagnostic_filter)
verbose_logger.addFilter(_diagnostic_filter)
_ANSI_ESCAPE: Final = re.compile(r"\x1b\[[0-9;]*m")
def print_startup_message(message: str) -> None:
"""Write a boot-time message meant for the operator's terminal.
Plain mode prints it unchanged, colors and spacing included. With JSON logs on, every line of
stdout is read by a log pipeline, so the message goes out as one INFO record of the proxy logger
instead, on stdout like the print whatever handlers the logging config gave that logger, with the
ANSI color codes and surrounding blank lines removed, located at the caller and emitted at any log
level; a blank spacer produces no record.
"""
if not litellm.json_logs:
print(message) # noqa: T201 # plain mode: this is the operator-terminal print it replaces
return
text: Final = _ANSI_ESCAPE.sub("", message).strip("\n")
if not text:
return
frame: Final = inspect.currentframe()
caller: Final = frame.f_back if frame is not None else None
record: Final = verbose_proxy_logger.makeRecord(
verbose_proxy_logger.name,
logging.INFO,
caller.f_code.co_filename if caller is not None else __file__,
caller.f_lineno if caller is not None else 0,
text,
(),
None,
func=caller.f_code.co_name if caller is not None else None,
)
# The print this replaces shows at any log level, so the record skips the level checks of the
# proxy logger (at the root's WARNING in the CLI process until initialize() runs) and of its JSON
# handler (LITELLM_LOG). It also skips the logger's stdout cap: that cap bounds echoed request
# payloads, while a startup message is operator text the print showed whole (the required migration
# SQL, the policy list). It skips secret redaction too, which the print never had and whose sk-
# pattern turns a model named risk-assessment-gpt4o into riREDACTED. The processed record counts as
# scrubbed, so the handler's secret filter and the formatter leave it alone too.
if not _process_record(record, base64_limit=_get_max_base64_length_stdout_log(), text_limit=0, redact=False):
return
# The print wrote to stdout whatever the logging config said, so the record goes through the proxy's
# own JSON handler on stdout rather than the proxy logger's handlers: uvicorn applies an operator
# --log_config after proxy_cli's turn_on_json() and before initialize() re-installs the JSON handler
# through load_config(), and in that window, where the banner and the feedback box are emitted, a
# config that names "LiteLLM Proxy" has left the logger with no handler or with the operator's own.
_startup_json_handler().handle(record)
def _startup_json_handler() -> logging.Handler:
"""The proxy's JSON handler on stdout, filters included, for one startup message."""
handler: Final = LevelRoutingStreamHandler()
handler.setFormatter(JsonFormatter())
handler.addFilter(_secret_filter)
handler.addFilter(_correlation_filter)
return handler
def _suppress_loggers():
"""Suppress noisy loggers at INFO level"""

19
litellm/proxy/db/check_migration.py
View file

@ -4,7 +4,7 @@ import os
import subprocess
from typing import Final
from litellm._logging import verbose_logger
from litellm._logging import print_startup_message, verbose_logger
def extract_sql_commands(diff_output: str) -> list[str]:
@ -63,8 +63,8 @@ def check_prisma_schema_diff_helper(db_url: str) -> tuple[bool, list[str]]:
run_prisma,
)
except ImportError as e:
print( # noqa: T201 # boot-time operator output, same channel as this helper's other messages
f"Skipping the migration diff: litellm-proxy-extras has no Prisma runner. Error: {e}"
print_startup_message(
f"Skipping the migration diff: litellm-proxy-extras has no Prisma runner. Error: {e}",
)
return False, []
@ -89,22 +89,21 @@ def check_prisma_schema_diff_helper(db_url: str) -> tuple[bool, list[str]]:
sql_commands: Final = extract_sql_commands(result.stdout)
if sql_commands:
print("Changes to DB Schema detected") # noqa: T201
print("Required SQL commands:") # noqa: T201
for command in sql_commands:
print(command) # noqa: T201
print_startup_message(
"\n".join(("Changes to DB Schema detected", "Required SQL commands:", *sql_commands)),
)
return True, sql_commands
else:
return False, []
except subprocess.TimeoutExpired:
print( # noqa: T201 # boot-time operator output, same channel as this helper's other messages
print_startup_message(
f"Timed out after {timeout}s generating the migration diff. "
f"Raise {PRISMA_COMMAND_TIMEOUT_ENV_VAR} if this database needs longer."
f"Raise {PRISMA_COMMAND_TIMEOUT_ENV_VAR} if this database needs longer.",
)
return False, []
except subprocess.CalledProcessError as e:
error_message: Final = f"Failed to generate migration diff. Error: {e.stderr}"
print(error_message) # noqa: T201
print_startup_message(error_message)
return False, []

35
litellm/proxy/policy_engine/init_policies.py
View file

@ -6,10 +6,11 @@ Configuration structure:
- policy_attachments: Define WHERE policies apply (teams, keys, models)
"""
import sys
from collections.abc import Mapping, Sequence
from typing import TYPE_CHECKING, Any, Final, Optional
from litellm._logging import verbose_proxy_logger
from litellm._logging import print_startup_message, verbose_proxy_logger
from litellm.proxy.policy_engine.attachment_registry import get_attachment_registry
from litellm.proxy.policy_engine.policy_registry import get_policy_registry
from litellm.proxy.policy_engine.policy_validator import PolicyValidator
@ -32,12 +33,9 @@ def _print_policies_on_startup(
"""
Print loaded policies to console on startup (similar to model list).
"""
import sys
print( # noqa: T201
lines: Final = [
f"{_green_color_code}\nLiteLLM Policy Engine: Loaded {len(policies_config)} policies{_reset_color_code}\n"
)
sys.stdout.flush()
]
for policy_name, policy_data in policies_config.items():
guardrails = policy_data.get("guardrails", {})
@ -49,23 +47,21 @@ def _print_policies_on_startup(
guardrails_remove = guardrails.get("remove", []) if isinstance(guardrails, dict) else []
inherit_str = f" (inherits: {inherit})" if inherit else ""
print( # noqa: T201
f"{_blue_color_code} - {policy_name}{inherit_str}{_reset_color_code}"
)
lines.append(f"{_blue_color_code} - {policy_name}{inherit_str}{_reset_color_code}")
if description:
print(f" description: {description}") # noqa: T201
lines.append(f" description: {description}")
if guardrails_add:
print(f" guardrails.add: {guardrails_add}") # noqa: T201
lines.append(f" guardrails.add: {guardrails_add}")
if guardrails_remove:
print(f" guardrails.remove: {guardrails_remove}") # noqa: T201
lines.append(f" guardrails.remove: {guardrails_remove}")
if condition:
model_condition = condition.get("model") if isinstance(condition, dict) else None
if model_condition:
print(f" condition.model: {model_condition}") # noqa: T201
lines.append(f" condition.model: {model_condition}")
# Print attachments
if policy_attachments_config:
print( # noqa: T201
lines.append(
f"\n{_yellow_color_code}Policy Attachments: {len(policy_attachments_config)} attachment(s){_reset_color_code}"
)
for attachment in policy_attachments_config:
@ -86,13 +82,14 @@ def _print_policies_on_startup(
scope_parts.append(f"models={models}")
scope_str = ", ".join(scope_parts) if scope_parts else "all"
print(f" - {policy} -> {scope_str}") # noqa: T201
lines.append(f" - {policy} -> {scope_str}")
lines.append("")
print_startup_message("\n".join(lines))
else:
print( # noqa: T201
f"\n{_yellow_color_code}Warning: No policy_attachments configured. Policies will not be applied to any requests.{_reset_color_code}"
print_startup_message("\n".join(lines))
print_startup_message(
f"\n{_yellow_color_code}Warning: No policy_attachments configured. Policies will not be applied to any requests.{_reset_color_code}\n",
)
print() # noqa: T201
sys.stdout.flush()

116
litellm/proxy/proxy_cli.py
View file

@ -1,4 +1,5 @@
# ruff: noqa: T201
import functools
import importlib
import json
import os
@ -7,7 +8,7 @@ import re
import subprocess
import sys
import urllib.parse as urlparse
from collections.abc import Iterable, Mapping, Sequence
from collections.abc import Callable, Iterable, Mapping, Sequence
from pathlib import Path
from typing import TYPE_CHECKING, Any, Final
@ -18,6 +19,7 @@ from dotenv import load_dotenv
from pydantic import ConfigDict
import litellm
from litellm._logging import print_startup_message
from litellm.constants import DEFAULT_NUM_WORKERS_LITELLM_PROXY
from litellm.proxy.db.pgbouncer import (
PgBouncerError,
@ -202,6 +204,29 @@ def deprecated_v2_flag_passed_on_cli() -> bool:
return ctx.get_parameter_source("use_v2_migration_resolver") is ParameterSource.COMMANDLINE
def _print_enforce_prisma_migration_check_notice() -> None:
print_startup_message(
"\033[1;33mLiteLLM Proxy: --enforce_prisma_migration_check is "
"deprecated and has no effect, because the proxy always exits "
"when database setup fails at startup. You can safely remove it.\033[0m",
)
def _print_or_hold_enforce_prisma_migration_check_notice(
*, reads_config_file: bool, returns_before_config: bool
) -> Callable[[], None] | None:
"""Print the notice now, or, while a config file can still switch json_logs on, return a printer
to call once it is read; a boot that fails before that prints the held notice on the way out."""
if not reads_config_file or litellm.json_logs or returns_before_config:
_print_enforce_prisma_migration_check_notice()
return None
print_once: Final = functools.cache(_print_enforce_prisma_migration_check_notice)
ctx: Final = click.get_current_context(silent=True)
if ctx is not None:
ctx.call_on_close(print_once)
return print_once
class ProxyInitializationHelpers:
@staticmethod
def echo_litellm_version():
@ -318,7 +343,7 @@ class ProxyInitializationHelpers:
"timeout_graceful_shutdown": GracefulShutdownManager.get_timeout(),
}
if log_config is not None:
print(f"Using log_config: {log_config}")
print_startup_message(f"Using log_config: {log_config}")
uvicorn_args["log_config"] = log_config
elif litellm.json_logs:
# Use JSON log config for uvicorn to ensure all logs (including exceptions) are JSON
@ -331,10 +356,10 @@ class ProxyInitializationHelpers:
if "timeout_worker_healthcheck" in inspect.signature(uvicorn.Config.__init__).parameters:
uvicorn_args["timeout_worker_healthcheck"] = timeout_worker_healthcheck
else:
print(
print_startup_message(
f"\033[1;33mLiteLLM Proxy: --timeout_worker_healthcheck "
f"requires uvicorn>=0.37.0, but installed uvicorn=={uvicorn.__version__}. "
f"Ignoring the flag.\033[0m"
f"Ignoring the flag.\033[0m",
)
return uvicorn_args
@ -354,18 +379,18 @@ class ProxyInitializationHelpers:
import uvicorn
if max_requests_before_restart is None:
print(
print_startup_message(
"\033[1;33mLiteLLM Proxy: --max_requests_before_restart_jitter "
"has no effect without --max_requests_before_restart\033[0m\n"
"has no effect without --max_requests_before_restart\033[0m\n",
)
return
if "limit_max_requests_jitter" in inspect.signature(uvicorn.Config.__init__).parameters:
uvicorn_args["limit_max_requests_jitter"] = jitter
else:
print(
print_startup_message(
f"\033[1;33mLiteLLM Proxy: --max_requests_before_restart_jitter "
f"requires uvicorn>=0.41.0, but installed uvicorn=={uvicorn.__version__}. "
f"Ignoring the flag.\033[0m"
f"Ignoring the flag.\033[0m",
)
_apply_uvicorn_max_requests_jitter = apply_uvicorn_max_requests_jitter
@ -489,12 +514,12 @@ class ProxyInitializationHelpers:
from hypercorn.asyncio import serve
from hypercorn.config import Config
print(f"\033[1;32mLiteLLM Proxy: Starting server on {host}:{port} using Hypercorn\033[0m\n")
print_startup_message(f"\033[1;32mLiteLLM Proxy: Starting server on {host}:{port} using Hypercorn\033[0m\n")
config: Final = Config()
config.bind = [f"{host}:{port}"]
if ssl_certfile_path is not None and ssl_keyfile_path is not None:
print(
print_startup_message(
f"\033[1;32mLiteLLM Proxy: Using SSL with certfile: {ssl_certfile_path} and keyfile: {ssl_keyfile_path}\033[0m\n"
)
config.certfile = ssl_certfile_path
@ -527,14 +552,14 @@ class ProxyInitializationHelpers:
from granian import Granian
from granian.constants import Interfaces
print(f"\033[1;32mLiteLLM Proxy: Starting server on {host}:{port} using Granian\033[0m\n")
print_startup_message(f"\033[1;32mLiteLLM Proxy: Starting server on {host}:{port} using Granian\033[0m\n")
if max_requests_before_restart is not None:
print(
print_startup_message(
"\033[1;33mLiteLLM: --max_requests_before_restart is not supported by Granian "
"(Granian uses workers_lifetime in seconds, not a per-request limit).\033[0m\n"
"(Granian uses workers_lifetime in seconds, not a per-request limit).\033[0m\n",
)
if ciphers is not None:
print("\033[1;33mLiteLLM: --ciphers is not applied when using --run_granian.\033[0m\n")
print_startup_message("\033[1;33mLiteLLM: --ciphers is not applied when using --run_granian.\033[0m\n")
kwargs: Final[dict[str, object]] = {
"target": "litellm.proxy.proxy_server:app",
@ -547,7 +572,7 @@ class ProxyInitializationHelpers:
if granian_runtime_threads is not None:
kwargs["runtime_threads"] = granian_runtime_threads
if ssl_certfile_path is not None and ssl_keyfile_path is not None:
print(
print_startup_message(
f"\033[1;32mLiteLLM Proxy: Using SSL with certfile: {ssl_certfile_path} and keyfile: {ssl_keyfile_path}\033[0m\n"
)
kwargs["ssl_cert"] = Path(ssl_certfile_path)
@ -608,13 +633,15 @@ class ProxyInitializationHelpers:
\n
"""
)
print()
print(
print_startup_message("")
print_startup_message(
'\033[1;34mLiteLLM: Test your local proxy with: "litellm --test" This runs an openai.ChatCompletion request to your proxy [In a new terminal tab]\033[0m\n'
)
print(f"\033[1;34mLiteLLM: Curl Command Test for your local proxy\n {curl_command} \033[0m\n")
print("\033[1;34mDocs: https://docs.litellm.ai/docs/simple_proxy\033[0m\n")
print(f"\033[1;34mSee all Router/Swagger docs on http://0.0.0.0:{port} \033[0m\n")
print_startup_message(
f"\033[1;34mLiteLLM: Curl Command Test for your local proxy\n {curl_command} \033[0m\n"
)
print_startup_message("\033[1;34mDocs: https://docs.litellm.ai/docs/simple_proxy\033[0m\n")
print_startup_message(f"\033[1;34mSee all Router/Swagger docs on http://0.0.0.0:{port} \033[0m\n")
def load_config(self):
# note: This Loads the gunicorn config - has nothing to do with LiteLLM Proxy config
@ -634,7 +661,9 @@ class ProxyInitializationHelpers:
# gunicorn app function
return self.application
print(f"\033[1;32mLiteLLM Proxy: Starting server on {host}:{port} with {num_workers} workers\033[0m\n")
print_startup_message(
f"\033[1;32mLiteLLM Proxy: Starting server on {host}:{port} with {num_workers} workers\033[0m\n"
)
graceful_shutdown_timeout: Final = GracefulShutdownManager.get_timeout()
gunicorn_options: Final = {
"bind": f"{host}:{port}",
@ -652,9 +681,9 @@ class ProxyInitializationHelpers:
gunicorn_options["max_requests"] = max_requests_before_restart
if max_requests_before_restart_jitter is not None:
if max_requests_before_restart is None:
print(
print_startup_message(
"\033[1;33mLiteLLM Proxy: --max_requests_before_restart_jitter "
"has no effect without --max_requests_before_restart\033[0m\n"
"has no effect without --max_requests_before_restart\033[0m\n",
)
else:
gunicorn_options["max_requests_jitter"] = max_requests_before_restart_jitter
@ -669,7 +698,7 @@ class ProxyInitializationHelpers:
gunicorn_options["child_exit"] = child_exit
if ssl_certfile_path is not None and ssl_keyfile_path is not None:
print(
print_startup_message(
f"\033[1;32mLiteLLM Proxy: Using SSL with certfile: {ssl_certfile_path} and keyfile: {ssl_keyfile_path}\033[0m\n"
)
gunicorn_options["certfile"] = ssl_certfile_path
@ -693,7 +722,7 @@ class ProxyInitializationHelpers:
with open(os.devnull, "w") as devnull:
subprocess.Popen(command, stdout=devnull, stderr=devnull)
except Exception as e:
print(f"""
print_startup_message(f"""
LiteLLM Warning: proxy started with `ollama` model\n`ollama serve` failed with Exception{e}. \nEnsure you run `ollama serve`
""")
@ -762,7 +791,7 @@ class ProxyInitializationHelpers:
os.makedirs(multiproc_dir, exist_ok=True)
wipe_directory(multiproc_dir)
action: Final = "Using existing" if configured_dir else "Auto-created"
print(f"LiteLLM: {action} PROMETHEUS_MULTIPROC_DIR={multiproc_dir}")
print_startup_message(f"LiteLLM: {action} PROMETHEUS_MULTIPROC_DIR={multiproc_dir}")
return multiproc_dir
_maybe_setup_prometheus_multiproc_dir = maybe_setup_prometheus_multiproc_dir
@ -1172,12 +1201,13 @@ def run_server(
if validate_config is True:
ProxyInitializationHelpers.run_config_validation(config)
return
if enforce_prisma_migration_check:
print(
"\033[1;33mLiteLLM Proxy: --enforce_prisma_migration_check is "
"deprecated and has no effect, because the proxy always exits "
"when database setup fails at startup. You can safely remove it.\033[0m"
held_prisma_notice: Final = (
_print_or_hold_enforce_prisma_migration_check_notice(
reads_config_file=config is not None, returns_before_config=health is True or test is True
)
if enforce_prisma_migration_check
else None
)
if model and "ollama" in model and api_base is None:
ProxyInitializationHelpers.run_ollama_serve()
if health is True:
@ -1293,8 +1323,14 @@ def run_server(
import litellm
litellm.json_logs = True
# Workers that uvicorn or granian spawn import litellm afresh and read this at import,
# so they log JSON from their first record instead of plain text until load_config
# reaches the json_logs key.
os.environ["JSON_LOGS"] = "true"
litellm.turn_on_json()
if held_prisma_notice is not None:
held_prisma_notice()
### GENERAL SETTINGS ###
general_settings = _config.get("general_settings", {})
if general_settings is None:
@ -1461,7 +1497,7 @@ def run_server(
lifetime_params,
)
)
print(
print_startup_message(
postgres_connection_budget_message(
writer_limit=_url_query_value(os.getenv("DATABASE_URL"), "connection_limit")
or f"{db_connection_pool_limit}",
@ -1493,19 +1529,19 @@ def run_server(
env_value=os.getenv("USE_V2_MIGRATION_RESOLVER"),
)
if deprecated_v2_flag_passed_on_cli() and use_v2_resolver:
print(
print_startup_message(
"\033[1;33mLiteLLM Proxy: --use_v2_migration_resolver is "
"deprecated and has no effect, because the v2 migration "
"resolver is now the default. You can safely remove it. To "
"opt back into the legacy v1 resolver, pass "
"--use_legacy_migration_resolver.\033[0m"
"--use_legacy_migration_resolver.\033[0m",
)
if not use_v2_resolver:
print(
print_startup_message(
"\033[1;33mLiteLLM Proxy: Using the legacy (v1) migration "
"resolver. It performs the diff-and-force recovery that can "
"cause schema thrashing during rolling deploys where two "
"LiteLLM versions contend for the same DB.\033[0m"
"LiteLLM versions contend for the same DB.\033[0m",
)
try:
migrated: Final = PrismaManager.setup_database(
@ -1573,7 +1609,7 @@ def run_server(
# Skip server startup if requested (after all setup is done)
if skip_server_startup:
print("LiteLLM: Setup complete. Skipping server startup as requested.")
print_startup_message("LiteLLM: Setup complete. Skipping server startup as requested.")
return
# Auto-create PROMETHEUS_MULTIPROC_DIR for multi-worker setups
@ -1592,7 +1628,7 @@ def run_server(
)
except MetricsServerStartupError as error:
raise click.ClickException(str(error)) from error
print(
print_startup_message(
f"\033[1;32mLiteLLM: Serving Prometheus metrics on {host}:{prometheus_metrics_port}/metrics "
f"(pid {metrics_process.pid})\033[0m"
)
@ -1615,7 +1651,7 @@ def run_server(
run_gunicorn=run_gunicorn is True,
)
if single_worker_warning is not None:
print(f"\033[1;33m{single_worker_warning}\033[0m\n")
print_startup_message(f"\033[1;33m{single_worker_warning}\033[0m\n")
if run_gunicorn is False and run_hypercorn is False and run_granian is False:
if limit_concurrency is not None:
uvicorn_args["limit_concurrency"] = limit_concurrency
@ -1626,7 +1662,7 @@ def run_server(
jitter=max_requests_before_restart_jitter,
)
if ssl_certfile_path is not None and ssl_keyfile_path is not None:
print(
print_startup_message(
f"\033[1;32mLiteLLM Proxy: Using SSL with certfile: {ssl_certfile_path} and keyfile: {ssl_keyfile_path}\033[0m\n"
)
uvicorn_args["ssl_keyfile"] = ssl_keyfile_path

72
litellm/proxy/proxy_server.py
View file

@ -273,25 +273,27 @@ def generate_feedback_box():
# Select a random message
message: Final = random.choice(list_of_messages)
print() # noqa: T201
print("\033[1;37m" + "#" + "-" * box_width + "#\033[0m") # noqa: T201
print("\033[1;37m" + "#" + " " * box_width + "#\033[0m") # noqa: T201
print("\033[1;37m" + f"# {message:^59} #\033[0m") # noqa: T201
print( # noqa: T201
"\033[1;37m" + "# {:^59} #\033[0m".format("https://github.com/BerriAI/litellm/issues/new")
print_startup_message(
"\n".join(
(
"",
"\033[1;37m" + "#" + "-" * box_width + "#\033[0m",
"\033[1;37m" + "#" + " " * box_width + "#\033[0m",
"\033[1;37m" + f"# {message:^59} #\033[0m",
"\033[1;37m" + "# {:^59} #\033[0m".format("https://github.com/BerriAI/litellm/issues/new"),
"\033[1;37m" + "#" + " " * box_width + "#\033[0m",
"\033[1;37m" + "#" + "-" * box_width + "#\033[0m",
"",
" Thank you for using LiteLLM! - Krrish & Ishaan",
"",
"",
"",
"\033[1;31mGive Feedback / Get Help: https://github.com/BerriAI/litellm/issues/new\033[0m",
"",
"",
)
)
)
print("\033[1;37m" + "#" + " " * box_width + "#\033[0m") # noqa: T201
print("\033[1;37m" + "#" + "-" * box_width + "#\033[0m") # noqa: T201
print() # noqa: T201
print(" Thank you for using LiteLLM! - Krrish & Ishaan") # noqa: T201
print() # noqa: T201
print() # noqa: T201
print() # noqa: T201
print( # noqa: T201
"\033[1;31mGive Feedback / Get Help: https://github.com/BerriAI/litellm/issues/new\033[0m"
)
print() # noqa: T201
print() # noqa: T201
import contextlib
@ -304,6 +306,7 @@ import litellm._redis
from litellm import Router
from litellm._internal_context import service_target, with_service_target
from litellm._logging import (
print_startup_message,
redact_string,
verbose_proxy_logger,
verbose_router_logger,
@ -6412,17 +6415,13 @@ class ProxyConfig:
search_tools_parsed: Final[list[SearchToolTypedDict]] = []
print( # noqa: T201
"\033[32mLiteLLM: Proxy initialized with Search Tools:\033[0m"
)
print_startup_message("\033[32mLiteLLM: Proxy initialized with Search Tools:\033[0m")
for search_tool in search_tools_raw:
# Display loaded search tool
search_tool_name = search_tool.get("search_tool_name", "")
search_provider = search_tool.get("litellm_params", {}).get("search_provider", "")
print( # noqa: T201
f"\033[32m {search_tool_name} ({search_provider})\033[0m"
)
print_startup_message(f"\033[32m {search_tool_name} ({search_provider})\033[0m")
# Handle os.environ/ variables in litellm_params
litellm_params = search_tool.get("litellm_params", {})
@ -6598,7 +6597,7 @@ class ProxyConfig:
reset_color_code: Final = "\033[0m"
for key, value in litellm_settings.items():
if key == "cache" and value is True:
print(f"{blue_color_code}\nSetting Cache on Proxy") # noqa: T201
print_startup_message(f"{blue_color_code}\nSetting Cache on Proxy")
from litellm.caching.caching import Cache
cache_params = {}
@ -6789,7 +6788,7 @@ class ProxyConfig:
if PrometheusLogger is not None:
verbose_proxy_logger.debug("mounting metrics endpoint")
PrometheusLogger._mount_metrics_endpoint()
print( # noqa: T201
print_startup_message(
f"{blue_color_code} Initialized Success Callbacks - {litellm.success_callback} {reset_color_code}"
)
elif key == "failure_callback":
@ -6808,7 +6807,7 @@ class ProxyConfig:
# these are litellm callbacks - "langfuse", "sentry", "wandb"
else:
litellm.logging_callback_manager.add_litellm_failure_callback(callback)
print( # noqa: T201
print_startup_message(
f"{blue_color_code} Initialized Failure Callbacks - {litellm.failure_callback} {reset_color_code}"
)
elif key == "audit_log_callbacks":
@ -6833,7 +6832,7 @@ class ProxyConfig:
_store_audit_logs = litellm_settings.get("store_audit_logs", litellm.store_audit_logs)
if is_audit_logging_enabled(store_audit_logs=_store_audit_logs):
print( # noqa: T201
print_startup_message(
f"{blue_color_code} Initialized Audit Log Callbacks - {litellm.audit_log_callbacks} {reset_color_code}"
)
else:
@ -7232,9 +7231,7 @@ class ProxyConfig:
if model_list:
router_params["model_list"] = model_list
validate_auto_router_capability_limits(model_list, limit=_license_check.auto_router_capability_limit())
print( # noqa: T201
"\033[32mLiteLLM: Proxy initialized with Config, Set models:\033[0m"
)
print_startup_message("\033[32mLiteLLM: Proxy initialized with Config, Set models:\033[0m")
for model in model_list:
### LOAD FROM os.environ/ ###
for k, v in model["litellm_params"].items():
@ -7251,7 +7248,7 @@ class ProxyConfig:
complexity_router_config=complexity_router_config,
config_file_path=config_file_path,
)
print(f"\033[32m {model.get('model_name', '')}\033[0m") # noqa: T201
print_startup_message(f"\033[32m {model.get('model_name', '')}\033[0m")
litellm_model_name = model["litellm_params"]["model"]
litellm_model_api_base = model["litellm_params"].get("api_base", None)
if "ollama" in litellm_model_name and litellm_model_api_base is None:
@ -9672,9 +9669,14 @@ async def initialize(
master_key, \
user_custom_auth, \
prisma_client
from litellm.proxy.common_utils.banner import show_banner
from litellm.proxy.common_utils.banner import LITELLM_BANNER, show_banner
show_banner()
# show_banner() also serves the `lite` shell, which runs against an uninitialized SDK (#46803), so
# the JSON-mode choice stays out of it.
if litellm.json_logs:
print_startup_message(LITELLM_BANNER)
else:
show_banner()
if os.getenv("LITELLM_DONT_SHOW_FEEDBACK_BOX", "").lower() != "true":
generate_feedback_box()
user_model = model
@ -11930,7 +11932,7 @@ class ProxyStartupEvent:
and proxy_logging_obj.slack_alerting_instance.alerting is not None
and prisma_client is not None
):
print("Alerting: Initializing Weekly/Monthly Spend Reports") # noqa: T201
print_startup_message("Alerting: Initializing Weekly/Monthly Spend Reports")
spend_report_frequency: Final[str] = general_settings.get("spend_report_frequency", "7d") or "7d"
days: Final = int(spend_report_frequency[:-1])

2
tests/integration/_support/process.py
View file

@ -987,6 +987,7 @@ def refused_boot_log(
overrides: Mapping[str, str],
*,
config: Path | None = None,
extra_arguments: tuple[str, ...] = (),
) -> str:
"""Start the proxy and return its log once it exits non-zero instead of becoming ready."""
root: Final = _proxy_root()
@ -1004,6 +1005,7 @@ def refused_boot_log(
"--num_workers",
"1",
*DB_PUSH,
*extra_arguments,
)
launch: Final = _launch(command, root, environment, output)
try:

409
tests/integration/configuration/test_json_logs_startup.py Normal file
View file

@ -0,0 +1,409 @@
"""JSON logging keeps every line the proxy writes during boot, serving and shutdown machine-parseable.
A deployment that turns on JSON logs ships the proxy's stdout to a log pipeline, and every line that is
not a JSON object breaks that pipeline's parser. This contract boots an owned proxy with JSON logs on,
serves a chat request on every configured model plus one rejected request, stops the proxy and
rejects any line of its output that is not one JSON record, while checking that the
startup information those lines used to carry is still present as JSON records. JSON mode is switched
on through the environment and through the config, at one and two uvicorn workers, at the INFO, the
default and the WARNING log level, under a concurrent burst that lands on both workers, under an operator
--log_config that names the proxy logger or gives it a handler of its own, and with a policy list longer
than the stdout cap. The model list keeps a model name the secret patterns would match. The deprecated
--enforce_prisma_migration_check notice is one JSON record under a config-only json_logs, and a boot whose
config cannot be loaded still prints it before failing.
"""
from __future__ import annotations
import re
import uuid
from collections.abc import Mapping
from concurrent.futures import ThreadPoolExecutor
from pathlib import Path
from typing import Final
import pytest
import yaml
from pydantic import JsonValue, TypeAdapter, ValidationError
from tests.integration._support.client import Gateway
from tests.integration._support.process import owned_proxy_process, refused_boot_log
JSON_RECORD: Final = TypeAdapter(dict[str, JsonValue])
PLAIN_LOG_LINE: Final = re.compile(r"^\d\d:\d\d:\d\d - LiteLLM")
# prisma-client-py starts the Prisma query engine on the proxy's own stdout, and the engine announces
# its build hash there in two writes (the text, then the newline), so with two workers the two
# announcements land glued together or glued to a neighbor's line. The text is not litellm's and is
# removed before the lines are read.
PRISMA_ENGINE_TEXT: Final = re.compile(r"query-engine [0-9a-f]{40}")
STARTED_WORKER: Final = "Started server process"
MODEL_LIST_HEADER: Final = "Proxy initialized with Config, Set models:"
CACHE_LINE: Final = "Setting Cache on Proxy"
# printed by the CLI process before any worker exists, so it proves the launcher itself logs JSON
CONNECTION_BUDGET_LINE: Final = "Postgres connection budget"
REDIS_AUTH_CACHE_LINE: Final = "enable_redis_auth_cache"
BANNER_FRAGMENT: Final = "██╗"
FEEDBACK_BOX_LINE: Final = "Give Feedback / Get Help"
WORKER_COUNTS: Final = (1, 2)
UNKNOWN_MODEL_STATUS: Final = 400
BURST_SIZE: Final = 16
CACHE_SETTINGS: Final[Mapping[str, JsonValue]] = {"cache": True, "cache_params": {"type": "redis"}}
# the policy list and the required migration SQL are the startup listings that outgrow the stdout cap
# (MAX_STRING_LENGTH_STDOUT_LOG, 4096 by default) that bounds echoed request payloads
STDOUT_CAP: Final = 4096
POLICY_COUNT: Final = 60
POLICY_LIST_HEADER: Final = "LiteLLM Policy Engine: Loaded"
TRUNCATION_MARKER: Final = "litellm_truncated"
DEPRECATED_FLAG: Final = "--enforce_prisma_migration_check"
DEPRECATED_FLAG_NOTICE: Final = "--enforce_prisma_migration_check is deprecated and has no effect"
UNPARSEABLE_CONFIG: Final = "model_list: [\n - bad: yaml: here\n"
# an operator logging config of the usual shape: it sets the proxy logger's level and gives it no handler
# of its own, and formats everything else, uvicorn included, as JSON through the root handler
OPERATOR_LOG_CONFIG: Final[Mapping[str, JsonValue]] = {
"version": 1,
"disable_existing_loggers": False,
"formatters": {"json": {"()": "litellm._logging.JsonFormatter"}},
"handlers": {"stdout": {"class": "logging.StreamHandler", "stream": "ext://sys.stdout", "formatter": "json"}},
"loggers": {"LiteLLM Proxy": {"level": "INFO"}},
"root": {"level": "INFO", "handlers": ["stdout"]},
}
# an operator logging config that silences the proxy logger with a handler of its own and keeps the rest JSON
OPERATOR_SILENCING_LOG_CONFIG: Final[Mapping[str, JsonValue]] = {
"version": 1,
"disable_existing_loggers": False,
"formatters": {"json": {"()": "litellm._logging.JsonFormatter"}},
"handlers": {
"stdout": {"class": "logging.StreamHandler", "stream": "ext://sys.stdout", "formatter": "json"},
"silence": {"class": "logging.NullHandler"},
},
"loggers": {"LiteLLM Proxy": {"level": "INFO", "handlers": ["silence"], "propagate": False}},
"root": {"level": "INFO", "handlers": ["stdout"]},
}
def _models() -> tuple[str, str]:
stamp: Final = uuid.uuid4().hex
return (f"json-logs-first-{stamp}", f"json-logs-second-{stamp}")
def _policies() -> dict[str, JsonValue]:
stamp: Final = uuid.uuid4().hex
return {
f"startup-listing-policy-{n:02d}-{stamp}": {
"description": f"policy {n:02d} of a listing that runs past the stdout cap meant for request payloads"
}
for n in range(POLICY_COUNT)
}
def _write_config(
directory: Path,
upstream_url: str,
models: tuple[str, ...],
litellm_settings: Mapping[str, JsonValue],
extra_sections: Mapping[str, JsonValue] | None = None,
) -> Path:
config: Final = directory / f"json_logs_{uuid.uuid4().hex}.yaml"
config.write_text(
yaml.safe_dump(
{
"model_list": [
{
"model_name": model,
"litellm_params": {
"model": f"openai/{model}",
"api_base": f"{upstream_url}/v1",
"api_key": "integration-provider-key",
},
}
for model in models
],
"litellm_settings": dict(litellm_settings),
"general_settings": {
"master_key": "os.environ/LITELLM_MASTER_KEY",
"database_url": "os.environ/DATABASE_URL",
"disable_prisma_schema_update": True,
},
**(extra_sections or {}),
},
sort_keys=False,
)
)
return config
def _record(line: str) -> dict[str, JsonValue] | None:
stripped: Final = line.strip()
if not stripped.startswith("{"):
return None
try:
return JSON_RECORD.validate_json(stripped)
except ValidationError:
return None
def _lines(log: str) -> tuple[str, ...]:
return tuple(line for line in PRISMA_ENGINE_TEXT.sub("", log).splitlines() if line.strip())
def _messages(log: str) -> tuple[str, ...]:
records: Final = tuple(record for record in (_record(line) for line in _lines(log)) if record is not None)
return tuple(str(record["message"]) for record in records if "message" in record)
def _non_json_lines(log: str) -> tuple[str, ...]:
return tuple(line for line in _lines(log) if _record(line) is None)
def _boot_serve_and_stop(
gateway: Gateway,
directory: Path,
config: Path,
overrides: Mapping[str, str],
workers: int,
models: tuple[str, ...],
extra_arguments: tuple[str, ...] = (),
) -> str:
with owned_proxy_process(
gateway,
directory,
overrides,
config=config,
workers=workers,
database_setup=(),
extra_arguments=extra_arguments,
) as owned:
_serve(owned.gateway, models)
log: Final = owned.log
return log.read_text()
def _serve(gateway: Gateway, models: tuple[str, ...]) -> None:
"""One chat per model and one rejected request, so the log holds the request-time lines of served
and failed requests next to the boot lines."""
for model in models:
reply: Final = gateway.chat(model, text=f"json logs {uuid.uuid4().hex}")
assert reply["model"] == model, reply
rejected: Final = gateway.request(
"POST",
"/v1/chat/completions",
{"model": f"missing-{uuid.uuid4().hex}", "messages": [{"role": "user", "content": "json logs"}]},
)
assert rejected.status_code == UNKNOWN_MODEL_STATUS, rejected.text
def _assert_every_line_is_json(
log: str, workers: int, models: tuple[str, ...], *, redis_auth_cache_line: bool = True
) -> None:
plaintext: Final = _non_json_lines(log)
assert plaintext == (), "plaintext lines with JSON logs on:\n" + "\n".join(plaintext)
messages: Final = _messages(log)
assert sum(STARTED_WORKER in message for message in messages) == workers, messages
assert any(MODEL_LIST_HEADER in message for message in messages), messages
for model in models:
assert any(model in message for message in messages), (model, messages)
assert any(CACHE_LINE in message for message in messages), messages
if redis_auth_cache_line:
assert any(REDIS_AUTH_CACHE_LINE in message for message in messages), messages
assert any(CONNECTION_BUDGET_LINE in message for message in messages), messages
assert sum(BANNER_FRAGMENT in message for message in messages) == workers, messages
assert sum(FEEDBACK_BOX_LINE in message for message in messages) == workers, messages
@pytest.mark.parametrize("workers", WORKER_COUNTS)
def test_json_logs_env_keeps_every_line_json(gateway: Gateway, tmp_path: Path, workers: int) -> None:
models: Final = _models()
config: Final = _write_config(tmp_path, gateway.upstream_url, models, CACHE_SETTINGS)
log: Final = _boot_serve_and_stop(
gateway, tmp_path, config, {"JSON_LOGS": "true", "LITELLM_LOG": "INFO"}, workers, models
)
_assert_every_line_is_json(log, workers, models)
@pytest.mark.parametrize("workers", WORKER_COUNTS)
def test_json_logs_config_keeps_every_line_json(gateway: Gateway, tmp_path: Path, workers: int) -> None:
models: Final = _models()
# cache is listed before json_logs on purpose: the cache branch logs the Redis auth-cache line
# while the config is still being read, so JSON mode must not depend on the key order.
config: Final = _write_config(tmp_path, gateway.upstream_url, models, {**CACHE_SETTINGS, "json_logs": True})
log: Final = _boot_serve_and_stop(gateway, tmp_path, config, {"LITELLM_LOG": "INFO"}, workers, models)
_assert_every_line_is_json(log, workers, models)
@pytest.mark.parametrize("workers", WORKER_COUNTS)
def test_json_logs_with_operator_log_config_keeps_every_line_json(
gateway: Gateway, tmp_path: Path, workers: int
) -> None:
"""uvicorn applies an operator `--log_config` after the CLI process switched JSON on and before
`initialize()` switches it on again through the config; a config that names the proxy logger leaves it
without handlers in that window, where the banner and the feedback box are emitted. They still arrive,
as JSON, one per worker, and the operator's own formatter keeps uvicorn's lines JSON. The Redis
auth-cache line is a regular logger line written in that same window; this log_config removed the
proxy logger's handlers and left propagation off, so it is lost at one worker on main and here alike."""
models: Final = _models()
config: Final = _write_config(tmp_path, gateway.upstream_url, models, {**CACHE_SETTINGS, "json_logs": True})
log_config: Final = tmp_path / "operator_log_config.yaml"
log_config.write_text(yaml.safe_dump(dict(OPERATOR_LOG_CONFIG)))
log: Final = _boot_serve_and_stop(
gateway, tmp_path, config, {"LITELLM_LOG": "INFO"}, workers, models, ("--log_config", str(log_config))
)
_assert_every_line_is_json(log, workers, models, redis_auth_cache_line=False)
@pytest.mark.parametrize("workers", WORKER_COUNTS)
def test_json_logs_with_operator_handler_on_the_proxy_logger_keeps_the_startup_notices(
gateway: Gateway, tmp_path: Path, workers: int
) -> None:
"""An operator `--log_config` that gives the proxy logger a handler of its own, here a NullHandler that
silences it, is in force in the window where the banner and the feedback box are emitted. The prints
they replace wrote to stdout whatever the logging config said, so both still arrive there as JSON, one
per worker. The Redis auth-cache line is a regular logger line written in that same window and goes
to the operator's handler, so it is silenced on main and here alike."""
models: Final = _models()
config: Final = _write_config(tmp_path, gateway.upstream_url, models, {**CACHE_SETTINGS, "json_logs": True})
log_config: Final = tmp_path / "operator_silencing_log_config.yaml"
log_config.write_text(yaml.safe_dump(dict(OPERATOR_SILENCING_LOG_CONFIG)))
log: Final = _boot_serve_and_stop(
gateway, tmp_path, config, {"LITELLM_LOG": "INFO"}, workers, models, ("--log_config", str(log_config))
)
_assert_every_line_is_json(log, workers, models, redis_auth_cache_line=False)
def test_json_logs_keeps_a_startup_listing_longer_than_the_stdout_cap_whole(gateway: Gateway, tmp_path: Path) -> None:
"""A startup listing longer than MAX_STRING_LENGTH_STDOUT_LOG reaches the pipeline whole. The policy list
stands in for every such listing (the required migration SQL on a drifted database is the other one):
the print it replaces showed it whole, and the stdout cap exists for echoed request payloads, so the JSON
record must not be middle-truncated."""
models: Final = _models()
policies: Final = _policies()
config: Final = _write_config(
tmp_path, gateway.upstream_url, models, {**CACHE_SETTINGS, "json_logs": True}, {"policies": policies}
)
log: Final = _boot_serve_and_stop(gateway, tmp_path, config, {"LITELLM_LOG": "INFO"}, 1, models)
_assert_every_line_is_json(log, 1, models)
listings: Final = tuple(message for message in _messages(log) if POLICY_LIST_HEADER in message)
assert len(listings) == 1, listings
assert len(listings[0]) > STDOUT_CAP, len(listings[0])
assert TRUNCATION_MARKER not in listings[0], listings[0]
for policy in policies:
assert policy in listings[0], (policy, listings[0])
def test_json_logs_model_list_keeps_a_model_name_the_secret_patterns_would_match(
gateway: Gateway, tmp_path: Path
) -> None:
"""The model list printed every model name as configured. The sk- key pattern matches inside a name such as
`risk-assessment-...`, so the JSON record must carry that name as the print did, not `riREDACTED`."""
models: Final = (f"risk-assessment-{uuid.uuid4().hex}",)
config: Final = _write_config(tmp_path, gateway.upstream_url, models, {**CACHE_SETTINGS, "json_logs": True})
log: Final = _boot_serve_and_stop(gateway, tmp_path, config, {"LITELLM_LOG": "INFO"}, 1, models)
_assert_every_line_is_json(log, 1, models)
assert f" {models[0]}" in _messages(log), _messages(log)
def test_json_logs_default_level_keeps_every_line_json(gateway: Gateway, tmp_path: Path) -> None:
"""Without LITELLM_LOG the proxy loggers sit at WARNING; the startup notices a plain-mode operator
always sees still arrive, as JSON, and nothing arrives as plaintext."""
models: Final = _models()
config: Final = _write_config(tmp_path, gateway.upstream_url, models, {**CACHE_SETTINGS, "json_logs": True})
log: Final = _boot_serve_and_stop(gateway, tmp_path, config, {}, 1, models)
plaintext: Final = _non_json_lines(log)
assert plaintext == (), "plaintext lines with JSON logs on:\n" + "\n".join(plaintext)
messages: Final = _messages(log)
assert sum(STARTED_WORKER in message for message in messages) == 1, messages
for fragment in (
BANNER_FRAGMENT,
FEEDBACK_BOX_LINE,
MODEL_LIST_HEADER,
CACHE_LINE,
CONNECTION_BUDGET_LINE,
*models,
):
assert any(fragment in message for message in messages), (fragment, messages)
def test_json_logs_warning_level_keeps_startup_notices(gateway: Gateway, tmp_path: Path) -> None:
"""`LITELLM_LOG=WARNING` drops the INFO logger lines; the startup notices a plain-mode operator sees at
every level still arrive, as JSON, and nothing arrives as plaintext."""
models: Final = _models()
config: Final = _write_config(tmp_path, gateway.upstream_url, models, {**CACHE_SETTINGS, "json_logs": True})
log: Final = _boot_serve_and_stop(gateway, tmp_path, config, {"LITELLM_LOG": "WARNING"}, 1, models)
plaintext: Final = _non_json_lines(log)
assert plaintext == (), "plaintext lines with JSON logs on:\n" + "\n".join(plaintext)
messages: Final = _messages(log)
assert not any(REDIS_AUTH_CACHE_LINE in message for message in messages), messages
for fragment in (
BANNER_FRAGMENT,
FEEDBACK_BOX_LINE,
MODEL_LIST_HEADER,
CACHE_LINE,
CONNECTION_BUDGET_LINE,
*models,
):
assert any(fragment in message for message in messages), (fragment, messages)
def test_json_logs_burst_on_two_workers_keeps_every_line_json(gateway: Gateway, tmp_path: Path) -> None:
"""Concurrent requests spread over both workers interleave their log writes; every line must still be
one complete JSON record."""
models: Final = _models()
config: Final = _write_config(tmp_path, gateway.upstream_url, models, {**CACHE_SETTINGS, "json_logs": True})
with owned_proxy_process(
gateway, tmp_path, {"LITELLM_LOG": "INFO"}, config=config, workers=2, database_setup=()
) as owned:
with ThreadPoolExecutor(max_workers=BURST_SIZE) as pool:
replies: Final = tuple(
pool.map(lambda index: owned.gateway.chat(models[index % 2], text=f"burst {index}"), range(BURST_SIZE))
)
assert tuple(reply["model"] for reply in replies) == tuple(models[index % 2] for index in range(BURST_SIZE))
log: Final = owned.log
text: Final = log.read_text()
plaintext: Final = _non_json_lines(text)
assert plaintext == (), "plaintext lines with JSON logs on:\n" + "\n".join(plaintext)
assert sum(STARTED_WORKER in message for message in _messages(text)) == 2, _messages(text)
def test_json_logs_config_makes_the_deprecated_flag_notice_one_json_record(gateway: Gateway, tmp_path: Path) -> None:
"""The notice waits until the config is read, so a json_logs set only there makes it a JSON record."""
models: Final = _models()
config: Final = _write_config(tmp_path, gateway.upstream_url, models, {"json_logs": True})
log: Final = _boot_serve_and_stop(gateway, tmp_path, config, {"LITELLM_LOG": "INFO"}, 1, models, (DEPRECATED_FLAG,))
plaintext: Final = _non_json_lines(log)
assert plaintext == (), "plaintext lines with JSON logs on:\n" + "\n".join(plaintext)
assert sum(DEPRECATED_FLAG_NOTICE in message for message in _messages(log)) == 1, _messages(log)
@pytest.mark.parametrize("config_text", (None, UNPARSEABLE_CONFIG), ids=("missing", "unparseable"))
def test_a_config_that_fails_to_load_still_prints_the_deprecated_flag_notice(
gateway: Gateway, tmp_path: Path, config_text: str | None
) -> None:
"""A boot with the deprecated flag and a config file that does not exist or does not parse exits
non-zero, and it still prints the notice once before the error, as it did before the notice waited
for the config."""
config: Final = tmp_path / f"unloadable_{uuid.uuid4().hex}.yaml"
if config_text is not None:
config.write_text(config_text)
log: Final = refused_boot_log(gateway, tmp_path, {}, config=config, extra_arguments=(DEPRECATED_FLAG,))
assert log.count(DEPRECATED_FLAG_NOTICE) == 1, log
def test_plain_logs_control_prints_startup_text(gateway: Gateway, tmp_path: Path) -> None:
models: Final = _models()
config: Final = _write_config(tmp_path, gateway.upstream_url, models, CACHE_SETTINGS)
log: Final = _boot_serve_and_stop(gateway, tmp_path, config, {"LITELLM_LOG": "INFO"}, 1, models)
plaintext: Final = _non_json_lines(log)
assert any(BANNER_FRAGMENT in line for line in plaintext), log
assert any(FEEDBACK_BOX_LINE in line for line in plaintext), log
assert any(CACHE_LINE in line for line in plaintext), log
assert any(MODEL_LIST_HEADER in line for line in plaintext), log
for model in models:
assert any(line.strip().endswith(model) or line.strip().endswith(f"{model}\x1b[0m") for line in plaintext), (
model,
log,
)
assert any(PLAIN_LOG_LINE.match(line) and REDIS_AUTH_CACHE_LINE in line for line in plaintext), log
assert any(line.startswith(f"LiteLLM Proxy: {CONNECTION_BUDGET_LINE}") for line in plaintext), log

5
tests/unit/proxy/db/conftest.py
View file

@ -100,6 +100,8 @@ if not earlier_calls and os.environ.get("FAKE_PRISMA_HANG_FIRST"):
grandchild = subprocess.Popen([sys.executable, "-c", "import time; time.sleep(600)"])
pathlib.Path(os.environ["FAKE_PRISMA_GRANDCHILD_PIDFILE"]).write_text(str(grandchild.pid))
time.sleep(600)
if os.environ.get("FAKE_PRISMA_STDOUT"):
sys.stdout.write(os.environ["FAKE_PRISMA_STDOUT"])
sys.exit(0)
"""
@ -111,6 +113,8 @@ class FakePrismaCli:
With FAKE_PRISMA_HANG_FIRST set it hangs on its first call from a process tree
of its own, the way the real CLI wraps Node around a Rust schema engine, so a
timeout that kills only the direct child leaves the rest of that tree running.
With FAKE_PRISMA_STDOUT set it writes that text to stdout, the way `migrate diff
--script` prints the SQL that would close a schema gap.
"""
calls_file: Path
@ -142,6 +146,7 @@ def fake_prisma_cli(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Generato
monkeypatch.setenv("FAKE_PRISMA_GRANDCHILD_PIDFILE", str(cli.grandchild_pidfile))
monkeypatch.setenv("LITELLM_PRISMA_COMMAND_TIMEOUT", "1")
monkeypatch.delenv("FAKE_PRISMA_HANG_FIRST", raising=False)
monkeypatch.delenv("FAKE_PRISMA_STDOUT", raising=False)
yield cli
if cli.grandchild_pidfile.exists():
try:

40
tests/unit/proxy/db/test_check_migration.py
View file

@ -1,6 +1,12 @@
import json
import logging
import sys
from typing import Final
import pytest
import litellm
from litellm._logging import turn_on_json, verbose_logger, verbose_proxy_logger
def test_check_migration_out_of_sync(mocker):
@ -69,3 +75,37 @@ def test_migrate_diff_without_the_prisma_runner_skips_instead_of_crashing_boot(m
monkeypatch.setitem(sys.modules, "litellm_proxy_extras.prisma_toolchain", None)
assert check_prisma_schema_diff_helper("postgresql://u:p@localhost:9/x") == (False, [])
def test_migrate_diff_announces_the_schema_changes_as_one_json_record(fake_prisma_cli, monkeypatch, capfd):
"""
The "Changes to DB Schema detected" notice and its SQL went to stdout as plaintext;
with json_logs on they are one INFO record, so a log pipeline keeps parsing. The
loggers sit at WARNING, as they do at boot without LITELLM_LOG, and the record
still goes out: the print it replaces had no level.
"""
from litellm.proxy.db.check_migration import check_prisma_schema_diff_helper
monkeypatch.setenv(
"FAKE_PRISMA_STDOUT", '-- AlterTable\nALTER TABLE "LiteLLM_UserTable" ADD COLUMN "nickname" TEXT;\n'
)
monkeypatch.setattr(litellm, "json_logs", True)
monkeypatch.setattr(sys, "excepthook", sys.excepthook) # turn_on_json swaps it; put it back at teardown
turn_on_json()
verbose_logger.setLevel(logging.WARNING)
verbose_proxy_logger.setLevel(logging.WARNING)
assert check_prisma_schema_diff_helper("postgresql://u:p@localhost:9/x") == (
True,
['ALTER TABLE "LiteLLM_UserTable" ADD COLUMN "nickname" TEXT;'],
)
out, err = capfd.readouterr()
assert err == ""
lines: Final = [raw for raw in out.splitlines() if raw.strip()]
assert len(lines) == 1, lines
record: Final = json.loads(lines[0])
assert record["level"] == "INFO"
assert record["message"] == (
'Changes to DB Schema detected\nRequired SQL commands:\nALTER TABLE "LiteLLM_UserTable" ADD COLUMN "nickname" TEXT;'
)

107
tests/unit/proxy/policy_engine/test_init_policies.py Normal file
View file

@ -0,0 +1,107 @@
"""The policy table printed at boot keeps its terminal form in plain mode and is JSON when json_logs is on."""
import json
import sys
from collections.abc import Mapping, Sequence
from typing import Final
import pytest
import litellm
from litellm._logging import turn_on_json
from litellm.proxy.policy_engine.init_policies import _print_policies_on_startup
POLICIES: Final[Mapping[str, Mapping[str, object]]] = {
"pii-policy": {
"inherit": "base",
"description": "Block PII",
"guardrails": {"add": ["pii-guard"], "remove": ["noop"]},
"condition": {"model": "gpt-4o"},
}
}
ATTACHMENTS: Final[Sequence[Mapping[str, object]]] = ({"policy": "pii-policy", "scope": "*", "teams": ["team-a"]},)
POLICY_ROWS: Final = (
" - pii-policy (inherits: base)",
" description: Block PII",
" guardrails.add: ['pii-guard']",
" guardrails.remove: ['noop']",
" condition.model: gpt-4o",
)
PLAIN_POLICY_TABLE: Final = (
"\x1b[92m\nLiteLLM Policy Engine: Loaded 1 policies\x1b[0m\n\n"
"\x1b[94m - pii-policy (inherits: base)\x1b[0m\n"
" description: Block PII\n"
" guardrails.add: ['pii-guard']\n"
" guardrails.remove: ['noop']\n"
" condition.model: gpt-4o\n"
)
NO_ATTACHMENTS_WARNING: Final = (
"Warning: No policy_attachments configured. Policies will not be applied to any requests."
)
@pytest.mark.parametrize(
("attachments", "expected"),
[
(
ATTACHMENTS,
PLAIN_POLICY_TABLE
+ "\n\x1b[93mPolicy Attachments: 1 attachment(s)\x1b[0m\n"
+ " - pii-policy -> scope=* (global), teams=['team-a']\n\n",
),
(None, PLAIN_POLICY_TABLE + f"\n\x1b[93m{NO_ATTACHMENTS_WARNING}\x1b[0m\n\n"),
],
ids=["with-attachments", "without-attachments"],
)
def test_policy_table_prints_unchanged_in_plain_mode(monkeypatch, capfd, attachments, expected):
"""Operators without JSON logs keep the colored policy table byte for byte."""
monkeypatch.setattr(litellm, "json_logs", False)
_print_policies_on_startup(POLICIES, attachments)
assert capfd.readouterr() == (expected, "")
def test_policy_table_is_one_json_record_in_json_mode(monkeypatch, capfd):
"""With JSON logs on the whole table, attachments included, is one INFO record without color codes."""
monkeypatch.setattr(litellm, "json_logs", True)
monkeypatch.setattr(sys, "excepthook", sys.excepthook) # turn_on_json swaps it; put it back at teardown
turn_on_json()
_print_policies_on_startup(POLICIES, ATTACHMENTS)
out, err = capfd.readouterr()
assert err == ""
lines: Final = [raw for raw in out.splitlines() if raw.strip()]
assert len(lines) == 1, lines
record: Final = json.loads(lines[0])
assert record["level"] == "INFO"
assert record["message"] == "\n".join(
(
"LiteLLM Policy Engine: Loaded 1 policies",
"",
*POLICY_ROWS,
"",
"Policy Attachments: 1 attachment(s)",
" - pii-policy -> scope=* (global), teams=['team-a']",
)
)
def test_missing_attachments_warning_is_a_second_json_record(monkeypatch, capfd):
"""Without attachments the table and the yellow warning are two INFO records on stdout, where the two
prints put them."""
monkeypatch.setattr(litellm, "json_logs", True)
monkeypatch.setattr(sys, "excepthook", sys.excepthook)
turn_on_json()
_print_policies_on_startup(POLICIES, None)
out, err = capfd.readouterr()
assert err == ""
records: Final = [json.loads(raw) for raw in out.splitlines() if raw.strip()]
assert [record["message"] for record in records] == [
"\n".join(("LiteLLM Policy Engine: Loaded 1 policies", "", *POLICY_ROWS)),
NO_ATTACHMENTS_WARNING,
]
assert {record["level"] for record in records} == {"INFO"}

156
tests/unit/proxy/test_proxy_cli.py
View file

@ -1,4 +1,5 @@
import inspect
import json
import logging
import os
from contextlib import nullcontext
@ -23,6 +24,8 @@ import yaml
from uvicorn.config import LOOP_FACTORIES
from uvicorn.importer import import_from_string
import litellm
from litellm._logging import turn_on_json
from litellm.constants import GRACEFUL_SHUTDOWN_TEARDOWN_HEADROOM_SECONDS
from litellm.proxy import proxy_cli
from litellm.proxy.proxy_cli import ProxyInitializationHelpers, run_server
@ -2875,6 +2878,22 @@ class TestRunServerDbSetup:
mock_setup_database.assert_called_once_with(use_migrate=True, use_v2_resolver=True)
assert ("--enforce_prisma_migration_check is deprecated and has no effect" in capsys.readouterr().out) is warned
@pytest.mark.parametrize("argument", ("--health", "--test"), ids=("health", "test"))
def test_the_retired_enforce_prisma_migration_check_opt_in_warns_on_health_and_test(self, argument, capsys):
"""`--health` and `--test` return before the config is read, where a server start emits the
deprecation line; they still print it, as they did when the notice came before those returns."""
from litellm.proxy.proxy_cli import ProxyInitializationHelpers, run_server
with (
patch.object(ProxyInitializationHelpers, "run_health_check") as mock_health_check,
patch.object(ProxyInitializationHelpers, "run_test_chat_completion") as mock_test_chat,
):
run_server.main([argument, "--enforce_prisma_migration_check"], standalone_mode=False)
expected_calls = (1, 0) if argument == "--health" else (0, 1)
assert (mock_health_check.call_count, mock_test_chat.call_count) == expected_calls
assert "--enforce_prisma_migration_check is deprecated and has no effect" in capsys.readouterr().out
@patch("subprocess.run")
@patch("atexit.register")
@patch("litellm.proxy.db.prisma_client.PrismaManager.setup_database", return_value=True)
@ -3935,3 +3954,140 @@ class TestValidateConfigFlag:
assert result.exit_code == 0, f"exit_code={result.exit_code}, output={result.output}"
assert "config OK" in result.output
mock_popen.assert_not_called()
class TestJsonLogsConfigReachesSpawnedWorkers:
"""`litellm_settings.json_logs: true` turns JSON on inside the CLI process, but the workers that
uvicorn or granian spawn are fresh interpreters that pick plain or JSON from the JSON_LOGS
environment variable at import. The CLI has to export it, or every worker logs plain text until
its own load_config reaches the key."""
@staticmethod
def _json_logs_env_after_boot(config_path: Path, litellm_settings: dict[str, object]) -> str | None:
config_path.write_text(yaml.dump({"model_list": [], "litellm_settings": litellm_settings}))
mock_proxy_config: Final = MagicMock()
mock_proxy_config.return_value.get_config = AsyncMock(return_value=yaml.safe_load(config_path.read_text()))
mock_proxy_module: Final = MagicMock(
app=MagicMock(),
ProxyConfig=mock_proxy_config,
KeyManagementSettings=MagicMock(),
save_worker_config=MagicMock(),
)
clean_env: Final = {k: v for k, v in os.environ.items() if k not in ("DATABASE_URL", "DIRECT_URL", "JSON_LOGS")}
with (
patch.dict(os.environ, clean_env, clear=True),
patch.dict(
"sys.modules",
{"proxy_server": mock_proxy_module, "litellm.proxy.proxy_server": mock_proxy_module},
),
patch("atexit.register"),
patch("litellm.turn_on_json"), # test-quality-ok: it swaps every logger handler in this pytest worker
):
run_server.main(["--config", str(config_path), "--local", "--skip_server_startup"], standalone_mode=False)
return os.environ.get("JSON_LOGS")
def test_json_logs_true_exports_json_logs_to_spawned_workers(self, tmp_path: Path, monkeypatch) -> None:
monkeypatch.setattr(litellm, "json_logs", False) # run_server flips it on; undone at teardown
assert self._json_logs_env_after_boot(tmp_path / "config.yaml", {"json_logs": True}) == "true"
def test_plain_logs_config_leaves_json_logs_unset(self, tmp_path: Path, monkeypatch) -> None:
monkeypatch.setattr(litellm, "json_logs", False)
assert self._json_logs_env_after_boot(tmp_path / "config.yaml", {}) is None
class TestEnforcePrismaMigrationCheckNoticeAroundTheConfigRead:
"""With a config file, the `--enforce_prisma_migration_check` deprecation notice waits until the config
is read, so a `json_logs: true` set only there makes it a JSON record. It is printed once either way,
and a boot whose config read fails still prints it before the error, as it did before the wait."""
NOTICE: Final = "--enforce_prisma_migration_check is deprecated and has no effect"
@staticmethod
def _boot(config_path: Path, get_config: AsyncMock) -> None:
mock_proxy_config: Final = MagicMock()
mock_proxy_config.return_value.get_config = get_config
mock_proxy_module: Final = MagicMock(
app=MagicMock(),
ProxyConfig=mock_proxy_config,
KeyManagementSettings=MagicMock(),
save_worker_config=MagicMock(),
)
clean_env: Final = {k: v for k, v in os.environ.items() if k not in ("DATABASE_URL", "DIRECT_URL", "JSON_LOGS")}
with (
patch.dict(os.environ, clean_env, clear=True),
patch.dict(
"sys.modules",
{"proxy_server": mock_proxy_module, "litellm.proxy.proxy_server": mock_proxy_module},
),
patch("atexit.register"),
patch("litellm.turn_on_json"), # test-quality-ok: it swaps every logger handler in this pytest worker
):
run_server.main(
["--config", str(config_path), "--local", "--skip_server_startup", "--enforce_prisma_migration_check"],
standalone_mode=False,
)
def test_a_config_read_that_fails_still_prints_the_notice(self, tmp_path: Path, monkeypatch, capsys) -> None:
monkeypatch.setattr(litellm, "json_logs", False)
missing: Final = tmp_path / "missing.yaml"
with pytest.raises(FileNotFoundError):
self._boot(missing, AsyncMock(side_effect=FileNotFoundError(f"Config file not found: {missing}")))
assert capsys.readouterr().out.count(self.NOTICE) == 1
def test_a_plain_config_prints_the_notice_once(self, tmp_path: Path, monkeypatch, capsys) -> None:
monkeypatch.setattr(litellm, "json_logs", False)
self._boot(tmp_path / "config.yaml", AsyncMock(return_value={"model_list": [], "litellm_settings": {}}))
assert capsys.readouterr().out.count(self.NOTICE) == 1
def test_a_config_only_json_logs_makes_the_notice_one_json_record(
self, tmp_path: Path, monkeypatch, capsys
) -> None:
monkeypatch.setattr(litellm, "json_logs", False) # run_server flips it on; undone at teardown
config: Final = {"model_list": [], "litellm_settings": {"json_logs": True}}
self._boot(tmp_path / "config.yaml", AsyncMock(return_value=config))
notice_lines: Final = [line for line in capsys.readouterr().out.splitlines() if self.NOTICE in line]
assert len(notice_lines) == 1, notice_lines
assert self.NOTICE in json.loads(notice_lines[0])["message"]
def test_granian_flag_notices_are_json_records_in_json_mode(monkeypatch, capfd):
"""The yellow notices for the flags Granian ignores (`--max_requests_before_restart`, `--ciphers`) are
printed while the launcher starts; with json_logs on they are INFO records on stdout like every other
line, not plaintext between the records."""
pytest.importorskip("granian")
monkeypatch.setattr(litellm, "json_logs", True)
monkeypatch.setattr(sys, "excepthook", sys.excepthook) # turn_on_json swaps it; put it back at teardown
turn_on_json()
with (
patch("granian.Granian") as mock_granian_cls,
patch("granian.constants.Interfaces", SimpleNamespace(ASGI="asgi")),
):
ProxyInitializationHelpers._init_granian_server(
host="0.0.0.0",
port=4000,
num_workers=2,
ssl_certfile_path=None,
ssl_keyfile_path=None,
max_requests_before_restart=100,
ciphers="ECDHE",
granian_runtime_threads=None,
)
mock_granian_cls.return_value.serve.assert_called_once()
out, err = capfd.readouterr()
assert err == ""
records: Final = [json.loads(raw) for raw in out.splitlines() if raw.strip()]
assert [record["message"] for record in records] == [
"LiteLLM Proxy: Starting server on 0.0.0.0:4000 using Granian",
"LiteLLM: --max_requests_before_restart is not supported by Granian "
"(Granian uses workers_lifetime in seconds, not a per-request limit).",
"LiteLLM: --ciphers is not applied when using --run_granian.",
]
assert {record["level"] for record in records} == {"INFO"}

55
tests/unit/proxy/test_proxy_server_endpoints_and_startup.py
View file

@ -7,6 +7,7 @@ import os
import re
import socket
import subprocess
import sys
import time
import types
from datetime import datetime, timedelta, timezone
@ -28,6 +29,7 @@ from fastapi.testclient import TestClient
import litellm
import litellm.proxy.proxy_server as proxy_server_module
from litellm._internal_context import current_service_target
from litellm._logging import turn_on_json, verbose_proxy_logger
from litellm.caching.caching import RedisCache
from litellm.caching.redis_cluster_cache import RedisClusterCache
from litellm.litellm_core_utils.get_model_cost_map import ModelCostMapReloaded
@ -46,6 +48,7 @@ from litellm.proxy.auth.user_api_key_auth import user_api_key_auth
from litellm.proxy.hooks.parallel_request_limiter_v3 import RequestRateLimiterStash
from litellm.proxy.proxy_server import app, aws_error_type_exception_handler, initialize, openai_exception_handler
from litellm.utils import _invalidate_model_cost_lowercase_map
from litellm.proxy.common_utils.banner import LITELLM_BANNER
from tests._master_key import MASTER_KEY
PUBLICLY_KNOWN_KEY: Final = "sk-" + "1234"
@ -14266,7 +14269,6 @@ async def test_moderations_response_carries_litellm_call_id_header():
async def test_moderations_failure_log_carries_the_callers_litellm_call_id(caplog):
"""LIT-7836: the /v1/moderations error line must carry the litellm_call_id the
client sent, rendered in the message and as a structured log record field."""
from litellm._logging import verbose_proxy_logger
from litellm.proxy._types import ProxyException
call_id = "moderations-call-7836"
@ -15588,3 +15590,54 @@ async def test_spend_capture_rate_check_job_clears_the_gauge_once_the_setting_is
call(api_provider="openai", capture_rate=None),
]
def test_feedback_box_is_one_json_record_in_json_mode(monkeypatch, capfd):
"""The feedback box is a dozen terminal lines; with JSON logs on it has to land as one parseable
record, not a plaintext line per row."""
monkeypatch.setattr(litellm, "json_logs", True)
monkeypatch.setattr(sys, "excepthook", sys.excepthook) # turn_on_json swaps it; put it back at teardown
turn_on_json()
proxy_server_module.generate_feedback_box()
out, err = capfd.readouterr()
assert err == ""
lines: Final = [raw for raw in out.splitlines() if raw.strip()]
assert len(lines) == 1, lines
record: Final = json.loads(lines[0])
assert "Give Feedback / Get Help: https://github.com/BerriAI/litellm/issues/new" in record["message"]
assert "Thank you for using LiteLLM! - Krrish & Ishaan" in record["message"]
assert "\x1b[" not in record["message"]
@pytest.mark.asyncio
async def test_initialize_prints_the_banner_between_blank_lines_in_plain_mode(monkeypatch, capfd):
"""Operators without JSON logs see the banner exactly as before, through show_banner()'s flushing
click.echo."""
monkeypatch.setattr(litellm, "json_logs", False)
monkeypatch.setenv("LITELLM_DONT_SHOW_FEEDBACK_BOX", "true")
await proxy_server_module.initialize()
out, err = capfd.readouterr()
assert err == ""
assert out.startswith(f"\n{LITELLM_BANNER}\n\n"), out
@pytest.mark.asyncio
async def test_initialize_prints_the_banner_as_one_json_record_in_json_mode(monkeypatch, capfd):
"""With JSON logs on, the six banner rows are one INFO record with their indentation kept, so the
first line a worker writes parses like every other."""
monkeypatch.setattr(litellm, "json_logs", True)
monkeypatch.setattr(sys, "excepthook", sys.excepthook) # turn_on_json swaps it; put it back at teardown
monkeypatch.setenv("LITELLM_DONT_SHOW_FEEDBACK_BOX", "true")
turn_on_json()
await proxy_server_module.initialize()
out, err = capfd.readouterr()
assert err == ""
records: Final = [json.loads(raw) for raw in out.splitlines() if raw.strip()]
banner_records: Final = [record for record in records if record["message"] == LITELLM_BANNER]
assert len(banner_records) == 1, records
assert banner_records[0]["level"] == "INFO"

207
tests/unit/test_logging.py
View file

@ -34,6 +34,7 @@ from litellm._logging import (
_parse_json_logs_env,
_plain_log_format,
_stdout_truncation_marker,
print_startup_message,
turn_on_json,
format_base64_size,
session_id_var,
@ -1824,3 +1825,209 @@ def test_print_verbose_formats_args_only_when_set_verbose_is_on(monkeypatch, cap
logging_module.print_verbose("key %s value %s", "k1", _StrCountingValue())
assert _StrCountingValue.str_calls == 1
assert capsys.readouterr().out == "key k1 value cached-vector\n"
def test_print_startup_message_prints_plain_mode_text_unchanged(monkeypatch, capfd):
"""Operators without JSON logs keep the colored, spaced terminal output exactly as before."""
monkeypatch.setattr(litellm, "json_logs", False)
print_startup_message("\n\033[1;32mLiteLLM: Proxy initialized with Config, Set models:\033[0m\n")
out, err = capfd.readouterr()
assert out == "\n\033[1;32mLiteLLM: Proxy initialized with Config, Set models:\033[0m\n\n"
assert err == ""
def test_print_startup_message_is_one_json_record_in_json_mode(monkeypatch, capfd):
"""With JSON logs on, a multi-line colored startup message is one INFO record with the colors and the
surrounding blank lines removed, so a log pipeline parses it like every other line. The logger sits
at WARNING, as it does in the CLI process, and the record still goes out: a print has no level."""
monkeypatch.setattr(litellm, "json_logs", True)
monkeypatch.setattr(sys, "excepthook", sys.excepthook) # turn_on_json swaps it; put it back at teardown
turn_on_json()
verbose_proxy_logger.setLevel(logging.WARNING)
print_startup_message(
"\n\033[1;32mLiteLLM: Proxy initialized with Config, Set models:\033[0m\n\033[32m gpt-4o\033[0m\n\n"
)
out, err = capfd.readouterr()
assert err == ""
lines: Final = [raw for raw in out.splitlines() if raw.strip()]
assert len(lines) == 1, lines
record: Final = json.loads(lines[0])
assert record["message"] == "LiteLLM: Proxy initialized with Config, Set models:\n gpt-4o"
assert record["level"] == "INFO"
assert record["component"] == "LiteLLM Proxy"
# the record names the call site, not the helper, so operators can find the message
assert str(record["logger"]).startswith("test_logging.py:"), record["logger"]
def test_print_startup_message_yellow_notice_is_an_info_record_on_stdout(monkeypatch, capfd):
"""An operator notice printed in yellow (an ignored flag, a deprecation) stays on stdout in JSON mode,
where the print put it, as one INFO record with the color codes removed."""
monkeypatch.setattr(litellm, "json_logs", True)
monkeypatch.setattr(sys, "excepthook", sys.excepthook)
turn_on_json()
verbose_proxy_logger.setLevel(logging.WARNING)
print_startup_message("\033[1;33mLiteLLM: --ciphers is not applied when using --run_granian.\033[0m\n")
out, err = capfd.readouterr()
assert err == ""
lines: Final = [raw for raw in out.splitlines() if raw.strip()]
assert len(lines) == 1, lines
record: Final = json.loads(lines[0])
assert record["message"] == "LiteLLM: --ciphers is not applied when using --run_granian."
assert record["level"] == "INFO"
def test_print_startup_message_shows_above_the_json_handler_level(monkeypatch, capfd):
"""`LITELLM_LOG=WARNING` sets the JSON handler to WARNING, which drops the INFO logger lines; the
startup notices still go out, as the prints they replace did at every level."""
monkeypatch.setattr(litellm, "json_logs", True)
monkeypatch.setattr(sys, "excepthook", sys.excepthook)
turn_on_json()
verbose_proxy_logger.setLevel(logging.INFO)
for json_handler in verbose_proxy_logger.handlers:
monkeypatch.setattr(json_handler, "level", logging.WARNING)
print_startup_message("\033[94m\nSetting Cache on Proxy\033[0m")
verbose_proxy_logger.info("enable_redis_auth_cache is not set")
out, _ = capfd.readouterr()
lines: Final = [raw for raw in out.splitlines() if raw.strip()]
assert [json.loads(raw)["message"] for raw in lines] == ["Setting Cache on Proxy"]
def test_print_startup_message_emits_json_after_an_operator_log_config_removed_the_handlers(
request, monkeypatch, capfd
):
"""uvicorn applies an operator `--log_config` after proxy_cli's turn_on_json() and before initialize()
re-installs the JSON handler through load_config(); a config that names "LiteLLM Proxy" leaves the
logger without handlers in that window, where the banner and the feedback box are emitted, and they
still go out as one JSON record each, on stdout."""
import logging.config
monkeypatch.setattr(litellm, "json_logs", True)
monkeypatch.setattr(sys, "excepthook", sys.excepthook)
turn_on_json()
request.addfinalizer(turn_on_json) # put the JSON handler back for the tests that follow
logging.config.dictConfig(
{"version": 1, "disable_existing_loggers": False, "loggers": {"LiteLLM Proxy": {"level": "INFO"}}}
)
assert verbose_proxy_logger.handlers == []
print_startup_message("\n\033[1;37m# Give Feedback / Get Help #\033[0m\n")
out, err = capfd.readouterr()
assert err == ""
lines: Final = [raw for raw in out.splitlines() if raw.strip()]
assert len(lines) == 1, lines
record: Final = json.loads(lines[0])
assert record["message"] == "# Give Feedback / Get Help #"
assert record["level"] == "INFO"
assert record["component"] == "LiteLLM Proxy"
OPERATOR_HANDLERS_ON_THE_PROXY_LOGGER: Final = {
"silenced": {"class": "logging.NullHandler"},
"stderr_warning": {
"class": "logging.StreamHandler",
"stream": "ext://sys.stderr",
"level": "WARNING",
"formatter": "plain",
},
}
@pytest.mark.parametrize("handler", sorted(OPERATOR_HANDLERS_ON_THE_PROXY_LOGGER))
def test_print_startup_message_stays_on_stdout_under_an_operator_handler_on_the_proxy_logger(
request, monkeypatch, capfd, handler: str
):
"""An operator `--log_config` that gives "LiteLLM Proxy" a handler of its own, a NullHandler that silences
it or a plain WARNING-level handler on stderr, is in force when the banner and the feedback box are
emitted; the print they replace wrote to stdout whatever the logging config said, so the record still
goes out as one JSON record on stdout and the operator's handler gets nothing."""
import logging.config
monkeypatch.setattr(litellm, "json_logs", True)
monkeypatch.setattr(sys, "excepthook", sys.excepthook)
turn_on_json()
request.addfinalizer(turn_on_json) # put the JSON handler back for the tests that follow
logging.config.dictConfig(
{
"version": 1,
"disable_existing_loggers": False,
"formatters": {"plain": {"format": "%(levelname)s %(message)s"}},
"handlers": {"proxy": OPERATOR_HANDLERS_ON_THE_PROXY_LOGGER[handler]},
"loggers": {"LiteLLM Proxy": {"level": "INFO", "handlers": ["proxy"], "propagate": False}},
}
)
assert len(verbose_proxy_logger.handlers) == 1 and not isinstance(
verbose_proxy_logger.handlers[0], LevelRoutingStreamHandler
)
print_startup_message("\n\033[1;37m# Give Feedback / Get Help #\033[0m\n")
out, err = capfd.readouterr()
assert err == "", err
lines: Final = [raw for raw in out.splitlines() if raw.strip()]
assert len(lines) == 1, lines
record: Final = json.loads(lines[0])
assert record["message"] == "# Give Feedback / Get Help #"
assert record["level"] == "INFO"
assert record["component"] == "LiteLLM Proxy"
def test_print_startup_message_keeps_a_listing_longer_than_the_stdout_cap_whole(monkeypatch, capfd):
"""The required migration SQL and the policy list run past MAX_STRING_LENGTH_STDOUT_LOG on a drifted
deployment. The prints they replace showed them whole, and that cap exists for echoed request payloads,
so the JSON record carries the whole listing too."""
monkeypatch.setenv("MAX_STRING_LENGTH_STDOUT_LOG", "500")
monkeypatch.setattr(litellm, "json_logs", True)
monkeypatch.setattr(sys, "excepthook", sys.excepthook) # turn_on_json swaps it; put it back at teardown
turn_on_json()
commands: Final = tuple(f'ALTER TABLE "LiteLLM_Table{n}" ADD COLUMN "column_{n}" TEXT;' for n in range(40))
listing: Final = "\n".join(("Changes to DB Schema detected", "Required SQL commands:", *commands))
assert len(listing) > 500
print_startup_message(listing)
out, err = capfd.readouterr()
assert err == ""
lines: Final = [raw for raw in out.splitlines() if raw.strip()]
assert len(lines) == 1, lines
record: Final = json.loads(lines[0])
assert LITELLM_TRUNCATED_PAYLOAD_FIELD not in record["message"]
assert record["message"] == listing
def test_print_startup_message_keeps_a_model_name_the_secret_patterns_would_match(monkeypatch, capfd):
"""The model list printed `risk-assessment-gpt4o` as configured; the sk- key pattern matches inside
that name, so the JSON record has to carry the name as the print did, not `riREDACTED`."""
monkeypatch.setattr(litellm, "json_logs", True)
monkeypatch.setattr(sys, "excepthook", sys.excepthook) # turn_on_json swaps it; put it back at teardown
turn_on_json()
assert secret_redaction.redact_string("risk-assessment-gpt4o") != "risk-assessment-gpt4o"
print_startup_message("\033[32m risk-assessment-gpt4o\033[0m")
out, err = capfd.readouterr()
assert err == ""
lines: Final = [raw for raw in out.splitlines() if raw.strip()]
assert len(lines) == 1, lines
assert json.loads(lines[0])["message"] == " risk-assessment-gpt4o"
def test_print_startup_message_blank_spacer_emits_no_json_record(monkeypatch, capfd):
"""A spacer that only pads the terminal has nothing to say to a log pipeline."""
monkeypatch.setattr(litellm, "json_logs", True)
monkeypatch.setattr(sys, "excepthook", sys.excepthook)
turn_on_json()
verbose_proxy_logger.setLevel(logging.INFO)
print_startup_message("")
print_startup_message("\n\033[0m\n")
assert capfd.readouterr() == ("", "")