Adding moving integration-like tests

Author: petyaslavova

redis/_parsers/base.py

@@ -129,9 +129,10 @@ def __del__(self):
     def on_connect(self, connection):
         "Called when the socket connects"
         self._sock = connection._sock
-        self._buffer = SocketBuffer(
-            self._sock, self.socket_read_size, connection.socket_timeout
-        )
+        timeout = connection.socket_timeout
+        if connection.tmp_relax_timeout != -1:
+            timeout = connection.tmp_relax_timeout
+        self._buffer = SocketBuffer(self._sock, self.socket_read_size, timeout)
         self.encoder = connection.encoder
 
     def on_disconnect(self):
@@ -203,7 +204,7 @@ def handle_push_response(self, response, **kwargs):
             return self.invalidation_push_handler_func(response)
         if msg_type in _MOVING_MESSAGE and self.node_moving_push_handler_func:
             if msg_type in _MOVING_MESSAGE:
-                host, port = response[2].split(":")
+                host, port = response[2].decode().split(":")
                 ttl = response[1]
                 id = 1  # Hardcoded value for sync parser
                 notification = NodeMovingEvent(id, host, port, ttl)

redis/connection.py

@@ -801,6 +801,10 @@ def update_current_socket_timeout(self, relax_timeout: Optional[float] = None):
                 f" to timeout {timeout}; relax_timeout: {relax_timeout}"
             )
             self._sock.settimeout(timeout)
+            self.update_parser_buffer_timeout(timeout)
+
+    def update_parser_buffer_timeout(self, timeout: Optional[float] = None):
+        if self._parser and self._parser._buffer:
             self._parser._buffer.socket_timeout = timeout
 
     def update_tmp_settings(
@@ -1894,7 +1898,7 @@ def disconnect_and_reconfigure_free_connections(
     def update_connections_current_timeout(
         self,
         relax_timeout: Optional[float],
-        include_available_connections: bool = False,
+        include_free_connections: bool = False,
     ):
         """
         Update the timeout either for all connections in the pool or just for the ones in use.
@@ -1912,7 +1916,7 @@ def update_connections_current_timeout(
         for conn in self._in_use_connections:
             self._update_connection_timeout(conn, relax_timeout)
 
-        if include_available_connections:
+        if include_free_connections:
             for conn in self._available_connections:
                 self._update_connection_timeout(conn, relax_timeout)
 
@@ -2157,8 +2161,6 @@ def update_active_connections_for_reconnect(
             connections_in_queue = {conn for conn in self.pool.queue if conn}
             for conn in self._connections:
                 if conn not in connections_in_queue:
-                    if tmp_relax_timeout != -1:
-                        conn.update_socket_timeout(tmp_relax_timeout)
                     self._update_connection_for_reconnect(
                         conn, tmp_host_address, tmp_relax_timeout
                     )
@@ -2177,14 +2179,24 @@ def disconnect_and_reconfigure_free_connections(
                         conn, tmp_host_address, tmp_relax_timeout
                     )
 
-    def update_connections_current_timeout(self, relax_timeout: Optional[float] = None):
+    def update_connections_current_timeout(
+        self,
+        relax_timeout: Optional[float] = None,
+        include_free_connections: bool = False,
+    ):
         logging.debug(
             f"***** Blocking Pool --> Updating timeouts. relax_timeout: {relax_timeout}"
         )
 
         with self._lock:
-            for conn in tuple(self._connections):
-                self._update_connection_timeout(conn, relax_timeout)
+            if include_free_connections:
+                for conn in tuple(self._connections):
+                    self._update_connection_timeout(conn, relax_timeout)
+            else:
+                connections_in_queue = {conn for conn in self.pool.queue if conn}
+                for conn in self._connections:
+                    if conn not in connections_in_queue:
+                        self._update_connection_timeout(conn, relax_timeout)
 
     def update_connections_tmp_settings(
         self,

redis/maintenance_events.py

@@ -323,25 +323,6 @@ def handle_event(self, notification: MaintenanceEvent):
         else:
             logging.error(f"Unhandled notification type: {notification}")
 
-    def handle_node_moved_event(self):
-        with self._lock:
-            self.pool.update_connection_kwargs_with_tmp_settings(
-                tmp_host_address=None,
-                tmp_relax_timeout=-1,
-            )
-            with self.pool._lock:
-                if self.config.is_relax_timeouts_enabled():
-                    # reset the timeout for existing connections
-                    self.pool.update_connections_current_timeout(
-                        relax_timeout=-1, include_available_connections=True
-                    )
-                    logging.debug("***** MOVING END--> TIMEOUTS RESET")
-
-                self.pool.update_connections_tmp_settings(
-                    tmp_host_address=None, tmp_relax_timeout=-1
-                )
-                logging.debug("***** MOVING END--> TMP SETTINGS ADDRESS RESET")
-
     def handle_node_moving_event(self, event: NodeMovingEvent):
         if (
             not self.config.proactive_reconnect
@@ -403,6 +384,26 @@ def handle_node_moving_event(self, event: NodeMovingEvent):
                 f"###### MOVING total execution time: {execution_time_us:.0f} microseconds"
             )
 
+    def handle_node_moved_event(self):
+        logging.debug("***** MOVING END--> Starting to revert the changes.")
+        with self._lock:
+            self.pool.update_connection_kwargs_with_tmp_settings(
+                tmp_host_address=None,
+                tmp_relax_timeout=-1,
+            )
+            with self.pool._lock:
+                if self.config.is_relax_timeouts_enabled():
+                    # reset the timeout for existing connections
+                    self.pool.update_connections_current_timeout(
+                        relax_timeout=-1, include_free_connections=True
+                    )
+                    logging.debug("***** MOVING END--> TIMEOUTS RESET")
+
+                self.pool.update_connections_tmp_settings(
+                    tmp_host_address=None, tmp_relax_timeout=-1
+                )
+                logging.debug("***** MOVING END--> TMP SETTINGS ADDRESS RESET")
+
 
 class MaintenanceEventConnectionHandler:
     def __init__(