From bb043fe29a7f680f8a5aee1ba5998f3fdb26e561 Mon Sep 17 00:00:00 2001 From: eugene-yao-zocdoc Date: Wed, 26 Aug 2026 19:48:03 -0400 Subject: [PATCH] test(redis): cover ElastiCache IAM failures Generated with AI Co-Authored-By: Claude Code --- tests/test_litellm/test_redis.py | 52 ++++++-------- .../test_redis_credential_provider.py | 70 +++++++++++++++++++ 2 files changed, 93 insertions(+), 29 deletions(-) diff --git a/tests/test_litellm/test_redis.py b/tests/test_litellm/test_redis.py index 21c23ba7866..43fb7563233 100644 --- a/tests/test_litellm/test_redis.py +++ b/tests/test_litellm/test_redis.py @@ -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 diff --git a/tests/test_litellm/test_redis_credential_provider.py b/tests/test_litellm/test_redis_credential_provider.py index de4afb7c313..336e30e12c6 100644 --- a/tests/test_litellm/test_redis_credential_provider.py +++ b/tests/test_litellm/test_redis_credential_provider.py @@ -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(