mirror of
https://github.com/BerriAI/litellm.git
synced 2026-09-09 22:31:41 +00:00
* fix(s3): bound object keys and download filenames to s3 limits Long OpenAI-compatible Responses API ids pushed the s3 object key past s3's 1024 UTF-8 byte cap, so the PUT failed with a 400 and the log record was dropped. Keys that still fit are unchanged, byte for byte. An oversized one now keeps a readable head of the file name and appends the sha256 of the full name. A configured path/alias prefix that is long enough to overflow on its own keeps whole leading path segments, so a prefix-scoped IAM policy or lifecycle rule still matches, and ends in a short digest of the full configured value so two operators do not land in the same folder. The Content-Disposition filename carried the same unbounded id and hit s3's 2048 byte metadata-header cap, so the upload still failed with MetadataTooLarge once the key was bounded. It is bounded the same way, head plus digest, so two records downloaded from the console stay distinct files. The full response id stays in the uploaded JSON payload. * fix(s3): keep the configured prefix whole and spend the whole key budget Shorten the response id first and only trim the operator's configured prefix when the prefix itself is what does not fit, so prefix scoped IAM policies and lifecycle rules keep matching. Trim by bytes rather than whole segments so the longest possible string prefix survives, and route the audit log key through the same shared builder. * chore(s3): trim the comments and docstrings the review flagged Keep the two external facts that are not visible from the code, the 1024 byte object key cap and the 2048 byte metadata header cap, and drop the rest.
2175 lines
77 KiB
Python
2175 lines
77 KiB
Python
import asyncio
|
||
from datetime import datetime
|
||
from unittest.mock import MagicMock, patch
|
||
|
||
import pytest
|
||
|
||
from litellm.integrations.s3_v2 import S3Logger
|
||
from litellm.types.utils import StandardLoggingPayload
|
||
|
||
|
||
class TestS3V2UnitTests:
|
||
"""Test that S3 v2 integration only uses safe_dumps and not json.dumps"""
|
||
|
||
def test_s3_v2_source_code_analysis(self):
|
||
"""Test that S3 v2 source code only imports and uses safe_dumps"""
|
||
import inspect
|
||
|
||
from litellm.integrations import s3_v2
|
||
|
||
# Get the source code of the s3_v2 module
|
||
source_code = inspect.getsource(s3_v2)
|
||
|
||
# Verify that json.dumps is not used directly in the code
|
||
assert (
|
||
"json.dumps(" not in source_code
|
||
), "S3 v2 should not use json.dumps directly"
|
||
|
||
@patch("asyncio.create_task")
|
||
@patch("litellm.integrations.s3_v2.CustomBatchLogger.periodic_flush")
|
||
def test_s3_v2_endpoint_url(self, mock_periodic_flush, mock_create_task):
|
||
"""testing s3 endpoint url"""
|
||
from unittest.mock import AsyncMock, MagicMock
|
||
|
||
from litellm.types.integrations.s3_v2 import s3BatchLoggingElement
|
||
|
||
# Mock periodic_flush and create_task to prevent async task creation during init
|
||
mock_periodic_flush.return_value = None
|
||
mock_create_task.return_value = None
|
||
|
||
# Mock response for all tests
|
||
mock_response = MagicMock()
|
||
mock_response.status_code = 200
|
||
mock_response.raise_for_status = MagicMock()
|
||
|
||
# Create a test batch logging element
|
||
test_element = s3BatchLoggingElement(
|
||
s3_object_key="2025-09-14/test-key.json",
|
||
payload={"test": "data"},
|
||
s3_object_download_filename="test-file.json",
|
||
)
|
||
|
||
# Test 1: Custom endpoint URL with bucket name
|
||
s3_logger = S3Logger(
|
||
s3_bucket_name="test-bucket",
|
||
s3_endpoint_url="https://s3.amazonaws.com",
|
||
s3_aws_access_key_id="test-key",
|
||
s3_aws_secret_access_key="test-secret",
|
||
s3_region_name="us-east-1",
|
||
)
|
||
|
||
s3_logger.async_httpx_client = AsyncMock()
|
||
s3_logger.async_httpx_client.put.return_value = mock_response
|
||
|
||
asyncio.run(s3_logger.async_upload_data_to_s3(test_element))
|
||
|
||
call_args = s3_logger.async_httpx_client.put.call_args
|
||
assert call_args is not None
|
||
url = call_args[0][0]
|
||
expected_url = "https://s3.amazonaws.com/test-bucket/2025-09-14/test-key.json"
|
||
assert url == expected_url, f"Expected URL {expected_url}, got {url}"
|
||
|
||
# Test 2: MinIO-compatible endpoint
|
||
s3_logger_minio = S3Logger(
|
||
s3_bucket_name="litellm-logs",
|
||
s3_endpoint_url="https://minio.example.com:9000",
|
||
s3_aws_access_key_id="minio-key",
|
||
s3_aws_secret_access_key="minio-secret",
|
||
s3_region_name="us-east-1",
|
||
)
|
||
|
||
s3_logger_minio.async_httpx_client = AsyncMock()
|
||
s3_logger_minio.async_httpx_client.put.return_value = mock_response
|
||
|
||
asyncio.run(s3_logger_minio.async_upload_data_to_s3(test_element))
|
||
|
||
call_args_minio = s3_logger_minio.async_httpx_client.put.call_args
|
||
assert call_args_minio is not None
|
||
url_minio = call_args_minio[0][0]
|
||
expected_minio_url = (
|
||
"https://minio.example.com:9000/litellm-logs/2025-09-14/test-key.json"
|
||
)
|
||
assert (
|
||
url_minio == expected_minio_url
|
||
), f"Expected MinIO URL {expected_minio_url}, got {url_minio}"
|
||
|
||
# Test 3: Custom endpoint without bucket name (should fall back to default)
|
||
s3_logger_no_bucket = S3Logger(
|
||
s3_endpoint_url="https://s3.amazonaws.com",
|
||
s3_aws_access_key_id="test-key",
|
||
s3_aws_secret_access_key="test-secret",
|
||
s3_region_name="us-east-1",
|
||
)
|
||
|
||
s3_logger_no_bucket.async_httpx_client = AsyncMock()
|
||
s3_logger_no_bucket.async_httpx_client.put.return_value = mock_response
|
||
|
||
asyncio.run(s3_logger_no_bucket.async_upload_data_to_s3(test_element))
|
||
|
||
call_args_no_bucket = s3_logger_no_bucket.async_httpx_client.put.call_args
|
||
assert call_args_no_bucket is not None
|
||
url_no_bucket = call_args_no_bucket[0][0]
|
||
# Should use default S3 URL format when bucket is missing (bucket becomes None in URL)
|
||
assert "s3.us-east-1.amazonaws.com" in url_no_bucket
|
||
assert "https://" in url_no_bucket
|
||
# Should not include the custom endpoint since bucket is missing
|
||
assert "https://s3.amazonaws.com/" not in url_no_bucket
|
||
|
||
# Test 4: Sync upload method with custom endpoint
|
||
s3_logger_sync = S3Logger(
|
||
s3_bucket_name="sync-bucket",
|
||
s3_endpoint_url="https://custom.s3.endpoint.com",
|
||
s3_aws_access_key_id="sync-key",
|
||
s3_aws_secret_access_key="sync-secret",
|
||
s3_region_name="us-east-1",
|
||
)
|
||
|
||
mock_sync_client = MagicMock()
|
||
mock_sync_client.put.return_value = mock_response
|
||
|
||
with patch(
|
||
"litellm.integrations.s3_v2._get_httpx_client",
|
||
return_value=mock_sync_client,
|
||
):
|
||
s3_logger_sync.upload_data_to_s3(test_element)
|
||
|
||
call_args_sync = mock_sync_client.put.call_args
|
||
assert call_args_sync is not None
|
||
url_sync = call_args_sync[0][0]
|
||
expected_sync_url = (
|
||
"https://custom.s3.endpoint.com/sync-bucket/2025-09-14/test-key.json"
|
||
)
|
||
assert (
|
||
url_sync == expected_sync_url
|
||
), f"Expected sync URL {expected_sync_url}, got {url_sync}"
|
||
|
||
# Test 5: Download method with custom endpoint
|
||
s3_logger_download = S3Logger(
|
||
s3_bucket_name="download-bucket",
|
||
s3_endpoint_url="https://download.s3.endpoint.com",
|
||
s3_aws_access_key_id="download-key",
|
||
s3_aws_secret_access_key="download-secret",
|
||
s3_region_name="us-east-1",
|
||
)
|
||
|
||
mock_download_response = MagicMock()
|
||
mock_download_response.status_code = 200
|
||
mock_download_response.json = MagicMock(return_value={"downloaded": "data"})
|
||
s3_logger_download.async_httpx_client = AsyncMock()
|
||
s3_logger_download.async_httpx_client.get.return_value = mock_download_response
|
||
|
||
result = asyncio.run(
|
||
s3_logger_download._download_object_from_s3(
|
||
"2025-09-14/download-test-key.json"
|
||
)
|
||
)
|
||
|
||
call_args_download = s3_logger_download.async_httpx_client.get.call_args
|
||
assert call_args_download is not None
|
||
url_download = call_args_download[0][0]
|
||
expected_download_url = "https://download.s3.endpoint.com/download-bucket/2025-09-14/download-test-key.json"
|
||
assert (
|
||
url_download == expected_download_url
|
||
), f"Expected download URL {expected_download_url}, got {url_download}"
|
||
|
||
assert result == {"downloaded": "data"}
|
||
|
||
@patch("asyncio.create_task")
|
||
@patch("litellm.integrations.s3_v2.CustomBatchLogger.periodic_flush")
|
||
def test_s3_v2_virtual_hosted_style(self, mock_periodic_flush, mock_create_task):
|
||
"""Test s3_use_virtual_hosted_style parameter for virtual-hosted-style URLs"""
|
||
from unittest.mock import AsyncMock, MagicMock
|
||
|
||
from litellm.types.integrations.s3_v2 import s3BatchLoggingElement
|
||
|
||
# Mock periodic_flush and create_task to prevent async task creation during init
|
||
mock_periodic_flush.return_value = None
|
||
mock_create_task.return_value = None
|
||
|
||
# Mock response for all tests
|
||
mock_response = MagicMock()
|
||
mock_response.status_code = 200
|
||
mock_response.raise_for_status = MagicMock()
|
||
|
||
# Create a test batch logging element
|
||
test_element = s3BatchLoggingElement(
|
||
s3_object_key="2025-09-14/test-key.json",
|
||
payload={"test": "data"},
|
||
s3_object_download_filename="test-file.json",
|
||
)
|
||
|
||
# Test 1: Virtual-hosted-style with custom endpoint
|
||
s3_logger_virtual = S3Logger(
|
||
s3_bucket_name="test-bucket",
|
||
s3_endpoint_url="https://s3.custom-endpoint.com",
|
||
s3_aws_access_key_id="test-key",
|
||
s3_aws_secret_access_key="test-secret",
|
||
s3_region_name="us-east-1",
|
||
s3_use_virtual_hosted_style=True,
|
||
)
|
||
|
||
s3_logger_virtual.async_httpx_client = AsyncMock()
|
||
s3_logger_virtual.async_httpx_client.put.return_value = mock_response
|
||
|
||
asyncio.run(s3_logger_virtual.async_upload_data_to_s3(test_element))
|
||
|
||
call_args = s3_logger_virtual.async_httpx_client.put.call_args
|
||
assert call_args is not None
|
||
url = call_args[0][0]
|
||
expected_url = (
|
||
"https://test-bucket.s3.custom-endpoint.com/2025-09-14/test-key.json"
|
||
)
|
||
assert (
|
||
url == expected_url
|
||
), f"Expected virtual-hosted-style URL {expected_url}, got {url}"
|
||
|
||
# Test 2: Path-style (default behavior with s3_use_virtual_hosted_style=False)
|
||
s3_logger_path = S3Logger(
|
||
s3_bucket_name="test-bucket",
|
||
s3_endpoint_url="https://s3.custom-endpoint.com",
|
||
s3_aws_access_key_id="test-key",
|
||
s3_aws_secret_access_key="test-secret",
|
||
s3_region_name="us-east-1",
|
||
s3_use_virtual_hosted_style=False,
|
||
)
|
||
|
||
s3_logger_path.async_httpx_client = AsyncMock()
|
||
s3_logger_path.async_httpx_client.put.return_value = mock_response
|
||
|
||
asyncio.run(s3_logger_path.async_upload_data_to_s3(test_element))
|
||
|
||
call_args_path = s3_logger_path.async_httpx_client.put.call_args
|
||
assert call_args_path is not None
|
||
url_path = call_args_path[0][0]
|
||
expected_path_url = (
|
||
"https://s3.custom-endpoint.com/test-bucket/2025-09-14/test-key.json"
|
||
)
|
||
assert (
|
||
url_path == expected_path_url
|
||
), f"Expected path-style URL {expected_path_url}, got {url_path}"
|
||
|
||
# Test 3: Virtual-hosted-style with http protocol
|
||
s3_logger_http = S3Logger(
|
||
s3_bucket_name="http-bucket",
|
||
s3_endpoint_url="http://minio.local:9000",
|
||
s3_aws_access_key_id="minio-key",
|
||
s3_aws_secret_access_key="minio-secret",
|
||
s3_region_name="us-east-1",
|
||
s3_use_virtual_hosted_style=True,
|
||
)
|
||
|
||
s3_logger_http.async_httpx_client = AsyncMock()
|
||
s3_logger_http.async_httpx_client.put.return_value = mock_response
|
||
|
||
asyncio.run(s3_logger_http.async_upload_data_to_s3(test_element))
|
||
|
||
call_args_http = s3_logger_http.async_httpx_client.put.call_args
|
||
assert call_args_http is not None
|
||
url_http = call_args_http[0][0]
|
||
expected_http_url = (
|
||
"http://http-bucket.minio.local:9000/2025-09-14/test-key.json"
|
||
)
|
||
assert (
|
||
url_http == expected_http_url
|
||
), f"Expected virtual-hosted-style URL with http {expected_http_url}, got {url_http}"
|
||
|
||
# Test 4: Sync upload method with virtual-hosted-style
|
||
s3_logger_sync_virtual = S3Logger(
|
||
s3_bucket_name="sync-bucket",
|
||
s3_endpoint_url="https://storage.example.com",
|
||
s3_aws_access_key_id="sync-key",
|
||
s3_aws_secret_access_key="sync-secret",
|
||
s3_region_name="us-east-1",
|
||
s3_use_virtual_hosted_style=True,
|
||
)
|
||
|
||
mock_sync_client = MagicMock()
|
||
mock_sync_client.put.return_value = mock_response
|
||
|
||
with patch(
|
||
"litellm.integrations.s3_v2._get_httpx_client",
|
||
return_value=mock_sync_client,
|
||
):
|
||
s3_logger_sync_virtual.upload_data_to_s3(test_element)
|
||
|
||
call_args_sync = mock_sync_client.put.call_args
|
||
assert call_args_sync is not None
|
||
url_sync = call_args_sync[0][0]
|
||
expected_sync_url = (
|
||
"https://sync-bucket.storage.example.com/2025-09-14/test-key.json"
|
||
)
|
||
assert (
|
||
url_sync == expected_sync_url
|
||
), f"Expected virtual-hosted-style sync URL {expected_sync_url}, got {url_sync}"
|
||
|
||
# Test 5: Download method with virtual-hosted-style
|
||
s3_logger_download_virtual = S3Logger(
|
||
s3_bucket_name="download-bucket",
|
||
s3_endpoint_url="https://download.endpoint.com",
|
||
s3_aws_access_key_id="download-key",
|
||
s3_aws_secret_access_key="download-secret",
|
||
s3_region_name="us-east-1",
|
||
s3_use_virtual_hosted_style=True,
|
||
)
|
||
|
||
mock_download_response = MagicMock()
|
||
mock_download_response.status_code = 200
|
||
mock_download_response.json = MagicMock(return_value={"downloaded": "data"})
|
||
s3_logger_download_virtual.async_httpx_client = AsyncMock()
|
||
s3_logger_download_virtual.async_httpx_client.get.return_value = (
|
||
mock_download_response
|
||
)
|
||
|
||
result = asyncio.run(
|
||
s3_logger_download_virtual._download_object_from_s3(
|
||
"2025-09-14/download-test-key.json"
|
||
)
|
||
)
|
||
|
||
call_args_download = s3_logger_download_virtual.async_httpx_client.get.call_args
|
||
assert call_args_download is not None
|
||
url_download = call_args_download[0][0]
|
||
expected_download_url = "https://download-bucket.download.endpoint.com/2025-09-14/download-test-key.json"
|
||
assert (
|
||
url_download == expected_download_url
|
||
), f"Expected virtual-hosted-style download URL {expected_download_url}, got {url_download}"
|
||
|
||
assert result == {"downloaded": "data"}
|
||
|
||
@patch("asyncio.create_task")
|
||
@patch("litellm.integrations.s3_v2.CustomBatchLogger.periodic_flush")
|
||
def test_s3_v2_put_url_encodes_spaces_in_object_key(
|
||
self, mock_periodic_flush, mock_create_task
|
||
):
|
||
import requests
|
||
from unittest.mock import AsyncMock
|
||
|
||
from litellm.types.integrations.s3_v2 import s3BatchLoggingElement
|
||
|
||
mock_periodic_flush.return_value = None
|
||
mock_create_task.return_value = None
|
||
|
||
mock_response = MagicMock()
|
||
mock_response.status_code = 200
|
||
mock_response.raise_for_status = MagicMock()
|
||
|
||
s3_object_key = "My Team/2025-09-14/test-key.json"
|
||
test_element = s3BatchLoggingElement(
|
||
s3_object_key=s3_object_key,
|
||
payload={"test": "data"},
|
||
s3_object_download_filename="test-file.json",
|
||
)
|
||
|
||
s3_logger = S3Logger(
|
||
s3_bucket_name="test-bucket",
|
||
s3_endpoint_url="https://s3.amazonaws.com",
|
||
s3_aws_access_key_id="test-key",
|
||
s3_aws_secret_access_key="test-secret",
|
||
s3_region_name="us-east-1",
|
||
)
|
||
s3_logger.async_httpx_client = AsyncMock()
|
||
s3_logger.async_httpx_client.put.return_value = mock_response
|
||
|
||
asyncio.run(s3_logger.async_upload_data_to_s3(test_element))
|
||
|
||
call_args = s3_logger.async_httpx_client.put.call_args
|
||
assert call_args is not None
|
||
actual_url = call_args[0][0]
|
||
raw_url = f"https://s3.amazonaws.com/test-bucket/{s3_object_key}"
|
||
expected_url = requests.Request("PUT", raw_url).prepare().url
|
||
assert actual_url == expected_url
|
||
assert " " not in actual_url
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_async_upload_retries_on_s3_503():
|
||
"""
|
||
Test that async_upload_data_to_s3 retries on transient S3 503 Slow Down
|
||
and succeeds on the second attempt.
|
||
"""
|
||
from unittest.mock import AsyncMock, MagicMock
|
||
|
||
from litellm.types.integrations.s3_v2 import s3BatchLoggingElement
|
||
|
||
logger = S3Logger(
|
||
s3_bucket_name="test-bucket",
|
||
s3_aws_access_key_id="test-key",
|
||
s3_aws_secret_access_key="test-secret",
|
||
s3_region_name="us-east-1",
|
||
)
|
||
|
||
test_element = s3BatchLoggingElement(
|
||
s3_object_key="2025-09-14/test-retry.json",
|
||
payload={"test": "retry"},
|
||
s3_object_download_filename="test-retry.json",
|
||
)
|
||
|
||
# First call returns 503, second call returns 200
|
||
response_503 = MagicMock()
|
||
response_503.status_code = 503
|
||
response_200 = MagicMock()
|
||
response_200.status_code = 200
|
||
response_200.raise_for_status = MagicMock()
|
||
|
||
logger.async_httpx_client = AsyncMock()
|
||
logger.async_httpx_client.put = AsyncMock(side_effect=[response_503, response_200])
|
||
|
||
with patch("asyncio.sleep", new_callable=AsyncMock) as mock_sleep:
|
||
await logger.async_upload_data_to_s3(test_element)
|
||
|
||
# Verify PUT was called twice (retry after 503)
|
||
assert logger.async_httpx_client.put.call_count == 2
|
||
# Verify sleep was called with the backoff delay
|
||
mock_sleep.assert_called_once_with(1) # 2**0 = 1s
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_async_upload_retries_on_s3_500():
|
||
"""
|
||
Test that async_upload_data_to_s3 retries on transient S3 500 errors.
|
||
"""
|
||
from unittest.mock import AsyncMock, MagicMock
|
||
|
||
from litellm.types.integrations.s3_v2 import s3BatchLoggingElement
|
||
|
||
logger = S3Logger(
|
||
s3_bucket_name="test-bucket",
|
||
s3_aws_access_key_id="test-key",
|
||
s3_aws_secret_access_key="test-secret",
|
||
s3_region_name="us-east-1",
|
||
)
|
||
|
||
test_element = s3BatchLoggingElement(
|
||
s3_object_key="2025-09-14/test-retry-500.json",
|
||
payload={"test": "retry-500"},
|
||
s3_object_download_filename="test-retry-500.json",
|
||
)
|
||
|
||
response_500 = MagicMock()
|
||
response_500.status_code = 500
|
||
response_200 = MagicMock()
|
||
response_200.status_code = 200
|
||
response_200.raise_for_status = MagicMock()
|
||
|
||
logger.async_httpx_client = AsyncMock()
|
||
logger.async_httpx_client.put = AsyncMock(side_effect=[response_500, response_200])
|
||
|
||
with patch("asyncio.sleep", new_callable=AsyncMock) as mock_sleep:
|
||
await logger.async_upload_data_to_s3(test_element)
|
||
|
||
assert logger.async_httpx_client.put.call_count == 2
|
||
mock_sleep.assert_called_once_with(1)
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_async_upload_exhausts_retries_on_persistent_503():
|
||
"""
|
||
Test that async_upload_data_to_s3 raises after exhausting all retries
|
||
on persistent S3 503.
|
||
"""
|
||
from unittest.mock import AsyncMock, MagicMock
|
||
|
||
from litellm.types.integrations.s3_v2 import s3BatchLoggingElement
|
||
|
||
logger = S3Logger(
|
||
s3_bucket_name="test-bucket",
|
||
s3_aws_access_key_id="test-key",
|
||
s3_aws_secret_access_key="test-secret",
|
||
s3_region_name="us-east-1",
|
||
)
|
||
|
||
test_element = s3BatchLoggingElement(
|
||
s3_object_key="2025-09-14/test-exhaust.json",
|
||
payload={"test": "exhaust"},
|
||
s3_object_download_filename="test-exhaust.json",
|
||
)
|
||
|
||
# All 3 attempts return 503
|
||
response_503 = MagicMock()
|
||
response_503.status_code = 503
|
||
response_503.raise_for_status = MagicMock(
|
||
side_effect=Exception("503 Service Unavailable")
|
||
)
|
||
|
||
logger.async_httpx_client = AsyncMock()
|
||
logger.async_httpx_client.put = AsyncMock(return_value=response_503)
|
||
|
||
with patch("asyncio.sleep", new_callable=AsyncMock) as mock_sleep:
|
||
with patch.object(logger, "handle_callback_failure") as mock_failure:
|
||
await logger.async_upload_data_to_s3(test_element)
|
||
|
||
# 3 PUT attempts total
|
||
assert logger.async_httpx_client.put.call_count == 3
|
||
# 2 sleeps (between attempts 1-2 and 2-3)
|
||
assert mock_sleep.call_count == 2
|
||
# Callback failure handler called after exhausting retries
|
||
mock_failure.assert_called_once_with(callback_name="S3Logger")
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_async_upload_no_retry_on_4xx():
|
||
"""
|
||
Test that async_upload_data_to_s3 does NOT retry on 4xx errors (client errors).
|
||
"""
|
||
from unittest.mock import AsyncMock, MagicMock
|
||
|
||
from litellm.types.integrations.s3_v2 import s3BatchLoggingElement
|
||
|
||
logger = S3Logger(
|
||
s3_bucket_name="test-bucket",
|
||
s3_aws_access_key_id="test-key",
|
||
s3_aws_secret_access_key="test-secret",
|
||
s3_region_name="us-east-1",
|
||
)
|
||
|
||
test_element = s3BatchLoggingElement(
|
||
s3_object_key="2025-09-14/test-no-retry.json",
|
||
payload={"test": "no-retry"},
|
||
s3_object_download_filename="test-no-retry.json",
|
||
)
|
||
|
||
response_403 = MagicMock()
|
||
response_403.status_code = 403
|
||
response_403.raise_for_status = MagicMock(side_effect=Exception("403 Forbidden"))
|
||
|
||
logger.async_httpx_client = AsyncMock()
|
||
logger.async_httpx_client.put = AsyncMock(return_value=response_403)
|
||
|
||
with patch.object(logger, "handle_callback_failure") as mock_failure:
|
||
await logger.async_upload_data_to_s3(test_element)
|
||
|
||
# Only 1 attempt — no retry for 4xx
|
||
assert logger.async_httpx_client.put.call_count == 1
|
||
mock_failure.assert_called_once_with(callback_name="S3Logger")
|
||
|
||
|
||
def test_sync_upload_retries_on_s3_503():
|
||
"""
|
||
Test that the sync upload_data_to_s3 retries on transient S3 503.
|
||
"""
|
||
from unittest.mock import MagicMock
|
||
|
||
from litellm.types.integrations.s3_v2 import s3BatchLoggingElement
|
||
|
||
logger = S3Logger(
|
||
s3_bucket_name="test-bucket",
|
||
s3_aws_access_key_id="test-key",
|
||
s3_aws_secret_access_key="test-secret",
|
||
s3_region_name="us-east-1",
|
||
)
|
||
|
||
test_element = s3BatchLoggingElement(
|
||
s3_object_key="2025-09-14/test-sync-retry.json",
|
||
payload={"test": "sync-retry"},
|
||
s3_object_download_filename="test-sync-retry.json",
|
||
)
|
||
|
||
response_503 = MagicMock()
|
||
response_503.status_code = 503
|
||
response_200 = MagicMock()
|
||
response_200.status_code = 200
|
||
response_200.raise_for_status = MagicMock()
|
||
|
||
mock_sync_client = MagicMock()
|
||
mock_sync_client.put = MagicMock(side_effect=[response_503, response_200])
|
||
|
||
with patch(
|
||
"litellm.integrations.s3_v2._get_httpx_client",
|
||
return_value=mock_sync_client,
|
||
):
|
||
with patch("time.sleep") as mock_sleep:
|
||
logger.upload_data_to_s3(test_element)
|
||
|
||
assert mock_sync_client.put.call_count == 2
|
||
mock_sleep.assert_called_once_with(1)
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_async_log_event_skips_when_standard_logging_object_missing():
|
||
"""
|
||
Reproduces the bug where _async_log_event_base raises ValueError when
|
||
kwargs has no standard_logging_object (e.g. call_type=afile_delete).
|
||
|
||
The S3 logger should skip gracefully, not raise.
|
||
"""
|
||
logger = S3Logger(
|
||
s3_bucket_name="test-bucket",
|
||
s3_region_name="us-east-1",
|
||
s3_aws_access_key_id="fake",
|
||
s3_aws_secret_access_key="fake",
|
||
)
|
||
|
||
kwargs_without_slo = {
|
||
"call_type": "afile_delete",
|
||
"model": None,
|
||
"litellm_call_id": "test-call-id",
|
||
}
|
||
|
||
start_time = datetime.utcnow()
|
||
end_time = datetime.utcnow()
|
||
|
||
# Spy on handle_callback_failure — should NOT be called if we skip gracefully.
|
||
# Without the fix, the ValueError is caught by the except block which calls
|
||
# handle_callback_failure. With the fix, we return early and never hit except.
|
||
with patch.object(logger, "handle_callback_failure") as mock_failure:
|
||
await logger._async_log_event_base(
|
||
kwargs=kwargs_without_slo,
|
||
response_obj=None,
|
||
start_time=start_time,
|
||
end_time=end_time,
|
||
)
|
||
|
||
assert not mock_failure.called, (
|
||
"handle_callback_failure should not be called — "
|
||
"missing standard_logging_object should be a graceful skip, not an error"
|
||
)
|
||
|
||
# Nothing should have been queued (catches the case where code falls
|
||
# through without returning and appends None to the queue)
|
||
assert (
|
||
len(logger.log_queue) == 0
|
||
), "log_queue should be empty when standard_logging_object is missing"
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_strip_base64_removes_file_and_nontext_entries():
|
||
logger = S3Logger(s3_strip_base64_files=True)
|
||
|
||
payload = {
|
||
"messages": [
|
||
{
|
||
"role": "user",
|
||
"content": [
|
||
{"type": "text", "text": "Hello world"},
|
||
{
|
||
"type": "image",
|
||
"file": {"file_data": "data:image/png;base64,AAAA"},
|
||
},
|
||
{
|
||
"type": "file",
|
||
"file": {"file_data": "data:application/pdf;base64,BBBB"},
|
||
},
|
||
],
|
||
},
|
||
{
|
||
"role": "assistant",
|
||
"content": [
|
||
{"type": "text", "text": "Response"},
|
||
{
|
||
"type": "audio",
|
||
"file": {"file_data": "data:audio/wav;base64,CCCC"},
|
||
},
|
||
],
|
||
},
|
||
]
|
||
}
|
||
|
||
stripped = await logger._strip_base64_from_messages(payload)
|
||
|
||
# 1️⃣ File/image/audio entries are removed
|
||
assert len(stripped["messages"][0]["content"]) == 1
|
||
assert stripped["messages"][0]["content"][0]["text"] == "Hello world"
|
||
|
||
assert len(stripped["messages"][1]["content"]) == 1
|
||
assert stripped["messages"][1]["content"][0]["text"] == "Response"
|
||
|
||
# 2️⃣ No 'file' keys remain
|
||
for msg in stripped["messages"]:
|
||
for content in msg["content"]:
|
||
assert "file" not in content
|
||
assert content.get("type") == "text"
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_strip_base64_keeps_non_file_content():
|
||
logger = S3Logger(s3_strip_base64_files=True)
|
||
|
||
payload = {
|
||
"messages": [
|
||
{
|
||
"role": "user",
|
||
"content": [
|
||
{"type": "text", "text": "Just text"},
|
||
{"type": "text", "text": "Another message"},
|
||
],
|
||
}
|
||
]
|
||
}
|
||
|
||
stripped = await logger._strip_base64_from_messages(payload)
|
||
|
||
# Should not modify pure text messages
|
||
assert stripped["messages"][0]["content"] == payload["messages"][0]["content"]
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_strip_base64_handles_empty_or_missing_messages():
|
||
logger = S3Logger(s3_strip_base64_files=True)
|
||
|
||
# Missing messages key
|
||
payload_no_messages = {}
|
||
stripped1 = await logger._strip_base64_from_messages(payload_no_messages)
|
||
assert stripped1 == payload_no_messages
|
||
|
||
# Empty messages list
|
||
payload_empty = {"messages": []}
|
||
stripped2 = await logger._strip_base64_from_messages(payload_empty)
|
||
assert stripped2 == payload_empty
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_strip_base64_mixed_nested_objects():
|
||
"""
|
||
Handles weird/nested content structures gracefully.
|
||
"""
|
||
logger = S3Logger(s3_strip_base64_files=True)
|
||
|
||
payload = {
|
||
"messages": [
|
||
{
|
||
"role": "system",
|
||
"content": [
|
||
{"type": "text", "text": "Keep me"},
|
||
{"type": "custom", "metadata": "ignore but non-text"},
|
||
{"foo": "bar"},
|
||
{"file": {"file_data": "data:application/pdf;base64,XXX"}},
|
||
],
|
||
"extra": {"trace_id": "123"},
|
||
}
|
||
]
|
||
}
|
||
|
||
stripped = await logger._strip_base64_from_messages(payload)
|
||
|
||
# Custom/non-text and file entries removed
|
||
content = stripped["messages"][0]["content"]
|
||
assert len(content) == 2
|
||
assert {"type": "text", "text": "Keep me"} in content
|
||
assert {"foo": "bar"} in content
|
||
# Extra metadata preserved
|
||
assert stripped["messages"][0]["extra"]["trace_id"] == "123"
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_s3_verify_false_handling(monkeypatch: pytest.MonkeyPatch):
|
||
"""
|
||
Test that s3_verify=False is properly handled and not treated as None.
|
||
|
||
This is a regression test for the bug where s3_verify=False was being
|
||
ignored because 'False or s3_verify' would evaluate to s3_verify (None).
|
||
"""
|
||
from unittest.mock import AsyncMock, patch
|
||
|
||
import litellm
|
||
|
||
# Set up s3_callback_params with s3_verify=False
|
||
monkeypatch.setattr(
|
||
litellm,
|
||
"s3_callback_params",
|
||
{
|
||
"s3_bucket_name": "test-bucket",
|
||
"s3_endpoint_url": "https://localhost:443",
|
||
"s3_aws_access_key_id": "minioadmin",
|
||
"s3_aws_secret_access_key": "minioadmin",
|
||
"s3_region_name": "us-east-1",
|
||
"s3_verify": False, # This should NOT be ignored
|
||
"s3_use_ssl": False, # This should also NOT be ignored
|
||
},
|
||
)
|
||
|
||
with patch("asyncio.create_task"):
|
||
with patch(
|
||
"litellm.integrations.s3_v2.get_async_httpx_client"
|
||
) as mock_get_client:
|
||
mock_client = AsyncMock()
|
||
mock_get_client.return_value = mock_client
|
||
|
||
# Create logger
|
||
logger = S3Logger()
|
||
|
||
# Verify s3_verify is False, not None
|
||
assert (
|
||
logger.s3_verify is False
|
||
), f"Expected s3_verify=False, got {logger.s3_verify}"
|
||
assert (
|
||
logger.s3_use_ssl is False
|
||
), f"Expected s3_use_ssl=False, got {logger.s3_use_ssl}"
|
||
|
||
# Verify that get_async_httpx_client was called with ssl_verify=False
|
||
mock_get_client.assert_called_once()
|
||
call_kwargs = mock_get_client.call_args.kwargs
|
||
assert (
|
||
"params" in call_kwargs
|
||
), "params should be passed to get_async_httpx_client"
|
||
assert call_kwargs["params"] == {
|
||
"ssl_verify": False
|
||
}, f"Expected ssl_verify=False in params, got {call_kwargs.get('params')}"
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_s3_verify_none_handling(monkeypatch: pytest.MonkeyPatch):
|
||
"""
|
||
Test that s3_verify=None uses default behavior.
|
||
"""
|
||
from unittest.mock import AsyncMock, patch
|
||
|
||
import litellm
|
||
|
||
# Set up s3_callback_params without s3_verify
|
||
monkeypatch.setattr(
|
||
litellm,
|
||
"s3_callback_params",
|
||
{
|
||
"s3_bucket_name": "test-bucket",
|
||
"s3_aws_access_key_id": "test-key",
|
||
"s3_aws_secret_access_key": "test-secret",
|
||
"s3_region_name": "us-east-1",
|
||
},
|
||
)
|
||
|
||
with patch("asyncio.create_task"):
|
||
with patch(
|
||
"litellm.integrations.s3_v2.get_async_httpx_client"
|
||
) as mock_get_client:
|
||
mock_client = AsyncMock()
|
||
mock_get_client.return_value = mock_client
|
||
|
||
# Create logger without explicit s3_verify
|
||
logger = S3Logger()
|
||
|
||
# Verify s3_verify is None (default)
|
||
assert (
|
||
logger.s3_verify is None
|
||
), f"Expected s3_verify=None, got {logger.s3_verify}"
|
||
|
||
# Verify that get_async_httpx_client was called
|
||
mock_get_client.assert_called_once()
|
||
call_kwargs = mock_get_client.call_args.kwargs
|
||
# When s3_verify is None, params={'ssl_verify': None} which is fine - uses default behavior
|
||
# The important thing is it's not False
|
||
if "params" in call_kwargs and call_kwargs["params"] is not None:
|
||
assert call_kwargs["params"].get("ssl_verify") is None
|
||
# Either params is None or params={'ssl_verify': None} is acceptable
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_s3_verify_false_creates_httpx_client_with_verify_false(monkeypatch: pytest.MonkeyPatch):
|
||
"""
|
||
Test that when s3_verify=False, the actual httpx client has verify=False.
|
||
|
||
This validates that ssl_verify=False flows through to the httpx.AsyncClient.
|
||
"""
|
||
from unittest.mock import patch
|
||
|
||
import litellm
|
||
|
||
# Set up s3_callback_params with s3_verify=False
|
||
monkeypatch.setattr(
|
||
litellm,
|
||
"s3_callback_params",
|
||
{
|
||
"s3_bucket_name": "test-bucket",
|
||
"s3_endpoint_url": "https://localhost:443",
|
||
"s3_aws_access_key_id": "minioadmin",
|
||
"s3_aws_secret_access_key": "minioadmin",
|
||
"s3_region_name": "us-east-1",
|
||
"s3_verify": False,
|
||
},
|
||
)
|
||
|
||
with patch("asyncio.create_task"):
|
||
# Create logger - this creates the httpx client
|
||
logger = S3Logger()
|
||
|
||
# Verify the logger has s3_verify=False
|
||
assert logger.s3_verify is False
|
||
|
||
# Check the actual httpx client has verify=False
|
||
# The async_httpx_client.client is the actual httpx.AsyncClient
|
||
if hasattr(logger.async_httpx_client, "client"):
|
||
httpx_client = logger.async_httpx_client.client
|
||
# Check the _verify attribute (httpx internal)
|
||
if hasattr(httpx_client, "_verify"):
|
||
assert (
|
||
httpx_client._verify is False
|
||
), f"Expected httpx client _verify=False, got {httpx_client._verify}"
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_s3_verify_false_async_client(monkeypatch: pytest.MonkeyPatch):
|
||
"""
|
||
Test that the async httpx client respects s3_verify=False.
|
||
"""
|
||
from unittest.mock import AsyncMock, MagicMock, patch
|
||
|
||
import litellm
|
||
from litellm.types.integrations.s3_v2 import s3BatchLoggingElement
|
||
|
||
# Set up s3_callback_params with s3_verify=False
|
||
monkeypatch.setattr(
|
||
litellm,
|
||
"s3_callback_params",
|
||
{
|
||
"s3_bucket_name": "test-bucket",
|
||
"s3_endpoint_url": "https://localhost:443",
|
||
"s3_aws_access_key_id": "minioadmin",
|
||
"s3_aws_secret_access_key": "minioadmin",
|
||
"s3_region_name": "us-east-1",
|
||
"s3_verify": False,
|
||
},
|
||
)
|
||
|
||
with patch("asyncio.create_task"):
|
||
logger = S3Logger()
|
||
|
||
# Verify s3_verify is False
|
||
assert logger.s3_verify is False
|
||
|
||
# Create test element
|
||
test_element = s3BatchLoggingElement(
|
||
s3_object_key="2025-11-03/test-key.json",
|
||
payload={"test": "data"},
|
||
s3_object_download_filename="test-file.json",
|
||
)
|
||
|
||
# Mock the async httpx client's put method
|
||
mock_response = MagicMock()
|
||
mock_response.status_code = 200
|
||
mock_response.raise_for_status = MagicMock()
|
||
logger.async_httpx_client.put = AsyncMock(return_value=mock_response)
|
||
|
||
# Call async upload
|
||
await logger.async_upload_data_to_s3(test_element)
|
||
|
||
# Verify put was called
|
||
assert logger.async_httpx_client.put.called
|
||
|
||
# Check that the async httpx client was created with verify=False
|
||
if hasattr(logger.async_httpx_client, "client"):
|
||
httpx_client = logger.async_httpx_client.client
|
||
if hasattr(httpx_client, "_verify"):
|
||
assert (
|
||
httpx_client._verify is False
|
||
), f"Expected async httpx client _verify=False, got {httpx_client._verify}"
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_strip_base64_recursive_redaction():
|
||
logger = S3Logger(s3_strip_base64_files=True)
|
||
payload = {
|
||
"messages": [
|
||
{
|
||
"content": [
|
||
{"type": "text", "text": "normal text"},
|
||
{
|
||
"type": "text",
|
||
"text": "data:image/png;base64,iVBORw0KGgoAAAANSUhEUg",
|
||
},
|
||
{
|
||
"type": "text",
|
||
"text": "Nested: {'data': 'data:application/pdf;base64,AAA...'}",
|
||
},
|
||
{"file": {"file_data": "data:application/pdf;base64,AAAA"}},
|
||
{"metadata": {"preview": "data:audio/mp3;base64,AAAAA=="}},
|
||
]
|
||
}
|
||
]
|
||
}
|
||
|
||
result = await logger._strip_base64_from_messages(payload)
|
||
content = result["messages"][0]["content"]
|
||
|
||
# Dropped file-type entries
|
||
assert not any("file" in c for c in content)
|
||
|
||
# Base64 redacted globally
|
||
import json
|
||
|
||
for c in content:
|
||
if isinstance(c, dict):
|
||
s = json.dumps(c).lower()
|
||
# "[base64_redacted]" is fine, but raw base64 is not
|
||
assert "base64," not in s, f"Found real base64 blob in: {s}"
|
||
|
||
|
||
# --------------------------------------------------------------
|
||
# Shared fixture that silences asyncio.create_task during tests
|
||
# --------------------------------------------------------------
|
||
@pytest.fixture(autouse=True)
|
||
def patch_asyncio_create_task():
|
||
"""Prevent 'no running event loop' errors when S3Logger calls asyncio.create_task()."""
|
||
with patch("asyncio.create_task"):
|
||
yield
|
||
|
||
|
||
# --------------------------------------------------------------
|
||
# Parametrized prefix combination test
|
||
# --------------------------------------------------------------
|
||
@pytest.mark.parametrize(
|
||
"use_team_prefix,use_key_prefix,team_alias,key_alias,expected_prefix",
|
||
[
|
||
(False, False, "teamA", "keyA", ""),
|
||
(True, False, "teamA", "keyA", "teamA/"),
|
||
(False, True, "teamA", "keyA", "keyA/"),
|
||
(True, True, "teamA", "keyA", "teamA/keyA/"),
|
||
(True, True, None, "keyA", "keyA/"),
|
||
(True, True, "teamA", None, "teamA/"),
|
||
(True, True, None, None, ""),
|
||
],
|
||
)
|
||
def test_s3_object_key_prefix_combinations(
|
||
use_team_prefix, use_key_prefix, team_alias, key_alias, expected_prefix
|
||
):
|
||
"""
|
||
Validate correct S3 prefix composition for team alias + key alias combinations.
|
||
"""
|
||
with patch("litellm.integrations.s3_v2.get_s3_object_key") as mock_get_key:
|
||
mock_get_key.return_value = "mocked/s3/object/key.json"
|
||
|
||
logger = S3Logger(
|
||
s3_bucket_name="test-bucket",
|
||
s3_region_name="us-east-1",
|
||
s3_use_team_prefix=use_team_prefix,
|
||
s3_use_key_prefix=use_key_prefix,
|
||
)
|
||
|
||
payload = StandardLoggingPayload(
|
||
id="abc123",
|
||
metadata={
|
||
"user_api_key_team_alias": team_alias,
|
||
"user_api_key_alias": key_alias,
|
||
},
|
||
messages=[{"role": "user", "content": [{"type": "text", "text": "hi"}]}],
|
||
)
|
||
|
||
result = logger.create_s3_batch_logging_element(datetime.utcnow(), payload)
|
||
assert result is not None
|
||
mock_get_key.assert_called_once()
|
||
|
||
prefix_arg = mock_get_key.call_args.kwargs.get("prefix")
|
||
assert prefix_arg == expected_prefix, (
|
||
f"Expected prefix '{expected_prefix}', got '{prefix_arg}' "
|
||
f"for team={team_alias}, key={key_alias}, "
|
||
f"use_team_prefix={use_team_prefix}, use_key_prefix={use_key_prefix}"
|
||
)
|
||
|
||
|
||
# --------------------------------------------------------------
|
||
# Test prefix priority and concatenation
|
||
# --------------------------------------------------------------
|
||
def test_prefix_priority_and_path_construction():
|
||
"""
|
||
Validate that prefix components are ordered and joined with '/' only once.
|
||
"""
|
||
with patch("litellm.integrations.s3_v2.get_s3_object_key") as mock_get_key:
|
||
mock_get_key.return_value = "mocked/key"
|
||
|
||
logger = S3Logger(s3_use_team_prefix=True, s3_use_key_prefix=True)
|
||
payload = StandardLoggingPayload(
|
||
id="xyz999",
|
||
metadata={
|
||
"user_api_key_team_alias": "Team-Alpha",
|
||
"user_api_key_alias": "API-12345",
|
||
},
|
||
messages=[],
|
||
)
|
||
|
||
logger.create_s3_batch_logging_element(datetime.utcnow(), payload)
|
||
prefix_arg = mock_get_key.call_args.kwargs.get("prefix", "")
|
||
|
||
assert prefix_arg == "Team-Alpha/API-12345/"
|
||
assert "//" not in prefix_arg
|
||
|
||
|
||
# --------------------------------------------------------------
|
||
# Test when prefixes are disabled
|
||
# --------------------------------------------------------------
|
||
def test_prefix_absent_when_flags_disabled():
|
||
"""
|
||
Verify prefix is omitted entirely when prefix flags are False.
|
||
"""
|
||
with patch("litellm.integrations.s3_v2.get_s3_object_key") as mock_get_key:
|
||
mock_get_key.return_value = "mocked/key"
|
||
|
||
logger = S3Logger(s3_use_team_prefix=False, s3_use_key_prefix=False)
|
||
payload = StandardLoggingPayload(
|
||
id="no-prefix",
|
||
metadata={
|
||
"user_api_key_team_alias": "team-x",
|
||
"user_api_key_alias": "key-x",
|
||
},
|
||
messages=[],
|
||
)
|
||
|
||
logger.create_s3_batch_logging_element(datetime.utcnow(), payload)
|
||
prefix_arg = mock_get_key.call_args.kwargs.get("prefix", None)
|
||
assert prefix_arg == "", f"Expected empty prefix, got {prefix_arg}"
|
||
|
||
|
||
# --------------------------------------------------------------
|
||
# Integration-style test (asyncio fixture will patch create_task)
|
||
# --------------------------------------------------------------
|
||
@pytest.mark.asyncio
|
||
async def test_combined_prefix_reflects_in_s3_object_key():
|
||
"""
|
||
Integration-style test ensuring final s3_object_key includes both prefixes correctly.
|
||
"""
|
||
logger = S3Logger(s3_use_team_prefix=True, s3_use_key_prefix=True)
|
||
payload = StandardLoggingPayload(
|
||
id="int-test",
|
||
metadata={
|
||
"user_api_key_team_alias": "myteam",
|
||
"user_api_key_alias": "apikey",
|
||
},
|
||
messages=[],
|
||
)
|
||
|
||
result = logger.create_s3_batch_logging_element(datetime.utcnow(), payload)
|
||
key = result.s3_object_key
|
||
assert "myteam/apikey/" in key, f"Expected both prefixes in key: {key}"
|
||
|
||
|
||
def test_s3_object_key_sanitizes_slashes_in_file_name():
|
||
"""Response ids containing slashes (e.g. bedrock batch job ARNs) must not
|
||
create nested S3 folders; only path/prefix/date slashes are separators."""
|
||
from litellm.integrations.s3 import get_s3_object_key
|
||
|
||
start_time = datetime(2026, 2, 11, 0, 35, 18, 391582)
|
||
file_name = "time-00-35-18-391582_arn:aws:bedrock:us-east-1:123456789012:model-invocation-job/gl18r6skk9yy"
|
||
|
||
key = get_s3_object_key(
|
||
s3_path="LiteLLMAPPLogs",
|
||
prefix="myteam/",
|
||
start_time=start_time,
|
||
s3_file_name=file_name,
|
||
)
|
||
|
||
assert key == (
|
||
"LiteLLMAPPLogs/myteam/2026-02-11/"
|
||
"time-00-35-18-391582_arn:aws:bedrock:us-east-1:123456789012:model-invocation-job_gl18r6skk9yy.json"
|
||
)
|
||
|
||
|
||
def test_create_s3_batch_logging_element_flat_key_for_arn_response_id():
|
||
"""End-to-end through the s3_v2 element builder: an ARN response id must
|
||
yield a flat file directly under the date segment."""
|
||
logger = S3Logger(s3_use_team_prefix=False, s3_use_key_prefix=False)
|
||
payload = StandardLoggingPayload(
|
||
id="arn:aws:bedrock:us-east-1:123456789012:model-invocation-job/gl18r6skk9yy",
|
||
metadata={},
|
||
messages=[],
|
||
)
|
||
|
||
start_time = datetime(2026, 2, 11, 0, 35, 18, 391582)
|
||
result = logger.create_s3_batch_logging_element(start_time, payload)
|
||
|
||
assert result is not None
|
||
date_segment = "2026-02-11/"
|
||
file_segment = result.s3_object_key.split(date_segment, 1)[1]
|
||
assert "/" not in file_segment, f"Expected flat file under date segment, got: {result.s3_object_key}"
|
||
assert file_segment.endswith("model-invocation-job_gl18r6skk9yy.json")
|
||
|
||
|
||
# --------------------------------------------------------------
|
||
# object keys bounded to S3's 1024 UTF-8 byte limit
|
||
# --------------------------------------------------------------
|
||
def _oversized_response_id() -> str:
|
||
return "resp_" + "A" * 1100
|
||
|
||
|
||
def test_s3_object_key_at_the_byte_limit_is_left_alone():
|
||
"""A key that still fits is left byte-identical."""
|
||
from litellm.constants import MAX_S3_OBJECT_KEY_BYTES
|
||
from litellm.integrations.s3 import get_s3_object_key
|
||
|
||
start_time = datetime(2026, 8, 24, 6, 18, 41, 948021)
|
||
fixed_len = len("input/2026-08-24/.json")
|
||
file_name = "x" * (MAX_S3_OBJECT_KEY_BYTES - fixed_len)
|
||
|
||
key = get_s3_object_key(s3_path="input", prefix="", start_time=start_time, s3_file_name=file_name)
|
||
|
||
assert key == f"input/2026-08-24/{file_name}.json"
|
||
assert len(key.encode("utf-8")) == MAX_S3_OBJECT_KEY_BYTES
|
||
|
||
|
||
def test_s3_object_key_is_bounded_for_oversized_response_id():
|
||
"""An oversized Responses API id is shortened to a readable head plus a digest."""
|
||
import hashlib
|
||
|
||
from litellm.constants import MAX_S3_OBJECT_KEY_BYTES
|
||
from litellm.integrations.s3 import get_s3_object_key
|
||
|
||
start_time = datetime(2026, 8, 24, 6, 18, 41, 948021)
|
||
file_name = f"time-06-18-41-948021_{_oversized_response_id()}"
|
||
|
||
key = get_s3_object_key(s3_path="input", prefix="DefaultTeamProd/", start_time=start_time, s3_file_name=file_name)
|
||
|
||
assert len(key.encode("utf-8")) <= MAX_S3_OBJECT_KEY_BYTES
|
||
assert key.startswith("input/DefaultTeamProd/2026-08-24/time-06-18-41-948021_resp_")
|
||
assert key.endswith(f"_{hashlib.sha256(file_name.encode('utf-8')).hexdigest()}.json")
|
||
|
||
|
||
@pytest.mark.parametrize(
|
||
"s3_path,prefix",
|
||
[
|
||
("input", ""),
|
||
("a" * 900, ""),
|
||
("input", "team-" + "b" * 900 + "/"),
|
||
("c" * 600, "team-" + "d" * 600 + "/key-" + "e" * 600 + "/"),
|
||
# many short segments, so the trim lands exactly on the budget edge
|
||
("", "ssss/" * 200),
|
||
],
|
||
)
|
||
def test_s3_object_key_is_bounded_for_long_paths_and_aliases(s3_path: str, prefix: str):
|
||
"""Long paths, team aliases and key aliases stay within the cap."""
|
||
from litellm.constants import MAX_S3_OBJECT_KEY_BYTES
|
||
from litellm.integrations.s3 import get_s3_object_key
|
||
|
||
key = get_s3_object_key(
|
||
s3_path=s3_path,
|
||
prefix=prefix,
|
||
start_time=datetime(2026, 8, 24, 6, 18, 41, 948021),
|
||
s3_file_name=f"time-06-18-41-948021_{_oversized_response_id()}",
|
||
)
|
||
|
||
assert len(key.encode("utf-8")) <= MAX_S3_OBJECT_KEY_BYTES
|
||
assert key.endswith(".json")
|
||
assert "/2026-08-24/" in key or key.startswith("2026-08-24/")
|
||
assert "/" not in key.rsplit("2026-08-24/", 1)[1]
|
||
|
||
|
||
def test_s3_object_key_trimmed_prefixes_stay_distinct_per_operator():
|
||
"""Prefixes that differ only past the trim point keep separate folders."""
|
||
from litellm.constants import MAX_S3_OBJECT_KEY_BYTES
|
||
from litellm.integrations.s3 import get_s3_object_key
|
||
|
||
start_time = datetime(2026, 8, 24, 6, 18, 41, 948021)
|
||
keys = [
|
||
get_s3_object_key(
|
||
s3_path="input",
|
||
prefix="team-" + "b" * 1000 + suffix + "/",
|
||
start_time=start_time,
|
||
s3_file_name=f"time-06-18-41-948021_{_oversized_response_id()}",
|
||
)
|
||
for suffix in ("-one", "-two")
|
||
]
|
||
|
||
assert keys[0] != keys[1]
|
||
assert all(key.startswith("input/team-" + "b" * 900) for key in keys)
|
||
assert all(len(key.encode("utf-8")) == MAX_S3_OBJECT_KEY_BYTES for key in keys)
|
||
|
||
|
||
def test_s3_object_key_bounded_prefix_never_splits_a_multibyte_character():
|
||
"""A multibyte prefix is trimmed on a character boundary."""
|
||
from litellm.constants import MAX_S3_OBJECT_KEY_BYTES
|
||
from litellm.integrations.s3 import get_s3_object_key
|
||
|
||
s3_path = "\u65e5\u672c\u8a9e" * 200
|
||
|
||
key = get_s3_object_key(
|
||
s3_path=s3_path,
|
||
prefix="\u30c1\u30fc\u30e0" * 200 + "/",
|
||
start_time=datetime(2026, 8, 24, 6, 18, 41, 948021),
|
||
s3_file_name=f"time-06-18-41-948021_{_oversized_response_id()}",
|
||
)
|
||
|
||
assert len(key.encode("utf-8")) <= MAX_S3_OBJECT_KEY_BYTES
|
||
assert key.startswith(s3_path[:100])
|
||
assert "\ufffd" not in key
|
||
|
||
|
||
def test_s3_object_key_stays_unique_for_ids_sharing_a_head():
|
||
"""Ids sharing a visible head still get distinct keys."""
|
||
from litellm.integrations.s3 import get_s3_object_key
|
||
|
||
start_time = datetime(2026, 8, 24, 6, 18, 41, 948021)
|
||
keys = {
|
||
get_s3_object_key(
|
||
s3_path="input",
|
||
prefix="",
|
||
start_time=start_time,
|
||
s3_file_name=f"time-06-18-41-948021_{_oversized_response_id()}{suffix}",
|
||
)
|
||
for suffix in ("first", "second", "third")
|
||
}
|
||
|
||
assert len(keys) == 3
|
||
|
||
|
||
def test_s3_object_key_bounding_matches_the_documented_layout():
|
||
"""The bounded key is `<prefix>/<date>/<head>_<sha256>.json`."""
|
||
import hashlib
|
||
|
||
from litellm.integrations.s3 import get_s3_object_key
|
||
|
||
file_name = f"time-06-18-41-948021_{_oversized_response_id()}"
|
||
|
||
key = get_s3_object_key(
|
||
s3_path="input",
|
||
prefix="team/",
|
||
start_time=datetime(2026, 8, 24, 6, 18, 41, 948021),
|
||
s3_file_name=file_name,
|
||
)
|
||
|
||
digest = hashlib.sha256(file_name.encode("utf-8")).hexdigest()
|
||
assert key == f"input/team/2026-08-24/{file_name[:64]}_{digest}.json"
|
||
|
||
|
||
def test_s3_object_key_keeps_the_configured_prefix_when_only_the_id_overflows():
|
||
"""A 940 byte configured prefix survives whole when only the id overflows."""
|
||
from litellm.constants import MAX_S3_OBJECT_KEY_BYTES
|
||
from litellm.integrations.s3 import get_s3_object_key
|
||
|
||
prefix = "team-" + "b" * 934 + "/"
|
||
|
||
key = get_s3_object_key(
|
||
s3_path="",
|
||
prefix=prefix,
|
||
start_time=datetime(2026, 8, 24, 6, 18, 41, 948021),
|
||
s3_file_name=f"time-06-18-41-948021_{_oversized_response_id()}",
|
||
)
|
||
|
||
assert key.startswith(prefix + "2026-08-24/")
|
||
assert len(key.encode("utf-8")) == MAX_S3_OBJECT_KEY_BYTES
|
||
|
||
|
||
def test_s3_object_key_spends_the_whole_budget_when_the_prefix_must_be_trimmed():
|
||
"""A trimmed prefix keeps every byte the budget allows, not whole segments."""
|
||
from litellm.constants import MAX_S3_OBJECT_KEY_BYTES
|
||
from litellm.integrations.s3 import get_s3_object_key
|
||
|
||
s3_path = "p" * 400 + "/" + "q" * 600
|
||
|
||
key = get_s3_object_key(
|
||
s3_path=s3_path,
|
||
prefix="",
|
||
start_time=datetime(2026, 8, 24, 6, 18, 41, 948021),
|
||
s3_file_name="time-06-18-41-948021_abc",
|
||
)
|
||
|
||
assert len(key.encode("utf-8")) == MAX_S3_OBJECT_KEY_BYTES
|
||
assert key.startswith("p" * 400 + "/" + "q" * 500)
|
||
|
||
|
||
def test_s3_object_key_keeps_a_single_segment_path_as_far_as_it_fits():
|
||
"""A path with no separator is kept as far as it fits, never dropped to the bucket root."""
|
||
from litellm.constants import MAX_S3_OBJECT_KEY_BYTES
|
||
from litellm.integrations.s3 import get_s3_object_key
|
||
|
||
key = get_s3_object_key(
|
||
s3_path="a" * 1050,
|
||
prefix="",
|
||
start_time=datetime(2026, 8, 24, 6, 18, 41, 948021),
|
||
s3_file_name="time-06-18-41-948021_chatcmpl-xyz",
|
||
)
|
||
|
||
assert len(key.encode("utf-8")) == MAX_S3_OBJECT_KEY_BYTES
|
||
assert key.startswith("a" * 900)
|
||
|
||
|
||
def test_create_s3_batch_logging_element_bounds_key_and_keeps_full_response_id():
|
||
"""The batch element bounds the key and keeps the full response id in the payload."""
|
||
from litellm.constants import MAX_S3_OBJECT_KEY_BYTES
|
||
|
||
logger = S3Logger(s3_use_team_prefix=True, s3_use_key_prefix=True)
|
||
response_id = _oversized_response_id()
|
||
payload = StandardLoggingPayload(
|
||
id=response_id,
|
||
metadata={"user_api_key_team_alias": "DefaultTeamProd", "user_api_key_alias": "prod-key"},
|
||
messages=[],
|
||
)
|
||
|
||
result = logger.create_s3_batch_logging_element(datetime(2026, 8, 24, 6, 18, 41, 948021), payload)
|
||
|
||
assert result is not None
|
||
assert len(result.s3_object_key.encode("utf-8")) <= MAX_S3_OBJECT_KEY_BYTES
|
||
assert result.s3_object_key.startswith("DefaultTeamProd/prod-key/2026-08-24/")
|
||
assert result.payload["id"] == response_id
|
||
|
||
|
||
def test_s3_object_download_filename_is_bounded_for_oversized_response_id():
|
||
"""The Content-Disposition filename is bounded too, or the PUT fails with MetadataTooLarge."""
|
||
from litellm.constants import MAX_S3_OBJECT_DOWNLOAD_FILENAME_BYTES
|
||
from litellm.integrations.s3 import get_s3_object_download_filename
|
||
|
||
file_name = get_s3_object_download_filename(datetime(2026, 8, 24, 6, 18, 41, 948021), _oversized_response_id())
|
||
|
||
assert len(file_name.encode("utf-8")) <= MAX_S3_OBJECT_DOWNLOAD_FILENAME_BYTES
|
||
assert file_name.startswith("time-2026-08-24T06-18-41-948021_resp_")
|
||
assert file_name.endswith(".json")
|
||
|
||
|
||
def test_s3_object_download_filenames_stay_distinct_when_shortened():
|
||
"""Shortened filenames stay distinct."""
|
||
from litellm.integrations.s3 import get_s3_object_download_filename
|
||
|
||
start_time = datetime(2026, 8, 24, 6, 18, 41, 948021)
|
||
file_names = {
|
||
get_s3_object_download_filename(start_time, _oversized_response_id() + suffix)
|
||
for suffix in ("first", "second", "third")
|
||
}
|
||
|
||
assert len(file_names) == 3
|
||
|
||
|
||
def test_s3_object_download_filename_short_id_is_unchanged():
|
||
"""An ordinary response id keeps the filename it had before."""
|
||
from litellm.integrations.s3 import get_s3_object_download_filename
|
||
|
||
file_name = get_s3_object_download_filename(datetime(2026, 8, 24, 6, 18, 41, 948021), "resp_abc123")
|
||
|
||
assert file_name == "time-2026-08-24T06-18-41-948021_resp_abc123.json"
|
||
|
||
|
||
def test_create_s3_batch_logging_element_bounds_the_download_filename():
|
||
"""The batch element carries a bounded Content-Disposition filename."""
|
||
from litellm.constants import MAX_S3_OBJECT_DOWNLOAD_FILENAME_BYTES
|
||
|
||
logger = S3Logger()
|
||
payload = StandardLoggingPayload(id=_oversized_response_id(), metadata={}, messages=[])
|
||
|
||
result = logger.create_s3_batch_logging_element(datetime(2026, 8, 24, 6, 18, 41, 948021), payload)
|
||
|
||
assert result is not None
|
||
assert len(result.s3_object_download_filename.encode("utf-8")) <= MAX_S3_OBJECT_DOWNLOAD_FILENAME_BYTES
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_audit_log_object_key_is_bounded_for_a_long_configured_path():
|
||
"""Audit log keys are bounded by the same builder."""
|
||
from litellm.constants import MAX_S3_OBJECT_KEY_BYTES
|
||
|
||
logger = S3Logger()
|
||
logger.s3_path = "audit-archive/" + "z" * 1100
|
||
|
||
await logger.async_log_audit_log_event({"id": "1a4f7bd0-6f1e-4d0a-9b3c-9f2e1d5a7c88"})
|
||
|
||
assert len(logger.log_queue) == 1
|
||
assert len(logger.log_queue[0].s3_object_key.encode("utf-8")) <= MAX_S3_OBJECT_KEY_BYTES
|
||
assert logger.log_queue[0].s3_object_key.startswith("audit-archive/" + "z" * 900)
|
||
|
||
|
||
def test_s3_object_download_filename_drops_characters_that_break_the_header():
|
||
"""A quote or separator in the response id cannot escape the quoted header value."""
|
||
from litellm.integrations.s3 import get_s3_object_download_filename
|
||
|
||
file_name = get_s3_object_download_filename(datetime(2026, 8, 24, 6, 18, 41, 948021), 'resp_a"b/c')
|
||
|
||
assert file_name == "time-2026-08-24T06-18-41-948021_resp_a_b_c.json"
|
||
|
||
|
||
# --------------------------------------------------------------
|
||
# params_source / s3_callback_params_override (audit-log decoupling)
|
||
# --------------------------------------------------------------
|
||
def test_s3_callback_params_override_uses_alternate_dict(monkeypatch):
|
||
"""`s3_callback_params_override` makes the logger read its config from
|
||
the override dict instead of `litellm.s3_callback_params`."""
|
||
import litellm
|
||
|
||
monkeypatch.setattr(litellm, "s3_callback_params", {"s3_bucket_name": "normal-bucket"})
|
||
logger = S3Logger(
|
||
s3_callback_params_override={
|
||
"s3_bucket_name": "audit-bucket",
|
||
"s3_path": "audit-prefix",
|
||
"s3_region_name": "us-west-2",
|
||
}
|
||
)
|
||
assert logger.s3_bucket_name == "audit-bucket"
|
||
assert logger.s3_path == "audit-prefix"
|
||
assert logger.s3_region_name == "us-west-2"
|
||
|
||
|
||
def test_s3_callback_params_override_does_not_mutate_inputs(monkeypatch):
|
||
"""Resolving `os.environ/X` markers must not mutate the override dict
|
||
or `litellm.s3_callback_params`."""
|
||
import litellm
|
||
|
||
monkeypatch.setenv("MY_AUDIT_BUCKET", "resolved-bucket")
|
||
override = {"s3_bucket_name": "os.environ/MY_AUDIT_BUCKET"}
|
||
monkeypatch.setattr(litellm, "s3_callback_params", {"s3_bucket_name": "os.environ/MY_AUDIT_BUCKET"})
|
||
logger = S3Logger(s3_callback_params_override=override)
|
||
assert logger.s3_bucket_name == "resolved-bucket"
|
||
assert override["s3_bucket_name"] == "os.environ/MY_AUDIT_BUCKET"
|
||
assert (
|
||
litellm.s3_callback_params["s3_bucket_name"] == "os.environ/MY_AUDIT_BUCKET"
|
||
)
|
||
|
||
|
||
def test_s3_callback_params_override_none_falls_back_to_global(monkeypatch):
|
||
"""No override → behaves exactly as today (reads `litellm.s3_callback_params`)."""
|
||
import litellm
|
||
|
||
monkeypatch.setattr(litellm, "s3_callback_params", {"s3_bucket_name": "from-global"})
|
||
logger = S3Logger()
|
||
assert logger.s3_bucket_name == "from-global"
|
||
|
||
|
||
def test_s3_callback_params_override_empty_dict_is_opt_in(monkeypatch):
|
||
"""An empty override dict skips the global entirely (env/IAM-only config)."""
|
||
import litellm
|
||
|
||
monkeypatch.setattr(litellm, "s3_callback_params", {"s3_bucket_name": "from-global"})
|
||
logger = S3Logger(s3_callback_params_override={})
|
||
assert logger.s3_bucket_name is None
|
||
|
||
|
||
def _expected_content_md5(payload: dict) -> str:
|
||
import base64
|
||
import hashlib
|
||
|
||
from litellm.litellm_core_utils.safe_json_dumps import safe_dumps
|
||
|
||
json_string = safe_dumps(payload)
|
||
return base64.b64encode(
|
||
hashlib.md5(json_string.encode("utf-8"), usedforsecurity=False).digest()
|
||
).decode()
|
||
|
||
|
||
def _require_non_security_md5(monkeypatch):
|
||
import hashlib
|
||
|
||
original_md5 = hashlib.md5
|
||
|
||
def fips_md5(data=b"", *, usedforsecurity=True):
|
||
if usedforsecurity:
|
||
raise ValueError("MD5 blocked for security use")
|
||
return original_md5(data, usedforsecurity=usedforsecurity)
|
||
|
||
monkeypatch.setattr(hashlib, "md5", fips_md5)
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_async_upload_sets_content_md5_header(monkeypatch):
|
||
"""
|
||
Object Lock buckets reject PUTs without a Content-MD5 header (AWS spec).
|
||
The async upload must send a base64 md5 of the exact signed body.
|
||
"""
|
||
from unittest.mock import AsyncMock, MagicMock
|
||
|
||
from litellm.types.integrations.s3_v2 import s3BatchLoggingElement
|
||
|
||
logger = S3Logger(
|
||
s3_bucket_name="test-bucket",
|
||
s3_aws_access_key_id="test-key",
|
||
s3_aws_secret_access_key="test-secret",
|
||
s3_region_name="us-east-1",
|
||
)
|
||
|
||
payload = {"test": "content-md5"}
|
||
test_element = s3BatchLoggingElement(
|
||
s3_object_key="2025-09-14/test-md5.json",
|
||
payload=payload,
|
||
s3_object_download_filename="test-md5.json",
|
||
)
|
||
_require_non_security_md5(monkeypatch)
|
||
|
||
response = MagicMock()
|
||
response.status_code = 200
|
||
response.raise_for_status = MagicMock()
|
||
logger.async_httpx_client = AsyncMock()
|
||
logger.async_httpx_client.put.return_value = response
|
||
|
||
await logger.async_upload_data_to_s3(test_element)
|
||
|
||
headers = logger.async_httpx_client.put.call_args.kwargs["headers"]
|
||
assert headers["Content-MD5"] == _expected_content_md5(payload)
|
||
assert "x-amz-server-side-encryption" not in headers
|
||
|
||
|
||
def test_sync_upload_sets_content_md5_header(monkeypatch):
|
||
"""The sync upload path must also send Content-MD5 for Object Lock buckets."""
|
||
from unittest.mock import MagicMock
|
||
|
||
from litellm.types.integrations.s3_v2 import s3BatchLoggingElement
|
||
|
||
logger = S3Logger(
|
||
s3_bucket_name="test-bucket",
|
||
s3_aws_access_key_id="test-key",
|
||
s3_aws_secret_access_key="test-secret",
|
||
s3_region_name="us-east-1",
|
||
)
|
||
|
||
payload = {"test": "sync-content-md5"}
|
||
test_element = s3BatchLoggingElement(
|
||
s3_object_key="2025-09-14/test-sync-md5.json",
|
||
payload=payload,
|
||
s3_object_download_filename="test-sync-md5.json",
|
||
)
|
||
_require_non_security_md5(monkeypatch)
|
||
|
||
response = MagicMock()
|
||
response.status_code = 200
|
||
response.raise_for_status = MagicMock()
|
||
mock_sync_client = MagicMock()
|
||
mock_sync_client.put.return_value = response
|
||
|
||
with patch(
|
||
"litellm.integrations.s3_v2._get_httpx_client",
|
||
return_value=mock_sync_client,
|
||
):
|
||
logger.upload_data_to_s3(test_element)
|
||
|
||
headers = mock_sync_client.put.call_args.kwargs["headers"]
|
||
assert headers["Content-MD5"] == _expected_content_md5(payload)
|
||
assert "x-amz-server-side-encryption" not in headers
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_async_upload_sets_server_side_encryption_header_when_configured():
|
||
"""
|
||
When s3_server_side_encryption is set (e.g. buckets with a KMS default
|
||
encryption policy), the PUT must carry x-amz-server-side-encryption.
|
||
"""
|
||
from unittest.mock import AsyncMock, MagicMock
|
||
|
||
from litellm.types.integrations.s3_v2 import s3BatchLoggingElement
|
||
|
||
logger = S3Logger(
|
||
s3_bucket_name="test-bucket",
|
||
s3_aws_access_key_id="test-key",
|
||
s3_aws_secret_access_key="test-secret",
|
||
s3_region_name="us-east-1",
|
||
s3_server_side_encryption="aws:kms",
|
||
)
|
||
|
||
test_element = s3BatchLoggingElement(
|
||
s3_object_key="2025-09-14/test-sse.json",
|
||
payload={"test": "sse"},
|
||
s3_object_download_filename="test-sse.json",
|
||
)
|
||
|
||
response = MagicMock()
|
||
response.status_code = 200
|
||
response.raise_for_status = MagicMock()
|
||
logger.async_httpx_client = AsyncMock()
|
||
logger.async_httpx_client.put.return_value = response
|
||
|
||
await logger.async_upload_data_to_s3(test_element)
|
||
|
||
headers = logger.async_httpx_client.put.call_args.kwargs["headers"]
|
||
assert headers["x-amz-server-side-encryption"] == "aws:kms"
|
||
|
||
|
||
def test_s3_server_side_encryption_read_from_callback_params(monkeypatch):
|
||
"""s3_server_side_encryption can be configured via s3_callback_params."""
|
||
import litellm
|
||
|
||
monkeypatch.setattr(
|
||
litellm,
|
||
"s3_callback_params",
|
||
{
|
||
"s3_bucket_name": "from-global",
|
||
"s3_server_side_encryption": "aws:kms",
|
||
},
|
||
)
|
||
logger = S3Logger()
|
||
assert logger.s3_server_side_encryption == "aws:kms"
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_async_upload_sets_sse_kms_key_id_header_when_configured():
|
||
"""
|
||
When s3_sse_kms_key_id is set alongside aws:kms, the PUT must carry
|
||
x-amz-server-side-encryption-aws-kms-key-id so objects are encrypted
|
||
with the customer-managed KMS key instead of the bucket default.
|
||
"""
|
||
from unittest.mock import AsyncMock, MagicMock
|
||
|
||
from litellm.types.integrations.s3_v2 import s3BatchLoggingElement
|
||
|
||
logger = S3Logger(
|
||
s3_bucket_name="test-bucket",
|
||
s3_aws_access_key_id="test-key",
|
||
s3_aws_secret_access_key="test-secret",
|
||
s3_region_name="us-east-1",
|
||
s3_server_side_encryption="aws:kms",
|
||
s3_sse_kms_key_id="arn:aws:kms:us-east-1:111122223333:key/test-key-id",
|
||
)
|
||
|
||
test_element = s3BatchLoggingElement(
|
||
s3_object_key="2025-09-14/test-sse-kms.json",
|
||
payload={"test": "sse-kms"},
|
||
s3_object_download_filename="test-sse-kms.json",
|
||
)
|
||
|
||
response = MagicMock()
|
||
response.status_code = 200
|
||
response.raise_for_status = MagicMock()
|
||
logger.async_httpx_client = AsyncMock()
|
||
logger.async_httpx_client.put.return_value = response
|
||
|
||
await logger.async_upload_data_to_s3(test_element)
|
||
|
||
headers = logger.async_httpx_client.put.call_args.kwargs["headers"]
|
||
assert headers["x-amz-server-side-encryption"] == "aws:kms"
|
||
assert headers["x-amz-server-side-encryption-aws-kms-key-id"] == (
|
||
"arn:aws:kms:us-east-1:111122223333:key/test-key-id"
|
||
)
|
||
|
||
|
||
def test_sync_upload_sets_sse_kms_key_id_header_when_configured():
|
||
"""The sync upload path must carry the same SSE-KMS headers."""
|
||
from unittest.mock import MagicMock
|
||
|
||
from litellm.types.integrations.s3_v2 import s3BatchLoggingElement
|
||
|
||
logger = S3Logger(
|
||
s3_bucket_name="test-bucket",
|
||
s3_aws_access_key_id="test-key",
|
||
s3_aws_secret_access_key="test-secret",
|
||
s3_region_name="us-east-1",
|
||
s3_server_side_encryption="aws:kms",
|
||
s3_sse_kms_key_id="arn:aws:kms:us-east-1:111122223333:key/test-key-id",
|
||
)
|
||
|
||
test_element = s3BatchLoggingElement(
|
||
s3_object_key="2025-09-14/test-sync-sse-kms.json",
|
||
payload={"test": "sync-sse-kms"},
|
||
s3_object_download_filename="test-sync-sse-kms.json",
|
||
)
|
||
|
||
response = MagicMock()
|
||
response.status_code = 200
|
||
response.raise_for_status = MagicMock()
|
||
mock_sync_client = MagicMock()
|
||
mock_sync_client.put.return_value = response
|
||
|
||
with patch(
|
||
"litellm.integrations.s3_v2._get_httpx_client",
|
||
return_value=mock_sync_client,
|
||
):
|
||
logger.upload_data_to_s3(test_element)
|
||
|
||
headers = mock_sync_client.put.call_args.kwargs["headers"]
|
||
assert headers["x-amz-server-side-encryption"] == "aws:kms"
|
||
assert headers["x-amz-server-side-encryption-aws-kms-key-id"] == (
|
||
"arn:aws:kms:us-east-1:111122223333:key/test-key-id"
|
||
)
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_async_upload_omits_kms_key_id_header_when_not_configured():
|
||
"""SSE without a key id must not emit the KMS key id header."""
|
||
from unittest.mock import AsyncMock, MagicMock
|
||
|
||
from litellm.types.integrations.s3_v2 import s3BatchLoggingElement
|
||
|
||
logger = S3Logger(
|
||
s3_bucket_name="test-bucket",
|
||
s3_aws_access_key_id="test-key",
|
||
s3_aws_secret_access_key="test-secret",
|
||
s3_region_name="us-east-1",
|
||
s3_server_side_encryption="AES256",
|
||
)
|
||
|
||
test_element = s3BatchLoggingElement(
|
||
s3_object_key="2025-09-14/test-aes256.json",
|
||
payload={"test": "aes256"},
|
||
s3_object_download_filename="test-aes256.json",
|
||
)
|
||
|
||
response = MagicMock()
|
||
response.status_code = 200
|
||
response.raise_for_status = MagicMock()
|
||
logger.async_httpx_client = AsyncMock()
|
||
logger.async_httpx_client.put.return_value = response
|
||
|
||
await logger.async_upload_data_to_s3(test_element)
|
||
|
||
headers = logger.async_httpx_client.put.call_args.kwargs["headers"]
|
||
assert headers["x-amz-server-side-encryption"] == "AES256"
|
||
assert "x-amz-server-side-encryption-aws-kms-key-id" not in headers
|
||
|
||
|
||
def test_s3_sse_kms_key_id_read_from_callback_params(monkeypatch):
|
||
"""s3_sse_kms_key_id can be configured via s3_callback_params."""
|
||
import litellm
|
||
|
||
monkeypatch.setattr(
|
||
litellm,
|
||
"s3_callback_params",
|
||
{
|
||
"s3_bucket_name": "from-global",
|
||
"s3_server_side_encryption": "aws:kms",
|
||
"s3_sse_kms_key_id": "arn:aws:kms:us-east-1:111122223333:key/test-key-id",
|
||
},
|
||
)
|
||
logger = S3Logger()
|
||
assert logger.s3_sse_kms_key_id == ("arn:aws:kms:us-east-1:111122223333:key/test-key-id")
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_async_upload_infers_aws_kms_when_only_key_id_set():
|
||
"""
|
||
Setting only s3_sse_kms_key_id must not produce an invalid request
|
||
(S3 rejects a key id without an algorithm); aws:kms is inferred.
|
||
"""
|
||
from unittest.mock import AsyncMock, MagicMock
|
||
|
||
from litellm.types.integrations.s3_v2 import s3BatchLoggingElement
|
||
|
||
logger = S3Logger(
|
||
s3_bucket_name="test-bucket",
|
||
s3_aws_access_key_id="test-key",
|
||
s3_aws_secret_access_key="test-secret",
|
||
s3_region_name="us-east-1",
|
||
s3_sse_kms_key_id="arn:aws:kms:us-east-1:111122223333:key/test-key-id",
|
||
)
|
||
|
||
test_element = s3BatchLoggingElement(
|
||
s3_object_key="2025-09-14/test-kms-only.json",
|
||
payload={"test": "kms-only"},
|
||
s3_object_download_filename="test-kms-only.json",
|
||
)
|
||
|
||
response = MagicMock()
|
||
response.status_code = 200
|
||
response.raise_for_status = MagicMock()
|
||
logger.async_httpx_client = AsyncMock()
|
||
logger.async_httpx_client.put.return_value = response
|
||
|
||
await logger.async_upload_data_to_s3(test_element)
|
||
|
||
headers = logger.async_httpx_client.put.call_args.kwargs["headers"]
|
||
assert headers["x-amz-server-side-encryption"] == "aws:kms"
|
||
assert headers["x-amz-server-side-encryption-aws-kms-key-id"] == (
|
||
"arn:aws:kms:us-east-1:111122223333:key/test-key-id"
|
||
)
|
||
|
||
|
||
def test_s3_sse_kms_key_id_read_from_audit_override_params(monkeypatch):
|
||
"""The audit-log override path must honor s3_sse_kms_key_id too."""
|
||
import litellm
|
||
|
||
monkeypatch.setattr(litellm, "s3_callback_params", {"s3_bucket_name": "normal-logs-bucket"})
|
||
logger = S3Logger(
|
||
s3_callback_params_override={
|
||
"s3_bucket_name": "audit-logs-bucket",
|
||
"s3_sse_kms_key_id": "arn:aws:kms:us-east-1:111122223333:key/audit-key-id",
|
||
}
|
||
)
|
||
assert logger.s3_bucket_name == "audit-logs-bucket"
|
||
assert logger.s3_sse_kms_key_id == ("arn:aws:kms:us-east-1:111122223333:key/audit-key-id")
|
||
|
||
|
||
def test_kms_key_id_dropped_when_algorithm_is_not_kms(monkeypatch):
|
||
"""
|
||
AES256 plus a KMS key id is an invalid S3 combination; the key id must be
|
||
dropped at init so uploads keep working instead of silently 400ing.
|
||
"""
|
||
import litellm
|
||
|
||
monkeypatch.setattr(
|
||
litellm,
|
||
"s3_callback_params",
|
||
{
|
||
"s3_bucket_name": "from-global",
|
||
"s3_server_side_encryption": "AES256",
|
||
"s3_sse_kms_key_id": "arn:aws:kms:us-east-1:111122223333:key/test-key-id",
|
||
},
|
||
)
|
||
logger = S3Logger()
|
||
assert logger.s3_server_side_encryption == "AES256"
|
||
assert logger.s3_sse_kms_key_id is None
|
||
|
||
|
||
def test_non_string_algorithm_is_dropped_and_valid_key_id_is_rescued(monkeypatch):
|
||
"""
|
||
A YAML boolean in s3_server_side_encryption must not crash logger init and
|
||
must not discard the valid key id; aws:kms is inferred from the key id.
|
||
"""
|
||
import litellm
|
||
|
||
monkeypatch.setattr(
|
||
litellm,
|
||
"s3_callback_params",
|
||
{
|
||
"s3_bucket_name": "from-global",
|
||
"s3_server_side_encryption": True,
|
||
"s3_sse_kms_key_id": "arn:aws:kms:us-east-1:111122223333:key/test-key-id",
|
||
},
|
||
)
|
||
logger = S3Logger()
|
||
assert logger.s3_server_side_encryption == "aws:kms"
|
||
assert logger.s3_sse_kms_key_id == ("arn:aws:kms:us-east-1:111122223333:key/test-key-id")
|
||
|
||
|
||
def test_non_string_key_id_is_dropped_and_valid_algorithm_is_kept(monkeypatch):
|
||
"""A mistyped key id (unquoted YAML number) must not disable the valid algorithm."""
|
||
import litellm
|
||
|
||
monkeypatch.setattr(
|
||
litellm,
|
||
"s3_callback_params",
|
||
{
|
||
"s3_bucket_name": "from-global",
|
||
"s3_server_side_encryption": "aws:kms",
|
||
"s3_sse_kms_key_id": 12345,
|
||
},
|
||
)
|
||
logger = S3Logger()
|
||
assert logger.s3_server_side_encryption == "aws:kms"
|
||
assert logger.s3_sse_kms_key_id is None
|
||
|
||
|
||
_ACCESS_KEY = "AKIAIOSFODNN7EXAMPLE"
|
||
_SECRET_KEY = "wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY"
|
||
_KEY_WITH_SPACE = "LOGS/LLM AI Projects/2026-08-04/time-13-01-00-abc.json"
|
||
|
||
|
||
def _signature_for(signer_cls, url: str, method: str, body: bytes | None, headers: dict[str, str]) -> str:
|
||
from botocore.awsrequest import AWSRequest
|
||
from botocore.credentials import Credentials
|
||
|
||
sent = {name.lower(): value for name, value in headers.items()}
|
||
signed_header_names = sent["authorization"].split("SignedHeaders=")[1].split(", ")[0].split(";")
|
||
request = AWSRequest(
|
||
method=method,
|
||
url=url,
|
||
data=body,
|
||
headers={name: sent[name] for name in signed_header_names if name in sent},
|
||
)
|
||
request.context["timestamp"] = sent["x-amz-date"]
|
||
signer = signer_cls(Credentials(_ACCESS_KEY, _SECRET_KEY), "s3", "us-east-1")
|
||
canonical_request = signer.canonical_request(request)
|
||
return signer.signature(signer.string_to_sign(request, canonical_request), request)
|
||
|
||
|
||
def _as_s3_canonicalizes(url: str) -> str:
|
||
"""
|
||
The path S3 rebuilds from the wire path: percent-encode everything outside the unreserved
|
||
set, without normalizing or double-encoding. `=` becomes `%3D`, `%20` stays `%20`.
|
||
"""
|
||
from urllib.parse import quote, unquote, urlsplit, urlunsplit
|
||
|
||
split = urlsplit(url)
|
||
return urlunsplit(split._replace(path=quote(unquote(split.path), safe="/~")))
|
||
|
||
|
||
def _assert_signed_for_s3_canonicalization(url: str, method: str, body: bytes | None, headers: dict[str, str]) -> None:
|
||
"""
|
||
S3 rebuilds the canonical request from the wire path with single percent-encoding, which
|
||
botocore models as S3SigV4Auth; plain SigV4Auth double-encodes it (%2520 for a space) and S3
|
||
answers 403 SignatureDoesNotMatch. Assert we sent an already-encoded path and signed it the
|
||
way S3 reads it.
|
||
"""
|
||
from botocore.auth import S3SigV4Auth, SigV4Auth
|
||
|
||
assert url == _as_s3_canonicalizes(url)
|
||
sent_signature = headers["Authorization"].split("Signature=")[1].strip()
|
||
assert sent_signature == _signature_for(S3SigV4Auth, url, method, body, headers)
|
||
assert sent_signature != _signature_for(SigV4Auth, url, method, body, headers)
|
||
|
||
|
||
def _logger_for_signing() -> S3Logger:
|
||
return S3Logger(
|
||
s3_bucket_name="logs-bucket",
|
||
s3_aws_access_key_id=_ACCESS_KEY,
|
||
s3_aws_secret_access_key=_SECRET_KEY,
|
||
s3_region_name="us-east-1",
|
||
)
|
||
|
||
|
||
def _element_with_space():
|
||
from litellm.types.integrations.s3_v2 import s3BatchLoggingElement
|
||
|
||
return s3BatchLoggingElement(
|
||
s3_object_key=_KEY_WITH_SPACE,
|
||
payload={"test": "sigv4"},
|
||
s3_object_download_filename="log.json",
|
||
)
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_async_upload_signs_object_key_with_space_the_way_s3_does():
|
||
from unittest.mock import AsyncMock, MagicMock
|
||
|
||
logger = _logger_for_signing()
|
||
response = MagicMock()
|
||
response.status_code = 200
|
||
response.raise_for_status = MagicMock()
|
||
logger.async_httpx_client = AsyncMock()
|
||
logger.async_httpx_client.put.return_value = response
|
||
|
||
await logger.async_upload_data_to_s3(_element_with_space())
|
||
|
||
call = logger.async_httpx_client.put.call_args
|
||
_assert_signed_for_s3_canonicalization(
|
||
url=call[0][0],
|
||
method="PUT",
|
||
body=call.kwargs["data"].encode("utf-8"),
|
||
headers=call.kwargs["headers"],
|
||
)
|
||
|
||
|
||
def test_sync_upload_signs_object_key_with_space_the_way_s3_does():
|
||
from unittest.mock import MagicMock
|
||
|
||
logger = _logger_for_signing()
|
||
response = MagicMock()
|
||
response.status_code = 200
|
||
response.raise_for_status = MagicMock()
|
||
mock_sync_client = MagicMock()
|
||
mock_sync_client.put.return_value = response
|
||
|
||
with patch("litellm.integrations.s3_v2._get_httpx_client", return_value=mock_sync_client):
|
||
logger.upload_data_to_s3(_element_with_space())
|
||
|
||
call = mock_sync_client.put.call_args
|
||
_assert_signed_for_s3_canonicalization(
|
||
url=call[0][0],
|
||
method="PUT",
|
||
body=call.kwargs["data"].encode("utf-8"),
|
||
headers=call.kwargs["headers"],
|
||
)
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_download_signs_object_key_with_space_the_way_s3_does():
|
||
from unittest.mock import AsyncMock, MagicMock
|
||
|
||
logger = _logger_for_signing()
|
||
response = MagicMock()
|
||
response.status_code = 200
|
||
response.json = MagicMock(return_value={"downloaded": "data"})
|
||
logger.async_httpx_client = AsyncMock()
|
||
logger.async_httpx_client.get.return_value = response
|
||
|
||
assert await logger._download_object_from_s3(_KEY_WITH_SPACE) == {"downloaded": "data"}
|
||
|
||
call = logger.async_httpx_client.get.call_args
|
||
_assert_signed_for_s3_canonicalization(
|
||
url=call[0][0],
|
||
method="GET",
|
||
body=None,
|
||
headers=call.kwargs["headers"],
|
||
)
|
||
|
||
_RESERVED_CHAR_KEYS = (
|
||
"2026-08-21/time-05-29-36_resp_bGl0ZWxsbTpjdXN0b20=.json",
|
||
"session=logs/2026-08-21/time-05-29-36_abc.json",
|
||
"a+b/2026-08-21/time-05-29-36_abc.json",
|
||
"a&b/2026-08-21/time-05-29-36_abc.json",
|
||
"a#b/2026-08-21/time-05-29-36_abc.json",
|
||
"a?b/2026-08-21/time-05-29-36_abc.json",
|
||
"a%b/2026-08-21/time-05-29-36_abc.json",
|
||
_KEY_WITH_SPACE,
|
||
)
|
||
|
||
|
||
def _element_for(s3_object_key: str):
|
||
from litellm.types.integrations.s3_v2 import s3BatchLoggingElement
|
||
|
||
return s3BatchLoggingElement(
|
||
s3_object_key=s3_object_key,
|
||
payload={"test": "sigv4"},
|
||
s3_object_download_filename="log.json",
|
||
)
|
||
|
||
|
||
def _expected_wire_url(s3_object_key: str) -> str:
|
||
"""The URL boto3 itself would put on the wire for this key."""
|
||
from urllib.parse import quote
|
||
|
||
return f"https://logs-bucket.s3.us-east-1.amazonaws.com/{quote(s3_object_key, safe='/')}"
|
||
|
||
|
||
@pytest.mark.parametrize("s3_object_key", _RESERVED_CHAR_KEYS)
|
||
@pytest.mark.asyncio
|
||
async def test_async_upload_percent_encodes_reserved_characters_in_object_key(s3_object_key):
|
||
from unittest.mock import AsyncMock, MagicMock
|
||
|
||
logger = _logger_for_signing()
|
||
response = MagicMock()
|
||
response.status_code = 200
|
||
response.raise_for_status = MagicMock()
|
||
logger.async_httpx_client = AsyncMock()
|
||
logger.async_httpx_client.put.return_value = response
|
||
|
||
await logger.async_upload_data_to_s3(_element_for(s3_object_key))
|
||
|
||
call = logger.async_httpx_client.put.call_args
|
||
assert call[0][0] == _expected_wire_url(s3_object_key)
|
||
_assert_signed_for_s3_canonicalization(
|
||
url=call[0][0],
|
||
method="PUT",
|
||
body=call.kwargs["data"].encode("utf-8"),
|
||
headers=call.kwargs["headers"],
|
||
)
|
||
|
||
|
||
@pytest.mark.parametrize("s3_object_key", _RESERVED_CHAR_KEYS)
|
||
def test_sync_upload_percent_encodes_reserved_characters_in_object_key(s3_object_key):
|
||
from unittest.mock import MagicMock
|
||
|
||
logger = _logger_for_signing()
|
||
response = MagicMock()
|
||
response.status_code = 200
|
||
response.raise_for_status = MagicMock()
|
||
mock_sync_client = MagicMock()
|
||
mock_sync_client.put.return_value = response
|
||
|
||
with patch("litellm.integrations.s3_v2._get_httpx_client", return_value=mock_sync_client):
|
||
logger.upload_data_to_s3(_element_for(s3_object_key))
|
||
|
||
call = mock_sync_client.put.call_args
|
||
assert call[0][0] == _expected_wire_url(s3_object_key)
|
||
_assert_signed_for_s3_canonicalization(
|
||
url=call[0][0],
|
||
method="PUT",
|
||
body=call.kwargs["data"].encode("utf-8"),
|
||
headers=call.kwargs["headers"],
|
||
)
|
||
|
||
|
||
@pytest.mark.parametrize("s3_object_key", _RESERVED_CHAR_KEYS)
|
||
@pytest.mark.asyncio
|
||
async def test_download_percent_encodes_reserved_characters_in_object_key(s3_object_key):
|
||
from unittest.mock import AsyncMock, MagicMock
|
||
|
||
logger = _logger_for_signing()
|
||
response = MagicMock()
|
||
response.status_code = 200
|
||
response.json = MagicMock(return_value={"downloaded": "data"})
|
||
logger.async_httpx_client = AsyncMock()
|
||
logger.async_httpx_client.get.return_value = response
|
||
|
||
assert await logger._download_object_from_s3(s3_object_key) == {"downloaded": "data"}
|
||
|
||
call = logger.async_httpx_client.get.call_args
|
||
assert call[0][0] == _expected_wire_url(s3_object_key)
|
||
_assert_signed_for_s3_canonicalization(
|
||
url=call[0][0],
|
||
method="GET",
|
||
body=None,
|
||
headers=call.kwargs["headers"],
|
||
)
|
||
|
||
|
||
def _s3_logger_for_region(region_name: str) -> S3Logger:
|
||
logger = S3Logger.__new__(S3Logger)
|
||
logger.s3_endpoint_url = None
|
||
logger.s3_bucket_name = "my-litellm-audit"
|
||
logger.s3_region_name = region_name
|
||
return logger
|
||
|
||
|
||
@pytest.mark.parametrize(
|
||
"region_name,expected_url",
|
||
[
|
||
(
|
||
"cn-northwest-1",
|
||
"https://my-litellm-audit.s3.cn-northwest-1.amazonaws.com.cn/2025-01-01/key.json",
|
||
),
|
||
(
|
||
"us-gov-west-1",
|
||
"https://my-litellm-audit.s3.us-gov-west-1.amazonaws.com/2025-01-01/key.json",
|
||
),
|
||
(
|
||
"us-east-1",
|
||
"https://my-litellm-audit.s3.us-east-1.amazonaws.com/2025-01-01/key.json",
|
||
),
|
||
],
|
||
)
|
||
def test_build_object_url_uses_partition_dns_suffix(region_name: str, expected_url: str) -> None:
|
||
assert _s3_logger_for_region(region_name)._build_object_url("2025-01-01/key.json") == expected_url
|