[Bug Fix] Ensure /redis/info works on GCP Redis (#11732)

* fix cache_redis_info

* mock_redis_client_list_restricted

Author: ishaan-jaff

litellm/proxy/caching_routes.py

@@ -159,6 +159,23 @@ async def cache_delete(request: Request):
         )
 
 
+def _get_redis_client_info(cache_instance) -> tuple[list, int]:
+    """
+    Helper function to safely get Redis client list information.
+
+    Returns:
+        tuple: (client_list, num_clients) where num_clients is -1 if CLIENT LIST is unavailable
+    """
+    try:
+        client_list = cache_instance.client_list()
+        return client_list, len(client_list)
+    except Exception as e:
+        verbose_proxy_logger.warning(
+            f"CLIENT LIST command failed (likely restricted on managed Redis): {str(e)}"
+        )
+        return ["CLIENT LIST command not available on this Redis instance"], -1
+
+
 @router.get(
     "/redis/info",
     dependencies=[Depends(user_api_key_auth)],
@@ -172,22 +189,27 @@ async def cache_redis_info():
             raise HTTPException(
                 status_code=503, detail="Cache not initialized. litellm.cache is None"
             )
-        if litellm.cache.type == "redis" and isinstance(
-            litellm.cache.cache, RedisCache
+
+        if not (
+            litellm.cache.type == "redis"
+            and isinstance(litellm.cache.cache, RedisCache)
         ):
-            client_list = litellm.cache.cache.client_list()
-            redis_info = litellm.cache.cache.info()
-            num_clients = len(client_list)
-            return {
-                "num_clients": num_clients,
-                "clients": client_list,
-                "info": redis_info,
-            }
-        else:
             raise HTTPException(
                 status_code=500,
-                detail=f"Cache type {litellm.cache.type} does not support flushing",
+                detail=f"Cache type {litellm.cache.type} does not support redis info",
             )
+
+        # Get client information (handles CLIENT LIST restrictions gracefully)
+        client_list, num_clients = _get_redis_client_info(litellm.cache.cache)
+
+        # Get Redis server information
+        redis_info = litellm.cache.cache.info()
+
+        return {
+            "num_clients": num_clients,
+            "clients": client_list,
+            "info": redis_info,
+        }
     except Exception as e:
         raise HTTPException(
             status_code=503,

tests/test_litellm/proxy/test_caching_routes.py

@@ -202,3 +202,73 @@ def test_cache_ping_with_redis_version_float(mock_redis_success):
     cache_params = data["health_check_cache_params"]
     assert isinstance(cache_params, dict)
     assert isinstance(cache_params.get("redis_version"), float)
+
+
+@pytest.fixture
+def mock_redis_client_list_restricted(mocker):
+    """Mock Redis cache where CLIENT LIST is restricted (like GCP Redis)"""
+
+    def mock_client_list():
+        raise Exception("ERR unknown command 'CLIENT'")
+
+    def mock_info():
+        return {
+            "redis_version": "6.2.7",
+            "used_memory": "1000000",
+            "connected_clients": "5",
+            "keyspace_hits": "1000",
+            "keyspace_misses": "100",
+        }
+
+    mock_cache = mocker.MagicMock()
+    mock_cache.type = "redis"
+    mock_cache.cache = RedisCache(host="localhost", port=6379, password="hello")
+    mock_cache.cache.client_list = mock_client_list
+    mock_cache.cache.info = mock_info
+
+    mocker.patch.object(litellm, "cache", mock_cache)
+    return mock_cache
+
+
+@pytest.fixture
+def mock_redis_client_list_success(mocker):
+    """Mock Redis cache where CLIENT LIST works normally"""
+
+    def mock_client_list():
+        return [
+            {"id": "1", "addr": "127.0.0.1:54321", "name": "client1"},
+            {"id": "2", "addr": "127.0.0.1:54322", "name": "client2"},
+        ]
+
+    def mock_info():
+        return {
+            "redis_version": "6.2.7",
+            "used_memory": "1000000",
+            "connected_clients": "2",
+        }
+
+    mock_cache = mocker.MagicMock()
+    mock_cache.type = "redis"
+    mock_cache.cache = RedisCache(host="localhost", port=6379, password="hello")
+    mock_cache.cache.client_list = mock_client_list
+    mock_cache.cache.info = mock_info
+
+    mocker.patch.object(litellm, "cache", mock_cache)
+    return mock_cache
+
+
+def test_cache_redis_info_no_cache():
+    """Test /cache/redis/info when no cache is initialized"""
+    original_cache = litellm.cache
+    litellm.cache = None
+
+    response = client.get(
+        "/cache/redis/info", headers={"Authorization": "Bearer sk-1234"}
+    )
+    assert response.status_code == 503
+
+    data = response.json()
+    assert "Cache not initialized" in data["detail"]
+
+    # Restore original cache
+    litellm.cache = original_cache