[GROW-2938] add a test for pipeline additional backoff

Author: zach-iee

tests/test_cluster.py

@@ -14,7 +14,7 @@
 
 import pytest
 
-from redis import Redis, crc
+from redis import Redis
 from redis.backoff import (
     ConstantBackoff,
     ExponentialBackoff,
@@ -928,7 +928,7 @@ def raise_error(target_node, *args, **kwargs):
                 rc = get_mocked_redis_client(
                     host=default_host,
                     port=default_port,
-                    retry=Retry(ConstantBackoff(1), 3),
+                    retry=Retry(ConstantBackoff(1), 10),
                 )
 
                 with pytest.raises(error):
@@ -2662,17 +2662,32 @@ def test_return_primary_if_invalid_node_index_is_returned(self, invalid_index):
             url="redis://[email protected]:7000",
             cluster_slots=default_cluster_slots,
         )
-        random_slot = random.randint(default_cluster_slots[0][0], default_cluster_slots[0][1])
+        random_slot = random.randint(
+            default_cluster_slots[0][0],
+            default_cluster_slots[0][1]
+        )
 
         ports = set()
         for _ in range(0, 10):
-            ports.add(rc.nodes_manager.get_node_from_slot(random_slot, read_from_replicas=True).port)
+            ports.add(
+                rc.nodes_manager.get_node_from_slot(
+                    random_slot,
+                    read_from_replicas=True
+                ).port
+            )
         assert ports == {default_port, 7003}
 
         ports = set()
-        with mock.patch.object(LoadBalancer, "get_server_index", return_value=invalid_index):
+        with mock.patch.object(
+          LoadBalancer, "get_server_index", return_value=invalid_index
+        ):
             for _ in range(0, 10):
-                ports.add(rc.nodes_manager.get_node_from_slot(random_slot, read_from_replicas=True).port)
+                ports.add(
+                    rc.nodes_manager.get_node_from_slot(
+                        random_slot,
+                        read_from_replicas=True
+                    ).port
+                )
         assert ports == {default_port}
 
 
@@ -3078,6 +3093,32 @@ def test_empty_stack(self, r):
         result = p.execute()
         assert result == []
 
+    @pytest.mark.parametrize("error", [ConnectionError, TimeoutError])
+    def test_additional_backoff_cluster_pipeline(self, r, error):
+        with patch.object(ConstantBackoff, "compute") as compute:
+
+            def _compute(target_node, *args, **kwargs):
+                return 1
+
+            compute.side_effect = _compute
+            with patch("redis.cluster.get_connection") as get_connection:
+
+                def raise_error(target_node, *args, **kwargs):
+                    get_connection.failed_calls += 1
+                    raise error("mocked error")
+
+                get_connection.side_effect = raise_error
+
+                r.set_retry(Retry(ConstantBackoff(1), 10))
+                pipeline = r.pipeline()
+
+                with pytest.raises(error):
+                    pipeline.get("bar")
+                    pipeline.get("bar")
+                    pipeline.execute()
+                # cluster pipeline does one more back off than a single Redis command
+                #   this is not required, but it's just how it's implemented as of now
+                assert compute.call_count == r.cluster_error_retry_attempts + 1
 
 @pytest.mark.onlycluster
 class TestReadOnlyPipeline: