Merge branch 'master' into feature/driver-info

Author: vchomakov

redis/asyncio/client.py

@@ -249,9 +249,11 @@ def __init__(
         ssl_exclude_verify_flags: Optional[List[VerifyFlags]] = None,
         ssl_ca_certs: Optional[str] = None,
         ssl_ca_data: Optional[str] = None,
+        ssl_ca_path: Optional[str] = None,
         ssl_check_hostname: bool = True,
         ssl_min_version: Optional[TLSVersion] = None,
         ssl_ciphers: Optional[str] = None,
+        ssl_password: Optional[str] = None,
         max_connections: Optional[int] = None,
         single_connection_client: bool = False,
         health_check_interval: int = 0,
@@ -371,9 +373,11 @@ def __init__(
                             "ssl_exclude_verify_flags": ssl_exclude_verify_flags,
                             "ssl_ca_certs": ssl_ca_certs,
                             "ssl_ca_data": ssl_ca_data,
+                            "ssl_ca_path": ssl_ca_path,
                             "ssl_check_hostname": ssl_check_hostname,
                             "ssl_min_version": ssl_min_version,
                             "ssl_ciphers": ssl_ciphers,
+                            "ssl_password": ssl_password,
                         }
                     )
             # This arg only used if no pool is passed in

redis/asyncio/connection.py

@@ -850,9 +850,11 @@ def __init__(
         ssl_exclude_verify_flags: Optional[List["ssl.VerifyFlags"]] = None,
         ssl_ca_certs: Optional[str] = None,
         ssl_ca_data: Optional[str] = None,
+        ssl_ca_path: Optional[str] = None,
         ssl_check_hostname: bool = True,
         ssl_min_version: Optional[TLSVersion] = None,
         ssl_ciphers: Optional[str] = None,
+        ssl_password: Optional[str] = None,
         **kwargs,
     ):
         if not SSL_AVAILABLE:
@@ -866,9 +868,11 @@ def __init__(
             exclude_verify_flags=ssl_exclude_verify_flags,
             ca_certs=ssl_ca_certs,
             ca_data=ssl_ca_data,
+            ca_path=ssl_ca_path,
             check_hostname=ssl_check_hostname,
             min_version=ssl_min_version,
             ciphers=ssl_ciphers,
+            password=ssl_password,
         )
         super().__init__(**kwargs)
 
@@ -923,10 +927,12 @@ class RedisSSLContext:
         "exclude_verify_flags",
         "ca_certs",
         "ca_data",
+        "ca_path",
         "context",
         "check_hostname",
         "min_version",
         "ciphers",
+        "password",
     )
 
     def __init__(
@@ -938,9 +944,11 @@ def __init__(
         exclude_verify_flags: Optional[List["ssl.VerifyFlags"]] = None,
         ca_certs: Optional[str] = None,
         ca_data: Optional[str] = None,
+        ca_path: Optional[str] = None,
         check_hostname: bool = False,
         min_version: Optional[TLSVersion] = None,
         ciphers: Optional[str] = None,
+        password: Optional[str] = None,
     ):
         if not SSL_AVAILABLE:
             raise RedisError("Python wasn't built with SSL support")
@@ -965,11 +973,13 @@ def __init__(
         self.exclude_verify_flags = exclude_verify_flags
         self.ca_certs = ca_certs
         self.ca_data = ca_data
+        self.ca_path = ca_path
         self.check_hostname = (
             check_hostname if self.cert_reqs != ssl.CERT_NONE else False
         )
         self.min_version = min_version
         self.ciphers = ciphers
+        self.password = password
         self.context: Optional[SSLContext] = None
 
     def get(self) -> SSLContext:
@@ -983,10 +993,16 @@ def get(self) -> SSLContext:
             if self.exclude_verify_flags:
                 for flag in self.exclude_verify_flags:
                     context.verify_flags &= ~flag
-            if self.certfile and self.keyfile:
-                context.load_cert_chain(certfile=self.certfile, keyfile=self.keyfile)
-            if self.ca_certs or self.ca_data:
-                context.load_verify_locations(cafile=self.ca_certs, cadata=self.ca_data)
+            if self.certfile or self.keyfile:
+                context.load_cert_chain(
+                    certfile=self.certfile,
+                    keyfile=self.keyfile,
+                    password=self.password,
+                )
+            if self.ca_certs or self.ca_data or self.ca_path:
+                context.load_verify_locations(
+                    cafile=self.ca_certs, capath=self.ca_path, cadata=self.ca_data
+                )
             if self.min_version is not None:
                 context.minimum_version = self.min_version
             if self.ciphers is not None:
@@ -1239,16 +1255,17 @@ def can_get_connection(self) -> bool:
         version="5.3.0",
     )
     async def get_connection(self, command_name=None, *keys, **options):
+        """Get a connected connection from the pool"""
         async with self._lock:
-            """Get a connected connection from the pool"""
             connection = self.get_available_connection()
-            try:
-                await self.ensure_connection(connection)
-            except BaseException:
-                await self.release(connection)
-                raise
 
-        return connection
+        # We now perform the connection check outside of the lock.
+        try:
+            await self.ensure_connection(connection)
+            return connection
+        except BaseException:
+            await self.release(connection)
+            raise
 
     def get_available_connection(self):
         """Get a connection from the pool, without making sure it is connected"""

redis/client.py

@@ -1035,10 +1035,22 @@ def is_health_check_response(self, response) -> bool:
         If there are no subscriptions redis responds to PING command with a
         bulk response, instead of a multi-bulk with "pong" and the response.
         """
-        return response in [
-            self.health_check_response,  # If there was a subscription
-            self.health_check_response_b,  # If there wasn't
-        ]
+        if self.encoder.decode_responses:
+            return (
+                response
+                in [
+                    self.health_check_response,  # If there is a subscription
+                    self.HEALTH_CHECK_MESSAGE,  # If there are no subscriptions and decode_responses=True
+                ]
+            )
+        else:
+            return (
+                response
+                in [
+                    self.health_check_response,  # If there is a subscription
+                    self.health_check_response_b,  # If there isn't a subscription and decode_responses=False
+                ]
+            )
 
     def check_health(self) -> None:
         conn = self.connection

redis/cluster.py

@@ -185,6 +185,7 @@ def parse_cluster_myshardid(resp, **options):
     "ssl",
     "ssl_ca_certs",
     "ssl_ca_data",
+    "ssl_ca_path",
     "ssl_certfile",
     "ssl_cert_reqs",
     "ssl_include_verify_flags",
@@ -2207,7 +2208,8 @@ def _sharded_message_generator(self):
 
     def _pubsubs_generator(self):
         while True:
-            yield from self.node_pubsub_mapping.values()
+            current_nodes = list(self.node_pubsub_mapping.values())
+            yield from current_nodes
 
     def get_sharded_message(
         self, ignore_subscribe_messages=False, timeout=0.0, target_node=None

tests/test_asyncio/test_connection_pool.py

@@ -222,6 +222,81 @@ async def test_pool_disconnect(self, master_host):
             await pool.disconnect(inuse_connections=False)
             assert conn.is_connected
 
+    async def test_lock_not_held_during_connection_establishment(self):
+        """
+        Test that the connection pool lock is not held during the
+        ensure_connection call, which involves socket connection and handshake.
+        This is important for performance under high load.
+        """
+        lock_states = []
+
+        class SlowConnectConnection(DummyConnection):
+            """Connection that simulates slow connection establishment"""
+
+            async def connect(self):
+                # Check if the pool's lock is held during connection
+                # We access the pool through the outer scope
+                lock_states.append(pool._lock.locked())
+                # Simulate slow connection
+                await asyncio.sleep(0.01)
+                self._connected = True
+
+        async with self.get_pool(connection_class=SlowConnectConnection) as pool:
+            # Get a connection - this should call connect() outside the lock
+            connection = await pool.get_connection()
+
+            # Verify the lock was NOT held during connect
+            assert len(lock_states) > 0, "connect() should have been called"
+            assert lock_states[0] is False, (
+                "Lock should not be held during connection establishment"
+            )
+
+            await pool.release(connection)
+
+    async def test_concurrent_connection_acquisition_performance(self):
+        """
+        Test that multiple concurrent connection acquisitions don't block
+        each other during connection establishment.
+        """
+        connection_delay = 0.05
+        num_connections = 3
+
+        class SlowConnectConnection(DummyConnection):
+            """Connection that simulates slow connection establishment"""
+
+            async def connect(self):
+                # Simulate slow connection (e.g., network latency, TLS handshake)
+                await asyncio.sleep(connection_delay)
+                self._connected = True
+
+        async with self.get_pool(
+            connection_class=SlowConnectConnection, max_connections=10
+        ) as pool:
+            # Start acquiring multiple connections concurrently
+            start_time = asyncio.get_running_loop().time()
+
+            # Try to get connections concurrently
+            connections = await asyncio.gather(
+                *[pool.get_connection() for _ in range(num_connections)]
+            )
+
+            elapsed_time = asyncio.get_running_loop().time() - start_time
+
+            # With proper lock handling, these should complete mostly in parallel
+            # If the lock was held during connect(), it would take num_connections * connection_delay
+            # With lock only during pop, it should take ~connection_delay (connections in parallel)
+            # We allow 2.5x overhead for system variance
+            max_allowed_time = connection_delay * 2.5
+            assert elapsed_time < max_allowed_time, (
+                f"Concurrent connections took {elapsed_time:.3f}s, "
+                f"expected < {max_allowed_time:.3f}s. "
+                f"This suggests lock was held during connection establishment."
+            )
+
+            # Clean up
+            for conn in connections:
+                await pool.release(conn)
+
 
 class TestBlockingConnectionPool:
     @asynccontextmanager

tests/test_asyncio/test_pubsub.py

@@ -671,6 +671,63 @@ async def test_send_pubsub_ping_message(self, r: redis.Redis):
         await p.aclose()
 
 
+@pytest.mark.onlynoncluster
+class TestPubSubHealthCheckResponse:
+    """Tests for health check response validation with different decode_responses settings"""
+
+    async def test_health_check_response_decode_false_list_format(self, r: redis.Redis):
+        """Test health_check_response includes list format with decode_responses=False"""
+        p = r.pubsub()
+        # List format: [b"pong", b"redis-py-health-check"]
+        assert [b"pong", b"redis-py-health-check"] in p.health_check_response
+        await p.aclose()
+
+    async def test_health_check_response_decode_false_bytes_format(
+        self, r: redis.Redis
+    ):
+        """Test health_check_response includes bytes format with decode_responses=False"""
+        p = r.pubsub()
+        # Bytes format: b"redis-py-health-check"
+        assert b"redis-py-health-check" in p.health_check_response
+        await p.aclose()
+
+    async def test_health_check_response_decode_true_list_format(self, create_redis):
+        """Test health_check_response includes list format with decode_responses=True"""
+        r = await create_redis(decode_responses=True)
+        p = r.pubsub()
+        # List format: ["pong", "redis-py-health-check"]
+        assert ["pong", "redis-py-health-check"] in p.health_check_response
+        await p.aclose()
+        await r.aclose()
+
+    async def test_health_check_response_decode_true_string_format(self, create_redis):
+        """Test health_check_response includes string format with decode_responses=True"""
+        r = await create_redis(decode_responses=True)
+        p = r.pubsub()
+        # String format: "redis-py-health-check" (THE FIX!)
+        assert "redis-py-health-check" in p.health_check_response
+        await p.aclose()
+        await r.aclose()
+
+    async def test_health_check_response_decode_false_excludes_string(
+        self, r: redis.Redis
+    ):
+        """Test health_check_response excludes string format with decode_responses=False"""
+        p = r.pubsub()
+        # String format should NOT be in the list when decode_responses=False
+        assert "redis-py-health-check" not in p.health_check_response
+        await p.aclose()
+
+    async def test_health_check_response_decode_true_excludes_bytes(self, create_redis):
+        """Test health_check_response excludes bytes format with decode_responses=True"""
+        r = await create_redis(decode_responses=True)
+        p = r.pubsub()
+        # Bytes format should NOT be in the list when decode_responses=True
+        assert b"redis-py-health-check" not in p.health_check_response
+        await p.aclose()
+        await r.aclose()
+
+
 @pytest.mark.onlynoncluster
 class TestPubSubConnectionKilled:
     @skip_if_server_version_lt("3.0.0")

tests/test_asyncio/test_ssl.py

@@ -141,3 +141,55 @@ def capture_context_create_default():
 
         finally:
             await r.aclose()
+
+    async def test_ssl_ca_path_parameter(self, request):
+        """Test that ssl_ca_path parameter is properly passed to SSLConnection"""
+        ssl_url = request.config.option.redis_ssl_url
+        parsed_url = urlparse(ssl_url)
+
+        # Test with a mock ca_path directory
+        test_ca_path = "/tmp/test_ca_certs"
+
+        r = redis.Redis(
+            host=parsed_url.hostname,
+            port=parsed_url.port,
+            ssl=True,
+            ssl_cert_reqs="none",
+            ssl_ca_path=test_ca_path,
+        )
+
+        try:
+            # Get the connection to verify ssl_ca_path is passed through
+            conn = r.connection_pool.make_connection()
+            assert isinstance(conn, redis.SSLConnection)
+
+            # Verify the ca_path is stored in the SSL context
+            assert conn.ssl_context.ca_path == test_ca_path
+        finally:
+            await r.aclose()
+
+    async def test_ssl_password_parameter(self, request):
+        """Test that ssl_password parameter is properly passed to SSLConnection"""
+        ssl_url = request.config.option.redis_ssl_url
+        parsed_url = urlparse(ssl_url)
+
+        # Test with a mock password for encrypted private key
+        test_password = "test_key_password"
+
+        r = redis.Redis(
+            host=parsed_url.hostname,
+            port=parsed_url.port,
+            ssl=True,
+            ssl_cert_reqs="none",
+            ssl_password=test_password,
+        )
+
+        try:
+            # Get the connection to verify ssl_password is passed through
+            conn = r.connection_pool.make_connection()
+            assert isinstance(conn, redis.SSLConnection)
+
+            # Verify the password is stored in the SSL context
+            assert conn.ssl_context.password == test_password
+        finally:
+            await r.aclose()