[GROW-2938] additional test to redis#1 (redis#2)

* [GROW-2938] add a test for invalid index returned from load balancer

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

tests/test_cluster.py

@@ -1,5 +1,6 @@
 import binascii
 import datetime
+import random
 import select
 import socket
 import socketserver
@@ -8,6 +9,7 @@
 import warnings
 from queue import LifoQueue, Queue
 from time import sleep
+from unittest import mock
 from unittest.mock import DEFAULT, Mock, call, patch
 
 import pytest
@@ -25,6 +27,7 @@
     REDIS_CLUSTER_HASH_SLOTS,
     REPLICA,
     ClusterNode,
+    LoadBalancer,
     NodesManager,
     RedisCluster,
     get_node_name,
@@ -927,7 +930,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):
@@ -2655,6 +2658,37 @@ def test_allow_custom_queue_class(self, queue_class):
         for node in rc.nodes_manager.nodes_cache.values():
             assert node.redis_connection.connection_pool.queue_class == queue_class
 
+    @pytest.mark.parametrize("invalid_index", [-10, 10])
+    def test_return_primary_if_invalid_node_index_is_returned(self, invalid_index):
+        rc = get_mocked_redis_client(
+            url="redis://my@DNS.com:7000",
+            cluster_slots=default_cluster_slots,
+        )
+        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
+            )
+        assert ports == {default_port, 7003}
+
+        ports = set()
+        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
+                )
+        assert ports == {default_port}
+
 
 @pytest.mark.onlycluster
 class TestClusterPubSubObject:
@@ -3085,6 +3119,33 @@ 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: