[async] Adding access to cluster client's nodes_manager and set_response_callback in ClusterPipeline objects (#3989)

Author: petyaslavova

redis/asyncio/cluster.py

@@ -1989,6 +1989,15 @@ def __init__(
             else TransactionStrategy(self)
         )
 
+    @property
+    def nodes_manager(self) -> "NodesManager":
+        """Get the nodes manager from the cluster client."""
+        return self.cluster_client.nodes_manager
+
+    def set_response_callback(self, command: str, callback: ResponseCallbackT) -> None:
+        """Set a custom response callback on the cluster client."""
+        self.cluster_client.set_response_callback(command, callback)
+
     async def initialize(self) -> "ClusterPipeline":
         await self._execution_strategy.initialize()
         return self

tests/test_asyncio/test_cluster.py

@@ -3162,6 +3162,44 @@ def cmd_init_mock(
 class TestClusterPipeline:
     """Tests for the ClusterPipeline class."""
 
+    async def test_pipeline_nodes_manager_property(self) -> None:
+        """
+        Test that ClusterPipeline exposes nodes_manager property
+        that delegates to the cluster client's nodes_manager.
+        """
+        r = await get_mocked_redis_client(host=default_host, port=default_port)
+        try:
+            pipeline = r.pipeline()
+            # Verify that nodes_manager property exists and returns the same object
+            # as the cluster client's nodes_manager
+            assert pipeline.nodes_manager is r.nodes_manager
+            # Verify that we can access nodes_manager attributes
+            assert pipeline.nodes_manager.default_node is not None
+        finally:
+            await r.aclose()
+
+    async def test_pipeline_set_response_callback(self) -> None:
+        """
+        Test that ClusterPipeline exposes set_response_callback method
+        that delegates to the cluster client's set_response_callback.
+        """
+        r = await get_mocked_redis_client(host=default_host, port=default_port)
+        try:
+            pipeline = r.pipeline()
+
+            # Define a custom callback
+            def custom_callback(response):
+                return f"custom_{response}"
+
+            # Set the callback via the pipeline
+            pipeline.set_response_callback("CUSTOM_CMD", custom_callback)
+
+            # Verify that the callback was set on the cluster client
+            assert "CUSTOM_CMD" in r.response_callbacks
+            assert r.response_callbacks["CUSTOM_CMD"] is custom_callback
+        finally:
+            await r.aclose()
+
     async def test_blocked_arguments(self, r: RedisCluster) -> None:
         """Test handling for blocked pipeline arguments."""
 

tests/test_asyncio/test_pipeline.py

@@ -466,6 +466,26 @@ async def test_pipeline_with_msetex(self, r):
                 {"key1_transaction": "value1", "key2_transaction": "value2"}, ex=10
             )
 
+    async def test_pipeline_json_module_access(self, r):
+        """
+        Test that pipeline can access the json() module.
+        The JSON module requires nodes_manager (for cluster) and set_response_callback
+        on the client during initialization.
+
+        """
+        pipeline = r.pipeline()
+
+        # This should not raise an AttributeError
+        json_pipeline = pipeline.json()
+
+        # Verify the JSON module was created successfully
+        assert json_pipeline is not None
+        assert json_pipeline.client is pipeline
+
+        # Verify that JSON callbacks were registered
+        assert "JSON.SET" in r.response_callbacks
+        assert "JSON.GET" in r.response_callbacks
+
 
 @pytest.mark.asyncio
 class TestAsyncPipelineOperationDurationMetricsRecording: