From 81f25633381039b099ebc03df19a6f0c56e40722 Mon Sep 17 00:00:00 2001 From: Davide Pugliese Date: Thu, 7 Aug 2025 14:01:14 +0200 Subject: [PATCH] Enhance logging for containers --- .dockerignore | 1 + .gitignore | 3 +- litellm/_logging.py | 51 +- tests/test_litellm/conftest.py | 74 +++ tests/test_litellm/test_logging_behavior.py | 638 ++++++++++++++++++++ 5 files changed, 760 insertions(+), 7 deletions(-) create mode 100644 tests/test_litellm/test_logging_behavior.py diff --git a/.dockerignore b/.dockerignore index 89c3c34bd71..766b7a1db67 100644 --- a/.dockerignore +++ b/.dockerignore @@ -10,3 +10,4 @@ tests *.tgz log.txt docker/Dockerfile.* +*.whl diff --git a/.gitignore b/.gitignore index f8d028ff47b..9613ef77d90 100644 --- a/.gitignore +++ b/.gitignore @@ -93,4 +93,5 @@ test.py litellm_config.yaml .cursor -.vscode/launch.json \ No newline at end of file +.vscode/launch.json +*.whl \ No newline at end of file diff --git a/litellm/_logging.py b/litellm/_logging.py index 8c23994f92a..9a8c8251ad4 100644 --- a/litellm/_logging.py +++ b/litellm/_logging.py @@ -4,21 +4,41 @@ import os import sys from datetime import datetime from logging import Formatter - set_verbose = False +def __strtobool(val: str) -> bool: + """Convert a string representation of truth to true (1) or false (0). + + True values are 'y', 'yes', 't', 'true', 'on', and '1'; false values + are 'n', 'no', 'f', 'false', 'off', and '0'. Raises ValueError if + 'val' is anything else. + """ + val = val.lower() + if val in ('y', 'yes', 't', 'true', 'on', '1'): + return True + elif val in ('n', 'no', 'f', 'false', 'off', '0'): + return False + else: + raise ValueError(f"invalid truth value {val!r}") + if set_verbose is True: logging.warning( "`litellm.set_verbose` is deprecated. Please set `os.environ['LITELLM_LOG'] = 'DEBUG'` for debug logs." ) -json_logs = bool(os.getenv("JSON_LOGS", False)) + +json_logs = __strtobool(os.getenv("JSON_LOGS", "False")) # Create a handler for the logger (you may need to adapt this based on your needs) log_level = os.getenv("LITELLM_LOG", "DEBUG") numeric_level: str = getattr(logging, log_level.upper()) handler = logging.StreamHandler() handler.setLevel(numeric_level) +log_file = os.getenv("LITELLM_LOG_FILE", "") +file_handler = None +if log_file: + file_handler = logging.FileHandler(log_file) + file_handler.setLevel(numeric_level) class JsonFormatter(Formatter): def __init__(self): super(JsonFormatter, self).__init__() @@ -40,6 +60,7 @@ class JsonFormatter(Formatter): return json.dumps(json_record) +json_formatter = JsonFormatter() # Function to set up exception handlers for JSON logging def _setup_json_exception_handlers(formatter): @@ -89,8 +110,10 @@ def _setup_json_exception_handlers(formatter): # Create a formatter and set it for the handler if json_logs: - handler.setFormatter(JsonFormatter()) - _setup_json_exception_handlers(JsonFormatter()) + handler.setFormatter(json_formatter) + if file_handler: + file_handler.setFormatter(json_formatter) + _setup_json_exception_handlers(json_formatter) else: formatter = logging.Formatter( "\033[92m%(asctime)s - %(name)s:%(levelname)s\033[0m: %(filename)s:%(lineno)s - %(message)s", @@ -98,11 +121,18 @@ else: ) handler.setFormatter(formatter) + if file_handler: + file_handler.setFormatter(formatter) verbose_proxy_logger = logging.getLogger("LiteLLM Proxy") verbose_router_logger = logging.getLogger("LiteLLM Router") verbose_logger = logging.getLogger("LiteLLM") +# Set logger levels +verbose_proxy_logger.setLevel(numeric_level) +verbose_router_logger.setLevel(numeric_level) +verbose_logger.setLevel(numeric_level) + # Add the handler to the logger verbose_router_logger.addHandler(handler) verbose_proxy_logger.addHandler(handler) @@ -123,6 +153,13 @@ def _suppress_loggers(): # Call the suppression function _suppress_loggers() +if file_handler: + verbose_router_logger.addHandler(file_handler) + verbose_proxy_logger.addHandler(file_handler) + verbose_logger.addHandler(file_handler) + + + ALL_LOGGERS = [ logging.getLogger(), verbose_logger, @@ -151,10 +188,10 @@ def _turn_on_json(): - Adds a JSON formatter to all loggers """ handler = logging.StreamHandler() - handler.setFormatter(JsonFormatter()) + handler.setFormatter(json_formatter) _initialize_loggers_with_handler(handler) # Set up exception handlers - _setup_json_exception_handlers(JsonFormatter()) + _setup_json_exception_handlers(json_formatter) def _turn_on_debug(): @@ -190,3 +227,5 @@ def _is_debugging_on() -> bool: if verbose_logger.isEnabledFor(logging.DEBUG) or set_verbose is True: return True return False + + diff --git a/tests/test_litellm/conftest.py b/tests/test_litellm/conftest.py index a88148f9d11..db4224ca6e4 100644 --- a/tests/test_litellm/conftest.py +++ b/tests/test_litellm/conftest.py @@ -3,15 +3,88 @@ import importlib import os import sys +import tempfile +import random +import string import pytest +# Set up a temporary log directory and file BEFORE importing litellm +temp_dir = tempfile.mkdtemp(prefix="litellm_test_") +test_log_file = os.path.join(temp_dir, "test_litellm.log") + +# Store original log file for cleanup +orig_log_file = os.getenv("LITELLM_LOG_FILE") + +# Set environment variables to use temporary files BEFORE importing litellm +os.environ["LITELLM_LOG_FILE"] = test_log_file + +# Import litellm after setting up the environment sys.path.insert( 0, os.path.abspath("../..") ) # Adds the parent directory to the system path import litellm + + +@pytest.fixture(scope="function") +def temp_log_file(): + """ + Creates a temporary log file in /tmp/litellm.log for testing. + Returns the path to the temporary log file and cleans it up after the test. + """ + # Generate a random number for the log file + random_number = ''.join(random.choices(string.digits, k=8)) + log_file_path = f"/tmp/litellm{random_number}.log" + + # Set the environment variable for litellm to use this temporary log file + original_log_file = os.environ.get("LITELLM_LOG_FILE") + os.environ["LITELLM_LOG_FILE"] = log_file_path + + yield log_file_path + + # Cleanup: Restore original environment variable and remove the temporary file + if original_log_file is not None: + os.environ["LITELLM_LOG_FILE"] = original_log_file + else: + os.environ.pop("LITELLM_LOG_FILE", None) + + # Remove the temporary log file if it exists + if os.path.exists(log_file_path): + try: + os.remove(log_file_path) + except OSError: + pass # Ignore errors if file can't be removed + + +@pytest.fixture(scope="session", autouse=True) +def cleanup_temp_log_dir(): + """ + Cleans up the temporary log directory created at module import time. + This runs once per test session after all tests are complete. + """ + yield + + if orig_log_file is not None: + os.environ["LITELLM_LOG_FILE"] = orig_log_file + else: + os.environ.pop("LITELLM_LOG_FILE", None) + + # Cleanup: Remove the temporary directory created at module import time + if os.path.exists(temp_dir): + try: + # Remove the test log file first + if os.path.exists(test_log_file): + os.remove(test_log_file) + + # Remove the temporary directory + import shutil + shutil.rmtree(temp_dir, ignore_errors=True) + except OSError: + pass # Ignore errors if cleanup fails + + @pytest.fixture(scope="function", autouse=True) def setup_and_teardown(): """ @@ -63,3 +136,4 @@ def pytest_collection_modifyitems(config, items): # Reorder the items list items[:] = custom_logger_tests + other_tests + diff --git a/tests/test_litellm/test_logging_behavior.py b/tests/test_litellm/test_logging_behavior.py new file mode 100644 index 00000000000..24f92838acc --- /dev/null +++ b/tests/test_litellm/test_logging_behavior.py @@ -0,0 +1,638 @@ +import os +import tempfile +import re +import json +from pathlib import Path +from datetime import datetime + +import pytest + +# Import the loggers from litellm._logging +from litellm._logging import verbose_logger, verbose_proxy_logger, verbose_router_logger + + +class TestLoggingBehavior: + """Test suite to verify logging behavior for all LiteLLM loggers.""" + + def read_log_file_contents(self, log_file_path): + """Helper method to read and return contents of log file.""" + if not os.path.exists(log_file_path): + return "" + + with open(log_file_path, 'r') as f: + return f.read() + + @pytest.fixture(autouse=True) + def setup_log_file(self, temp_log_file): + """Use the temp_log_file fixture to ensure proper isolation.""" + self.temp_log_path = temp_log_file + + # Set environment variable before importing/reloading + original_log_file = os.environ.get("LITELLM_LOG_FILE") + os.environ["LITELLM_LOG_FILE"] = temp_log_file + + # Force reload of the logging module to pick up new environment variable + import importlib + import litellm._logging + importlib.reload(litellm._logging) + + yield + + # Cleanup: Restore original environment variable + if original_log_file is not None: + os.environ["LITELLM_LOG_FILE"] = original_log_file + else: + os.environ.pop("LITELLM_LOG_FILE", None) + + # Reload again to restore original state + importlib.reload(litellm._logging) + + def test_verbose_logger_info_level(self): + """Test that verbose_logger writes to file with INFO level.""" + test_message = "INFO level test message from verbose_logger" + + # Log at INFO level + verbose_logger.info(test_message) + + # Force flush all handlers to ensure they write to disk + for handler in verbose_logger.handlers: + if hasattr(handler, 'flush'): + handler.flush() + + # Read log file contents + log_file_path = os.environ.get("LITELLM_LOG_FILE") + assert log_file_path is not None, "LITELLM_LOG_FILE environment variable should be set" + + log_contents = self.read_log_file_contents(log_file_path) + assert test_message in log_contents, f"Message '{test_message}' should be found in log file" + + def test_verbose_logger_debug_level(self): + """Test that verbose_logger writes to file with DEBUG level.""" + test_message = "DEBUG level test message from verbose_logger" + + # Log at DEBUG level + verbose_logger.debug(test_message) + + # Read log file contents + log_file_path = os.environ.get("LITELLM_LOG_FILE") + assert log_file_path is not None, "LITELLM_LOG_FILE environment variable should be set" + + log_contents = self.read_log_file_contents(log_file_path) + assert test_message in log_contents, f"Message '{test_message}' should be found in log file" + + def test_verbose_proxy_logger_info_level(self): + """Test that verbose_proxy_logger writes to file with INFO level.""" + test_message = "INFO level test message from verbose_proxy_logger" + + # Log at INFO level + verbose_proxy_logger.info(test_message) + + # Read log file contents + log_file_path = os.environ.get("LITELLM_LOG_FILE") + assert log_file_path is not None, "LITELLM_LOG_FILE environment variable should be set" + + log_contents = self.read_log_file_contents(log_file_path) + assert test_message in log_contents, f"Message '{test_message}' should be found in log file" + + def test_verbose_proxy_logger_debug_level(self): + """Test that verbose_proxy_logger writes to file with DEBUG level.""" + test_message = "DEBUG level test message from verbose_proxy_logger" + + # Log at DEBUG level + verbose_proxy_logger.debug(test_message) + + # Read log file contents + log_file_path = os.environ.get("LITELLM_LOG_FILE") + assert log_file_path is not None, "LITELLM_LOG_FILE environment variable should be set" + + log_contents = self.read_log_file_contents(log_file_path) + assert test_message in log_contents, f"Message '{test_message}' should be found in log file" + + def test_verbose_router_logger_info_level(self): + """Test that verbose_router_logger writes to file with INFO level.""" + test_message = "INFO level test message from verbose_router_logger" + + # Log at INFO level + verbose_router_logger.info(test_message) + + # Read log file contents + log_file_path = os.environ.get("LITELLM_LOG_FILE") + assert log_file_path is not None, "LITELLM_LOG_FILE environment variable should be set" + + log_contents = self.read_log_file_contents(log_file_path) + assert test_message in log_contents, f"Message '{test_message}' should be found in log file" + + def test_verbose_router_logger_debug_level(self): + """Test that verbose_router_logger writes to file with DEBUG level.""" + test_message = "DEBUG level test message from verbose_router_logger" + + # Log at DEBUG level + verbose_router_logger.debug(test_message) + + # Read log file contents + log_file_path = os.environ.get("LITELLM_LOG_FILE") + assert log_file_path is not None, "LITELLM_LOG_FILE environment variable should be set" + + log_contents = self.read_log_file_contents(log_file_path) + assert test_message in log_contents, f"Message '{test_message}' should be found in log file" + + def test_log_format_includes_timestamp_and_level(self): + """Test that log entries include timestamp and level information.""" + test_message = "Format test message" + + # Log at INFO level + verbose_logger.info(test_message) + + # Read log file contents + log_file_path = os.environ.get("LITELLM_LOG_FILE") + assert log_file_path is not None, "LITELLM_LOG_FILE environment variable should be set" + + log_contents = self.read_log_file_contents(log_file_path) + + # Check for timestamp format (should be in HH:MM:SS format based on _logging.py) + assert re.search(r'\d{2}:\d{2}:\d{2}', log_contents), "Log should contain timestamp in HH:MM:SS format" + + # Check for level information + assert 'INFO' in log_contents, "Log should contain INFO level indicator" + + # Check for logger name + assert 'LiteLLM' in log_contents, "Log should contain LiteLLM logger name" + + def test_multiple_loggers_write_to_same_file(self): + """Test that all loggers write to the same file.""" + messages = { + 'verbose_logger': "Message from verbose_logger", + 'verbose_proxy_logger': "Message from verbose_proxy_logger", + 'verbose_router_logger': "Message from verbose_router_logger" + } + + # Log messages from different loggers + verbose_logger.info(messages['verbose_logger']) + verbose_proxy_logger.info(messages['verbose_proxy_logger']) + verbose_router_logger.info(messages['verbose_router_logger']) + + # Read log file contents + log_file_path = os.environ.get("LITELLM_LOG_FILE") + assert log_file_path is not None, "LITELLM_LOG_FILE environment variable should be set" + + log_contents = self.read_log_file_contents(log_file_path) + + # Verify all messages are in the same file + for message in messages.values(): + assert message in log_contents, f"Message '{message}' should be found in log file" + + def test_log_file_is_not_empty(self): + """Test that the log file is not empty after logging.""" + # Log a message + verbose_logger.info("Test message to ensure file is not empty") + + # Read log file contents + log_file_path = os.environ.get("LITELLM_LOG_FILE") + assert log_file_path is not None, "LITELLM_LOG_FILE environment variable should be set" + + log_contents = self.read_log_file_contents(log_file_path) + + # Verify file is not empty + assert len(log_contents.strip()) > 0, "Log file should not be empty after logging" + + +class TestJSONLoggingBehavior: + """Test suite to verify JSON logging behavior for all LiteLLM loggers.""" + + def read_log_file_contents(self, log_file_path): + """Helper method to read and return contents of log file.""" + if not os.path.exists(log_file_path): + return "" + + with open(log_file_path, 'r') as f: + return f.read() + + @pytest.fixture(autouse=True) + def setup_json_logging(self, temp_log_file): + """Set up JSON logging environment and ensure proper isolation.""" + self.temp_log_path = temp_log_file + + # Store original environment variables + original_log_file = os.environ.get("LITELLM_LOG_FILE") + original_json_logs = os.environ.get("JSON_LOGS") + + # Set environment variables for JSON logging + os.environ["LITELLM_LOG_FILE"] = temp_log_file + os.environ["JSON_LOGS"] = "True" + + # Force reload of the logging module to pick up new environment variables + import importlib + import litellm._logging + importlib.reload(litellm._logging) + + yield + + # Cleanup: Restore original environment variables + if original_log_file is not None: + os.environ["LITELLM_LOG_FILE"] = original_log_file + else: + os.environ.pop("LITELLM_LOG_FILE", None) + + if original_json_logs is not None: + os.environ["JSON_LOGS"] = original_json_logs + else: + os.environ.pop("JSON_LOGS", None) + + # Reload again to restore original state + importlib.reload(litellm._logging) + + def test_verbose_logger_json_info_level(self): + """Test that verbose_logger writes JSON formatted logs at INFO level.""" + test_message = "JSON INFO level test message from verbose_logger" + + # Log at INFO level + verbose_logger.info(test_message) + + # Force flush all handlers to ensure they write to disk + for handler in verbose_logger.handlers: + if hasattr(handler, 'flush'): + handler.flush() + + # Read log file contents + log_file_path = os.environ.get("LITELLM_LOG_FILE") + assert log_file_path is not None, "LITELLM_LOG_FILE environment variable should be set" + + log_contents = self.read_log_file_contents(log_file_path) + assert log_contents.strip(), "Log file should not be empty" + + # Parse JSON and verify structure + log_lines = [line.strip() for line in log_contents.strip().split('\n') if line.strip()] + assert len(log_lines) > 0, "Should have at least one log line" + + # Find the line containing our test message + target_log = None + for line in log_lines: + try: + parsed = json.loads(line) + if parsed.get("message") == test_message: + target_log = parsed + break + except json.JSONDecodeError: + continue + + assert target_log is not None, f"Could not find JSON log entry with message: {test_message}" + + # Verify JSON structure + assert "message" in target_log, "JSON log should contain 'message' field" + assert "level" in target_log, "JSON log should contain 'level' field" + assert "timestamp" in target_log, "JSON log should contain 'timestamp' field" + + # Verify content + assert target_log["message"] == test_message + assert target_log["level"] == "INFO" + + # Verify timestamp is in ISO 8601 format + timestamp_str = target_log["timestamp"] + try: + datetime.fromisoformat(timestamp_str) + except ValueError: + pytest.fail(f"Timestamp '{timestamp_str}' is not in valid ISO 8601 format") + + def test_verbose_logger_json_debug_level(self): + """Test that verbose_logger writes JSON formatted logs at DEBUG level.""" + test_message = "JSON DEBUG level test message from verbose_logger" + + # Log at DEBUG level + verbose_logger.debug(test_message) + + # Read log file contents + log_file_path = os.environ.get("LITELLM_LOG_FILE") + assert log_file_path is not None, "LITELLM_LOG_FILE environment variable should be set" + + log_contents = self.read_log_file_contents(log_file_path) + assert log_contents.strip(), "Log file should not be empty" + + # Parse JSON and verify structure + log_lines = [line.strip() for line in log_contents.strip().split('\n') if line.strip()] + + # Find the line containing our test message + target_log = None + for line in log_lines: + try: + parsed = json.loads(line) + if parsed.get("message") == test_message: + target_log = parsed + break + except json.JSONDecodeError: + continue + + assert target_log is not None, f"Could not find JSON log entry with message: {test_message}" + assert target_log["level"] == "DEBUG" + + def test_verbose_proxy_logger_json_info_level(self): + """Test that verbose_proxy_logger writes JSON formatted logs at INFO level.""" + test_message = "JSON INFO level test message from verbose_proxy_logger" + + # Log at INFO level + verbose_proxy_logger.info(test_message) + + # Read log file contents + log_file_path = os.environ.get("LITELLM_LOG_FILE") + assert log_file_path is not None, "LITELLM_LOG_FILE environment variable should be set" + + log_contents = self.read_log_file_contents(log_file_path) + assert log_contents.strip(), "Log file should not be empty" + + # Parse JSON and verify structure + log_lines = [line.strip() for line in log_contents.strip().split('\n') if line.strip()] + + # Find the line containing our test message + target_log = None + for line in log_lines: + try: + parsed = json.loads(line) + if parsed.get("message") == test_message: + target_log = parsed + break + except json.JSONDecodeError: + continue + + assert target_log is not None, f"Could not find JSON log entry with message: {test_message}" + + # Verify JSON structure and content + assert target_log["message"] == test_message + assert target_log["level"] == "INFO" + + # Verify timestamp is in ISO 8601 format + timestamp_str = target_log["timestamp"] + try: + datetime.fromisoformat(timestamp_str) + except ValueError: + pytest.fail(f"Timestamp '{timestamp_str}' is not in valid ISO 8601 format") + + def test_verbose_proxy_logger_json_debug_level(self): + """Test that verbose_proxy_logger writes JSON formatted logs at DEBUG level.""" + test_message = "JSON DEBUG level test message from verbose_proxy_logger" + + # Log at DEBUG level + verbose_proxy_logger.debug(test_message) + + # Read log file contents + log_file_path = os.environ.get("LITELLM_LOG_FILE") + assert log_file_path is not None, "LITELLM_LOG_FILE environment variable should be set" + + log_contents = self.read_log_file_contents(log_file_path) + assert log_contents.strip(), "Log file should not be empty" + + # Parse JSON and verify structure + log_lines = [line.strip() for line in log_contents.strip().split('\n') if line.strip()] + + # Find the line containing our test message + target_log = None + for line in log_lines: + try: + parsed = json.loads(line) + if parsed.get("message") == test_message: + target_log = parsed + break + except json.JSONDecodeError: + continue + + assert target_log is not None, f"Could not find JSON log entry with message: {test_message}" + assert target_log["level"] == "DEBUG" + + def test_verbose_router_logger_json_info_level(self): + """Test that verbose_router_logger writes JSON formatted logs at INFO level.""" + test_message = "JSON INFO level test message from verbose_router_logger" + + # Log at INFO level + verbose_router_logger.info(test_message) + + # Read log file contents + log_file_path = os.environ.get("LITELLM_LOG_FILE") + assert log_file_path is not None, "LITELLM_LOG_FILE environment variable should be set" + + log_contents = self.read_log_file_contents(log_file_path) + assert log_contents.strip(), "Log file should not be empty" + + # Parse JSON and verify structure + log_lines = [line.strip() for line in log_contents.strip().split('\n') if line.strip()] + + # Find the line containing our test message + target_log = None + for line in log_lines: + try: + parsed = json.loads(line) + if parsed.get("message") == test_message: + target_log = parsed + break + except json.JSONDecodeError: + continue + + assert target_log is not None, f"Could not find JSON log entry with message: {test_message}" + + # Verify JSON structure and content + assert target_log["message"] == test_message + assert target_log["level"] == "INFO" + + # Verify timestamp is in ISO 8601 format + timestamp_str = target_log["timestamp"] + try: + datetime.fromisoformat(timestamp_str) + except ValueError: + pytest.fail(f"Timestamp '{timestamp_str}' is not in valid ISO 8601 format") + + def test_verbose_router_logger_json_debug_level(self): + """Test that verbose_router_logger writes JSON formatted logs at DEBUG level.""" + test_message = "JSON DEBUG level test message from verbose_router_logger" + + # Log at DEBUG level + verbose_router_logger.debug(test_message) + + # Read log file contents + log_file_path = os.environ.get("LITELLM_LOG_FILE") + assert log_file_path is not None, "LITELLM_LOG_FILE environment variable should be set" + + log_contents = self.read_log_file_contents(log_file_path) + assert log_contents.strip(), "Log file should not be empty" + + # Parse JSON and verify structure + log_lines = [line.strip() for line in log_contents.strip().split('\n') if line.strip()] + + # Find the line containing our test message + target_log = None + for line in log_lines: + try: + parsed = json.loads(line) + if parsed.get("message") == test_message: + target_log = parsed + break + except json.JSONDecodeError: + continue + + assert target_log is not None, f"Could not find JSON log entry with message: {test_message}" + assert target_log["level"] == "DEBUG" + + def test_json_output_is_valid_json(self): + """Test that all JSON log output can be parsed as valid JSON.""" + test_messages = [ + "JSON test message 1", + "JSON test message 2", + "JSON test message 3" + ] + + # Log messages from all loggers + verbose_logger.info(test_messages[0]) + verbose_proxy_logger.info(test_messages[1]) + verbose_router_logger.info(test_messages[2]) + + # Read log file contents + log_file_path = os.environ.get("LITELLM_LOG_FILE") + assert log_file_path is not None, "LITELLM_LOG_FILE environment variable should be set" + + log_contents = self.read_log_file_contents(log_file_path) + assert log_contents.strip(), "Log file should not be empty" + + # Parse each line as JSON + log_lines = [line.strip() for line in log_contents.strip().split('\n') if line.strip()] + parsed_logs = [] + + for line in log_lines: + try: + parsed = json.loads(line) + parsed_logs.append(parsed) + except json.JSONDecodeError as e: + pytest.fail(f"Failed to parse JSON log line: {line}. Error: {e}") + + assert len(parsed_logs) >= len(test_messages), f"Should have at least {len(test_messages)} parsed log entries" + + # Verify each parsed log has required fields + for parsed_log in parsed_logs: + assert isinstance(parsed_log, dict), "Parsed log should be a dictionary" + assert "message" in parsed_log, "Each log should have a 'message' field" + assert "level" in parsed_log, "Each log should have a 'level' field" + assert "timestamp" in parsed_log, "Each log should have a 'timestamp' field" + + def test_json_timestamp_iso8601_format(self): + """Test that JSON log timestamps are in ISO 8601 format.""" + test_message = "Timestamp format test message" + + # Log a message + verbose_logger.info(test_message) + + # Read log file contents + log_file_path = os.environ.get("LITELLM_LOG_FILE") + assert log_file_path is not None, "LITELLM_LOG_FILE environment variable should be set" + + log_contents = self.read_log_file_contents(log_file_path) + assert log_contents.strip(), "Log file should not be empty" + + # Parse JSON and verify timestamp format + log_lines = [line.strip() for line in log_contents.strip().split('\n') if line.strip()] + + # Find the line containing our test message + target_log = None + for line in log_lines: + try: + parsed = json.loads(line) + if parsed.get("message") == test_message: + target_log = parsed + break + except json.JSONDecodeError: + continue + + assert target_log is not None, f"Could not find JSON log entry with message: {test_message}" + + timestamp_str = target_log["timestamp"] + + # Verify timestamp can be parsed as ISO 8601 + try: + parsed_timestamp = datetime.fromisoformat(timestamp_str) + assert isinstance(parsed_timestamp, datetime), "Parsed timestamp should be a datetime object" + except ValueError as e: + pytest.fail(f"Timestamp '{timestamp_str}' is not in valid ISO 8601 format. Error: {e}") + + # Verify timestamp format matches expected pattern (YYYY-MM-DDTHH:MM:SS.ffffff) + import re + iso8601_pattern = r'^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}(\.\d+)?$' + assert re.match(iso8601_pattern, timestamp_str), f"Timestamp '{timestamp_str}' does not match ISO 8601 pattern" + + def test_json_logs_contain_expected_fields(self): + """Test that JSON logs contain all expected fields with correct types.""" + test_message = "Field validation test message" + + # Log a message + verbose_logger.info(test_message) + + # Read log file contents + log_file_path = os.environ.get("LITELLM_LOG_FILE") + assert log_file_path is not None, "LITELLM_LOG_FILE environment variable should be set" + + log_contents = self.read_log_file_contents(log_file_path) + assert log_contents.strip(), "Log file should not be empty" + + # Parse JSON and verify fields + log_lines = [line.strip() for line in log_contents.strip().split('\n') if line.strip()] + + # Find the line containing our test message + target_log = None + for line in log_lines: + try: + parsed = json.loads(line) + if parsed.get("message") == test_message: + target_log = parsed + break + except json.JSONDecodeError: + continue + + assert target_log is not None, f"Could not find JSON log entry with message: {test_message}" + + # Verify required fields exist and have correct types + assert "message" in target_log, "JSON log should contain 'message' field" + assert "level" in target_log, "JSON log should contain 'level' field" + assert "timestamp" in target_log, "JSON log should contain 'timestamp' field" + + assert isinstance(target_log["message"], str), "'message' field should be a string" + assert isinstance(target_log["level"], str), "'level' field should be a string" + assert isinstance(target_log["timestamp"], str), "'timestamp' field should be a string" + + # Verify field values + assert target_log["message"] == test_message + assert target_log["level"] in ["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"], "Level should be a valid log level" + + def test_multiple_json_loggers_write_to_same_file(self): + """Test that all loggers write JSON formatted logs to the same file.""" + messages = { + 'verbose_logger': "JSON message from verbose_logger", + 'verbose_proxy_logger': "JSON message from verbose_proxy_logger", + 'verbose_router_logger': "JSON message from verbose_router_logger" + } + + # Log messages from different loggers + verbose_logger.info(messages['verbose_logger']) + verbose_proxy_logger.info(messages['verbose_proxy_logger']) + verbose_router_logger.info(messages['verbose_router_logger']) + + # Read log file contents + log_file_path = os.environ.get("LITELLM_LOG_FILE") + assert log_file_path is not None, "LITELLM_LOG_FILE environment variable should be set" + + log_contents = self.read_log_file_contents(log_file_path) + assert log_contents.strip(), "Log file should not be empty" + + # Parse all JSON logs + log_lines = [line.strip() for line in log_contents.strip().split('\n') if line.strip()] + parsed_logs = [] + + for line in log_lines: + try: + parsed = json.loads(line) + parsed_logs.append(parsed) + except json.JSONDecodeError: + continue + + # Find logs for each message + found_messages = set() + for parsed_log in parsed_logs: + message = parsed_log.get("message", "") + if message in messages.values(): + found_messages.add(message) + + # Verify all messages are found in JSON format + for message in messages.values(): + assert message in found_messages, f"Message '{message}' should be found in JSON logs" \ No newline at end of file