scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-09-15 23:31:29 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 213

test(redis): cover ElastiCache IAM failures

Browse source 
Generated with AI

Co-Authored-By: Claude Code
This commit is contained in:
eugene-yao-zocdoc 2026-08-26 19:48:03 -04:00
parent f66891d80f
commit bb043fe29a
2 changed files with 93 additions and 29 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

52
tests/test_litellm/test_redis.py
View file

@ -455,37 +455,33 @@ def test_explicit_provider_wins_over_aws_iam(clean_redis_environment):
def test_gcp_wins_over_aws_iam(clean_redis_environment):
with patch("litellm._redis.create_gcp_iam_redis_connect_func") as mock_gcp:
mock_gcp.return_value = _gcp_marker_callback()
redis_kwargs = _get_redis_client_logic(
host="cache.example.com",
port=6379,
gcp_service_account="sa@example.com",
aws_iam_auth=True,
aws_iam_user_name="iam-user",
aws_iam_cache_name="cache.example.com",
aws_iam_region="us-east-1",
)
redis_kwargs = _get_redis_client_logic(
host="cache.example.com",
port=6379,
gcp_service_account="sa@example.com",
aws_iam_auth=True,
aws_iam_user_name="iam-user",
aws_iam_cache_name="cache.example.com",
aws_iam_region="us-east-1",
)
assert "credential_provider" not in redis_kwargs
assert redis_kwargs["redis_connect_func"] is mock_gcp.return_value
assert redis_kwargs["redis_connect_func"]._gcp_service_account == "sa@example.com"
def test_azure_wins_over_aws_iam(clean_redis_environment):
with patch("litellm._redis.create_azure_ad_redis_connect_func") as mock_azure:
mock_azure.return_value = MagicMock()
redis_kwargs = _get_redis_client_logic(
host="cache.example.com",
port=6379,
azure_redis_ad_token="true",
aws_iam_auth=True,
aws_iam_user_name="iam-user",
aws_iam_cache_name="cache.example.com",
aws_iam_region="us-east-1",
)
redis_kwargs = _get_redis_client_logic(
host="cache.example.com",
port=6379,
azure_redis_ad_token="true",
aws_iam_auth=True,
aws_iam_user_name="iam-user",
aws_iam_cache_name="cache.example.com",
aws_iam_region="us-east-1",
)
assert "credential_provider" not in redis_kwargs
assert redis_kwargs["redis_connect_func"] is mock_azure.return_value
assert redis_kwargs["redis_connect_func"]._azure_redis_ad_token is True
def test_async_cluster_installs_aws_iam_provider(clean_redis_environment):
@ -1732,11 +1728,9 @@ def test_async_sentinel_keeps_the_credential_provider_off_the_monitors(markers,
"redis_connect_func": SimpleNamespace(**markers),
}
with (
patch("litellm._redis.async_redis.Sentinel") as mock_sentinel_cls,
patch("litellm._redis._get_redis_client_logic", return_value=redis_kwargs),
):
get_redis_async_client()
with patch("litellm._redis.async_redis.Sentinel") as mock_sentinel_cls:
with patch("litellm._redis._get_redis_client_logic", return_value=redis_kwargs):
get_redis_async_client()
sentinel_kwargs = mock_sentinel_cls.call_args[1]["sentinel_kwargs"]
assert sentinel_kwargs["password"] == sentinel_password

70
tests/test_litellm/test_redis_credential_provider.py
View file

@ -1,4 +1,6 @@
import asyncio
import builtins
import sys
from types import SimpleNamespace
from urllib.parse import parse_qs, urlsplit
@ -87,6 +89,41 @@ def test_elasticache_provider_resolves_credentials_once_but_refreshes_signature(
assert rotating_credentials.calls == 3
def test_elasticache_provider_uses_botocore_session_credentials(monkeypatch):
credentials = _FakeCredentials("AKIA-SYNTHETIC")
monkeypatch.setattr("botocore.session.get_session", lambda: SimpleNamespace(get_credentials=lambda: credentials))
provider = ElastiCacheIAMCredentialProvider(
user_name="iam-user",
cache_name="cache.example.com",
region="us-east-1",
)
user_name, token = provider.get_credentials()
assert user_name == "iam-user"
assert "AKIA-SYNTHETIC" in token
def test_elasticache_provider_reports_missing_botocore(monkeypatch):
original_import = builtins.__import__
def import_without_botocore(name, *args, **kwargs):
if name == "botocore.session":
raise ImportError("synthetic missing dependency")
return original_import(name, *args, **kwargs)
monkeypatch.delitem(sys.modules, "botocore.session", raising=False)
monkeypatch.setattr(builtins, "__import__", import_without_botocore)
provider = ElastiCacheIAMCredentialProvider(
user_name="iam-user",
cache_name="cache.example.com",
region="us-east-1",
)
with pytest.raises(ImportError, match="pip install boto3"):
provider.get_credentials()
def test_elasticache_provider_reports_missing_credentials():
provider = ElastiCacheIAMCredentialProvider(
user_name="iam-user",
@ -99,6 +136,39 @@ def test_elasticache_provider_reports_missing_credentials():
provider.get_credentials()
def test_elasticache_provider_reports_missing_frozen_credentials():
provider = ElastiCacheIAMCredentialProvider(
user_name="iam-user",
cache_name="cache.example.com",
region="us-east-1",
credentials_resolver=_FakeResolver(SimpleNamespace(get_frozen_credentials=lambda: None)),
)
with pytest.raises(RuntimeError, match="Unable to resolve AWS credentials"):
provider.get_credentials()
def test_elasticache_provider_reports_missing_signing_dependency(monkeypatch):
original_import = builtins.__import__
def import_without_botocore_auth(name, *args, **kwargs):
if name == "botocore.auth":
raise ImportError("synthetic missing dependency")
return original_import(name, *args, **kwargs)
monkeypatch.delitem(sys.modules, "botocore.auth", raising=False)
monkeypatch.setattr(builtins, "__import__", import_without_botocore_auth)
provider = ElastiCacheIAMCredentialProvider(
user_name="iam-user",
cache_name="cache.example.com",
region="us-east-1",
credentials_resolver=_FakeResolver(_FakeCredentials("AKIA-SYNTHETIC")),
)
with pytest.raises(ImportError, match="pip install boto3"):
provider.get_credentials()
def test_elasticache_provider_recovers_after_a_failed_resolution():
resolver = _FakeResolver(None)
provider = ElastiCacheIAMCredentialProvider(