scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-10-09 03:18:44 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 109

test_list_containers_basic

Browse source 
test_encode_decode_helpers_roundtrip_in_cache_context
test_encrypt_response_id_success
test_s3_object_key_prefix_combinations[True-True-teamA-None-teamA/]
test_s3_object_key_prefix_combinations[True-True-teamA-keyA-teamA/keyA/]
test_s3_v2_endpoint_url
test_list_containers_with_params
This commit is contained in:
Sameer Kankute 2026-01-07 16:28:11 +05:30
parent 0319a4fce2
commit 4b56a1bbad
4 changed files with 86 additions and 85 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

50
tests/test_litellm/containers/test_container_api.py
View file

@ -163,18 +163,19 @@ class TestContainerAPI:
has_more=False
)
with patch('litellm.containers.main.base_llm_http_handler') as mock_handler:
mock_handler.container_list_handler.return_value = mock_response
response = list_containers(
custom_llm_provider="openai"
)
assert isinstance(response, ContainerListResponse) or type(response).__name__ == "ContainerListResponse"
assert len(response.data) == 2
assert response.data[0].id == "cntr_1"
assert response.data[1].id == "cntr_2"
assert response.has_more == False
with patch('litellm.containers.main.ProviderConfigManager'):
with patch('litellm.containers.main.base_llm_http_handler') as mock_handler:
mock_handler.container_list_handler.return_value = mock_response
response = list_containers(
custom_llm_provider="openai"
)
assert isinstance(response, ContainerListResponse) or type(response).__name__ == "ContainerListResponse"
assert len(response.data) == 2
assert response.data[0].id == "cntr_1"
assert response.data[1].id == "cntr_2"
assert response.has_more == False
def test_list_containers_with_params(self):
"""Test container listing with parameters."""
@ -196,18 +197,19 @@ class TestContainerAPI:
has_more=True
)
with patch('litellm.containers.main.base_llm_http_handler') as mock_handler:
mock_handler.container_list_handler.return_value = mock_response
response = list_containers(
limit=1,
order="desc",
after="cntr_prev",
custom_llm_provider="openai"
)
assert len(response.data) == 1
assert response.has_more == True
with patch('litellm.containers.main.ProviderConfigManager'):
with patch('litellm.containers.main.base_llm_http_handler') as mock_handler:
mock_handler.container_list_handler.return_value = mock_response
response = list_containers(
limit=1,
order="desc",
after="cntr_prev",
custom_llm_provider="openai"
)
assert len(response.data) == 1
assert response.has_more == True
@pytest.mark.asyncio
async def test_alist_containers_basic(self):

32
tests/test_litellm/integrations/gitlab/test_gitlab_prompt_manager.py
View file

@ -818,27 +818,29 @@ def test_cache_get_by_file_returns_exact_entry(mock_pm_cls, fake_managers):
assert beta and beta["id"] == "nested/beta"
@patch("litellm.integrations.gitlab.gitlab_client.GitLabClient")
@patch("litellm.integrations.gitlab.gitlab_prompt_manager.GitLabPromptManager")
def test_encode_decode_helpers_roundtrip_in_cache_context(mock_pm_cls, mock_client_cls, fake_managers):
def test_encode_decode_helpers_roundtrip_in_cache_context(fake_managers):
"""Test encode/decode roundtrip in cache context."""
tm, wrapper = fake_managers
tm._discoverable_ids = ["dir1/dir2/item"]
mock_pm_cls.return_value = wrapper
# Mock the GitLabClient to avoid real HTTP requests
mock_client = MagicMock()
mock_client_cls.return_value = mock_client
with patch("litellm.integrations.gitlab.gitlab_prompt_manager.GitLabPromptManager") as mock_pm_cls:
with patch("litellm.integrations.gitlab.gitlab_client.GitLabClient") as mock_client_cls:
mock_pm_cls.return_value = wrapper
mock_client_cls.return_value = mock_client
cache = GitLabPromptCache({"project": "g/s/r", "access_token": "tkn"})
cache.load_all()
cache = GitLabPromptCache({"project": "g/s/r", "access_token": "tkn"})
cache.load_all()
encoded = encode_prompt_id("dir1/dir2/item")
assert encoded in cache.list_ids()
encoded = encode_prompt_id("dir1/dir2/item")
assert encoded in cache.list_ids()
# decode → encode → lookup should still work
decoded = decode_prompt_id(encoded)
assert decoded == "dir1/dir2/item"
# decode → encode → lookup should still work
decoded = decode_prompt_id(encoded)
assert decoded == "dir1/dir2/item"
got = cache.get_by_id(decoded)
assert got is not None
assert got["id"] == "dir1/dir2/item"
got = cache.get_by_id(decoded)
assert got is not None
assert got["id"] == "dir1/dir2/item"

60
tests/test_litellm/integrations/test_s3_v2.py
View file

@ -25,18 +25,12 @@ class TestS3V2UnitTests:
"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):
def test_s3_v2_endpoint_url(self):
"""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
@ -512,35 +506,37 @@ def test_s3_object_key_prefix_combinations(
"""
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"
with patch("asyncio.create_task"):
with patch("litellm.integrations.s3_v2.CustomBatchLogger.periodic_flush"):
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,
)
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"}]}],
)
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()
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}"
)
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}"
)
# --------------------------------------------------------------

29
tests/test_litellm/test_responses_id_security.py
View file

@ -111,21 +111,22 @@ class TestEncryptResponseId:
id="resp_123", created_at=1234567890, output=[], status="completed"
)
with patch(
"litellm.proxy.hooks.responses_id_security.encrypt_value_helper"
) as mock_encrypt:
mock_encrypt.return_value = "encrypted_base64_value"
with patch.object(
responses_id_security, "_get_signing_key", return_value="test-key"
):
result = responses_id_security._encrypt_response_id(
mock_response, mock_user_api_key_dict
)
with patch("litellm.proxy.proxy_server.general_settings", {}):
with patch(
"litellm.proxy.hooks.responses_id_security.encrypt_value_helper"
) as mock_encrypt:
mock_encrypt.return_value = "encrypted_base64_value"
with patch.object(
responses_id_security, "_get_signing_key", return_value="test-key"
):
result = responses_id_security._encrypt_response_id(
mock_response, mock_user_api_key_dict
)
assert result.id == "resp_encrypted_base64_value"
assert result.id.startswith("resp_")
mock_encrypt.assert_called_once()
assert result.id == "resp_encrypted_base64_value"
assert result.id.startswith("resp_")
mock_encrypt.assert_called_once()
def test_encrypt_response_id_maintains_prefix(
self, responses_id_security, mock_user_api_key_dict