Added test cases

Author: vladvildanov

redis/cluster.py

@@ -2116,6 +2116,7 @@ def __init__(
         ) if not transaction else TransactionStrategy(
             self
         )
+        self.command_stack = self._execution_strategy.command_queue
 
     def __repr__(self):
         """ """
@@ -2137,7 +2138,7 @@ def __del__(self):
 
     def __len__(self):
         """ """
-        return len(self._execution_strategy.command_queue)
+        return len(self.command_stack)
 
     def __bool__(self):
         "Pipeline instances should  always evaluate to True on Python 3+"
@@ -2147,7 +2148,7 @@ def execute_command(self, *args, **kwargs):
         """
         Wrapper function for pipeline_execute_command
         """
-        return self.pipeline_execute_command(*args, **kwargs)
+        return self._execution_strategy.execute_command(*args, **kwargs)
 
     def pipeline_execute_command(self, *args, **options):
         """
@@ -2161,8 +2162,7 @@ def pipeline_execute_command(self, *args, **options):
         At some other point, you can then run: pipe.execute(),
         which will execute all commands queued in the pipe.
         """
-        self._execution_strategy.pipeline_execute_command(*args, **options)
-        return self
+        return self._execution_strategy.execute_command(*args, **options)
 
     def annotate_exception(self, exception, number, command):
         """
@@ -2557,6 +2557,7 @@ def pipeline_execute_command(self, *args, **options):
         self._command_queue.append(
             PipelineCommand(args, options, len(self._command_queue))
         )
+        return self._pipe
 
     @abstractmethod
     def execute(self, raise_on_error: bool = True) -> List[Any]:
@@ -2605,7 +2606,7 @@ def __init__(self, pipe: ClusterPipeline):
         self.command_flags = pipe.command_flags
 
     def execute_command(self, *args, **kwargs):
-        self.pipeline_execute_command(*args, **kwargs)
+        return self.pipeline_execute_command(*args, **kwargs)
 
     def _raise_first_error(self, stack):
         """
@@ -3150,6 +3151,10 @@ def _execute_transaction(
         except ResponseError as e:
             self.annotate_exception(e, 0, "MULTI")
             errors.append(e)
+        except (ClusterDownError, ConnectionError) as cluster_error:
+            self._cluster_error = True
+            self.annotate_exception(cluster_error, 0, "MULTI")
+            raise
 
         # and all the other commands
         for i, command in enumerate(self._command_queue):

tests/test_cluster.py

@@ -265,12 +265,12 @@ def moved_redirection_helper(request, failover=False):
     slot = 12182
     redirect_node = None
     # Get the current primary that holds this slot
-    prev_primary = rc._nodes_manager.get_node_from_slot(slot)
+    prev_primary = rc.nodes_manager.get_node_from_slot(slot)
     if failover:
-        if len(rc._nodes_manager.slots_cache[slot]) < 2:
+        if len(rc.nodes_manager.slots_cache[slot]) < 2:
             warnings.warn("Skipping this test since it requires to have a replica")
             return
-        redirect_node = rc._nodes_manager.slots_cache[slot][1]
+        redirect_node = rc.nodes_manager.slots_cache[slot][1]
     else:
         # Use one of the primaries to be the redirected node
         redirect_node = rc.get_primaries()[0]
@@ -290,7 +290,7 @@ def ok_response(connection, *args, **options):
 
         parse_response.side_effect = moved_redirect_effect
         assert rc.execute_command("SET", "foo", "bar") == "MOCK_OK"
-        slot_primary = rc._nodes_manager.slots_cache[slot][0]
+        slot_primary = rc.nodes_manager.slots_cache[slot][0]
         assert slot_primary == redirect_node
         if failover:
             assert rc.get_node(host=r_host, port=r_port).server_type == PRIMARY
@@ -722,7 +722,7 @@ def test_all_nodes(self, r):
         """
         Set a list of nodes and it should be possible to iterate over all
         """
-        nodes = [node for node in r._nodes_manager.nodes_cache.values()]
+        nodes = [node for node in r.nodes_manager.nodes_cache.values()]
 
         for i, node in enumerate(r.get_nodes()):
             assert node in nodes
@@ -734,7 +734,7 @@ def test_all_nodes_masters(self, r):
         """
         nodes = [
             node
-            for node in r._nodes_manager.nodes_cache.values()
+            for node in r.nodes_manager.nodes_cache.values()
             if node.server_type == PRIMARY
         ]
 
@@ -805,7 +805,7 @@ def test_get_node_from_key(self, r):
         """
         key = "bar"
         slot = r.keyslot(key)
-        slot_nodes = r._nodes_manager.slots_cache.get(slot)
+        slot_nodes = r.nodes_manager.slots_cache.get(slot)
         primary = slot_nodes[0]
         assert r.get_node_from_key(key, replica=False) == primary
         replica = r.get_node_from_key(key, replica=True)
@@ -1003,8 +1003,8 @@ class TestClusterRedisCommands:
 
     def test_case_insensitive_command_names(self, r):
         assert (
-            r._cluster_response_callbacks["cluster slots"]
-            == r._cluster_response_callbacks["CLUSTER SLOTS"]
+            r.cluster_response_callbacks["cluster slots"]
+            == r.cluster_response_callbacks["CLUSTER SLOTS"]
         )
 
     def test_get_and_set(self, r):
@@ -1275,7 +1275,7 @@ def test_cluster_addslotsrange(self, r):
 
     @skip_if_redis_enterprise()
     def test_cluster_countkeysinslot(self, r):
-        node = r._nodes_manager.get_node_from_slot(1)
+        node = r.nodes_manager.get_node_from_slot(1)
         mock_node_resp(node, 2)
         assert r.cluster_countkeysinslot(1) == 2
 
@@ -1450,7 +1450,7 @@ def test_cluster_save_config(self, r):
     @skip_if_redis_enterprise()
     def test_cluster_get_keys_in_slot(self, r):
         response = ["{foo}1", "{foo}2"]
-        node = r._nodes_manager.get_node_from_slot(12182)
+        node = r.nodes_manager.get_node_from_slot(12182)
         mock_node_resp(node, response)
         keys = r.cluster_get_keys_in_slot(12182, 4)
         assert keys == response
@@ -1476,7 +1476,7 @@ def test_cluster_setslot(self, r):
             r.cluster_failover(node, "STATE")
 
     def test_cluster_setslot_stable(self, r):
-        node = r._nodes_manager.get_node_from_slot(12182)
+        node = r.nodes_manager.get_node_from_slot(12182)
         mock_node_resp(node, "OK")
         assert r.cluster_setslot_stable(12182) is True
         assert node.redis_connection.connection.read_response.called
@@ -1562,7 +1562,7 @@ def test_info(self, r):
         r.set("z{1}", 3)
         # Get node that handles the slot
         slot = r.keyslot("x{1}")
-        node = r._nodes_manager.get_node_from_slot(slot)
+        node = r.nodes_manager.get_node_from_slot(slot)
         # Run info on that node
         info = r.info(target_nodes=node)
         assert isinstance(info, dict)
@@ -1618,7 +1618,7 @@ def test_slowlog_get_limit(self, r, slowlog):
 
     def test_slowlog_length(self, r, slowlog):
         r.get("foo")
-        node = r._nodes_manager.get_node_from_slot(key_slot(b"foo"))
+        node = r.nodes_manager.get_node_from_slot(key_slot(b"foo"))
         slowlog_len = r.slowlog_len(target_nodes=node)
         assert isinstance(slowlog_len, int)
 
@@ -1644,7 +1644,7 @@ def test_memory_stats(self, r):
         # put a key into the current db to make sure that "db.<current-db>"
         # has data
         r.set("foo", "bar")
-        node = r._nodes_manager.get_node_from_slot(key_slot(b"foo"))
+        node = r.nodes_manager.get_node_from_slot(key_slot(b"foo"))
         stats = r.memory_stats(target_nodes=node)
         assert isinstance(stats, dict)
         for key, value in stats.items():
@@ -2530,7 +2530,7 @@ class TestNodesManager:
     """
 
     def test_load_balancer(self, r):
-        n_manager = r._nodes_manager
+        n_manager = r.nodes_manager
         lb = n_manager.read_load_balancer
         slot_1 = 1257
         slot_2 = 8975
@@ -3428,7 +3428,7 @@ def test_readonly_pipeline_from_readonly_client(self, request):
             mock_all_nodes_resp(ro, "MOCK_OK")
             assert readonly_pipe.read_from_replicas is True
             assert readonly_pipe.get(key).get(key).execute() == ["MOCK_OK", "MOCK_OK"]
-            slot_nodes = ro._nodes_manager.slots_cache[ro.keyslot(key)]
+            slot_nodes = ro.nodes_manager.slots_cache[ro.keyslot(key)]
             if len(slot_nodes) > 1:
                 executed_on_replica = False
                 for node in slot_nodes:
@@ -3468,7 +3468,7 @@ def test_readonly_pipeline_with_reading_from_replicas_strategies(
             mock_all_nodes_resp(ro, "MOCK_OK")
             assert readonly_pipe.load_balancing_strategy == load_balancing_strategy
             assert readonly_pipe.get(key).get(key).execute() == ["MOCK_OK", "MOCK_OK"]
-            slot_nodes = ro._nodes_manager.slots_cache[ro.keyslot(key)]
+            slot_nodes = ro.nodes_manager.slots_cache[ro.keyslot(key)]
             executed_on_replicas_only = True
             for node in slot_nodes:
                 if node.server_type == PRIMARY: