Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
22 changes: 22 additions & 0 deletions health_check/contrib/redis.py
Original file line number Diff line number Diff line change
Expand Up @@ -53,6 +53,28 @@ class Redis(HealthCheck):
dataclasses.field(repr=False, default=None)
)

def __repr__(self):
# include client host name and logical database number to identify them
if self.client_factory is not None:
client = self.client_factory()
else:
# Use the deprecated client parameter (user manages lifecycle)
client = self.client

try:
conn_kwargs = client.connection_pool.connection_kwargs
host = conn_kwargs["host"]
db = conn_kwargs["db"]
return f"Redis(client=RedisClient(host={host}, db={db}))"
except (AttributeError, KeyError):
pass

try:
hosts = [node.name for node in client.startup_nodes]
return f"Redis(client=RedisCluster(hosts={hosts!r}))"
except AttributeError:
return super().__repr__()
Comment thread
codingjoe marked this conversation as resolved.
Comment thread
codingjoe marked this conversation as resolved.
Comment on lines +56 to +76
Copy link

Copilot AI Feb 23, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The repr method creates a Redis client from client_factory (line 59) but never closes it, causing a resource leak. The run() method properly closes all factory-created clients (line 127), establishing the pattern that factory-created clients must be closed.

Since repr is synchronous and clients require async cleanup (aclose()), this approach is fundamentally incompatible with proper resource management. The repr method should not create clients at all.

Consider storing connection information (host, db) during initialization or caching it on first access in an instance variable. For the deprecated client parameter, you can safely access connection details since the client lifecycle is managed externally.

Copilot uses AI. Check for mistakes.

Comment thread
codingjoe marked this conversation as resolved.
def __post_init__(self):
# Validate that exactly one of client or client_factory is provided
if self.client is not None and self.client_factory is not None:
Expand Down
122 changes: 121 additions & 1 deletion tests/contrib/test_redis.py
Original file line number Diff line number Diff line change
Expand Up @@ -149,7 +149,127 @@ async def test_redis__validation_neither_param(self):
):
RedisHealthCheck()

@pytest.mark.integration
def test_redis__repr_standard_client(self):
"""Verify repr includes host and db for a standard Redis client."""
from redis.asyncio import Redis as RedisClient

check = RedisHealthCheck(
client_factory=lambda: RedisClient(host="myhost", port=6379, db=2)
)
assert repr(check) == "Redis(client=RedisClient(host=myhost, db=2))"

def test_redis__repr_from_url(self):
"""Verify repr includes host and db when client is created via from_url."""
from redis.asyncio import Redis as RedisClient

check = RedisHealthCheck(
client_factory=lambda: RedisClient.from_url(
"redis://cache.example.com:6379/3"
)
)
assert "host=cache.example.com" in repr(check), (
"repr should include the host from the Redis URL"
)
assert "db=3" in repr(check), "repr should include the db from the Redis URL"

def test_redis__repr_deprecated_client(self):
"""Verify repr includes host and db when using deprecated client parameter."""
from redis.asyncio import Redis as RedisClient

with pytest.warns(DeprecationWarning):
check = RedisHealthCheck(
client=RedisClient(host="oldhost", port=6379, db=5)
)
assert "host=oldhost" in repr(check), (
"repr should include the host from the deprecated client"
)
assert "db=5" in repr(check), (
"repr should include the db from the deprecated client"
)

def test_redis__repr_sentinel_client(self):
"""Verify repr falls back gracefully for Sentinel clients without host/db."""
from redis.asyncio import Sentinel

check = RedisHealthCheck(
client_factory=lambda: Sentinel([("localhost", 26379)]).master_for(
"mymaster"
)
)
# Sentinel clients don't expose host/db in connection_pool.connection_kwargs
# __repr__ should fall back to the default dataclass repr without raising
assert repr(check) == "Redis()"

def test_redis__repr_cluster_client(self):
"""Verify repr includes startup node hosts for RedisCluster clients."""
from redis.asyncio import RedisCluster
from redis.asyncio.cluster import ClusterNode

check = RedisHealthCheck(
client_factory=lambda: RedisCluster(
startup_nodes=[ClusterNode("node1", 7000), ClusterNode("node2", 7001)]
)
)
assert "node1:7000" in repr(check), (
"repr should include the first cluster node host:port"
)
assert "node2:7001" in repr(check), (
"repr should include the second cluster node host:port"
)

def test_redis__repr_excludes_password(self):
"""Verify repr never leaks passwords for standard Redis clients."""
from redis.asyncio import Redis as RedisClient

check = RedisHealthCheck(
client_factory=lambda: RedisClient(
host="myhost",
port=6379,
db=0,
password="supersecret", # noqa: S106
)
)
assert "supersecret" not in repr(check), (
"repr must never expose the Redis password"
)

def test_redis__repr_excludes_password_from_url(self):
"""Verify repr never leaks passwords embedded in a Redis URL."""
from redis.asyncio import Redis as RedisClient

check = RedisHealthCheck(
client_factory=lambda: RedisClient.from_url(
"redis://admin:supersecret@cache.example.com:6379/3"
)
)
result = repr(check)
assert "supersecret" not in result, (
"repr must never expose the password from a Redis URL"
)
assert "admin" not in result, (
"repr must never expose the username from a Redis URL"
)

def test_redis__repr_excludes_cluster_password(self):
"""Verify repr never leaks passwords for RedisCluster clients."""
from redis.asyncio import RedisCluster
from redis.asyncio.cluster import ClusterNode

check = RedisHealthCheck(
client_factory=lambda: RedisCluster(
startup_nodes=[ClusterNode("node1", 7000)],
password="clusterpass", # noqa: S106
username="clusteruser",
)
)
result = repr(check)
assert "clusterpass" not in result, (
"repr must never expose the cluster password"
)
assert "clusteruser" not in result, (
"repr must never expose the cluster username"
)
Comment thread
codingjoe marked this conversation as resolved.

@pytest.mark.asyncio
async def test_redis__real_connection(self):
"""Ping real Redis server when REDIS_URL is configured."""
Expand Down