Wrapper for fault injector and mock proxy service + cluster maint notifications e2e tests for node slots migration and node fail over (#3882)

* Adding fault injector abstraction that would be able to work with re fault injector as well as with proxy server

* Adding wrapper that will allow using the mock proxy for e2e tests execution. Updating standalone maint notifications tests. Adding cluster maint notifications test with slot migration and node fail over. Fix in the connection logic when applying relaxed_timeout.

* Apply suggestions from code review

---------

Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>

Author: petyaslavova

docker-compose.yml

@@ -134,6 +134,8 @@ services:
       API_PORT: "4000"
       ENABLE_LOGGING: true
       SIMULATE_CLUSTER: true
+      DEFAULT_INTERCEPTORS: "cluster,hitless,logger"
+
     ports:
       - "15379:15379"
       - "15380:15380"

redis/_parsers/socket.py

@@ -62,7 +62,7 @@ def _read_from_socket(
             sock.settimeout(timeout)
         try:
             while True:
-                data = self._sock.recv(socket_read_size)
+                data = sock.recv(socket_read_size)
                 # an empty string indicates the server shutdown the socket
                 if isinstance(data, bytes) and len(data) == 0:
                     raise ConnectionError(SERVER_CLOSED_CONNECTION_ERROR)

redis/cluster.py

@@ -2027,10 +2027,10 @@ def initialize(
                 # Make sure cluster mode is enabled on this node
                 try:
                     cluster_slots = str_if_bytes(r.execute_command("CLUSTER SLOTS"))
+                    # For some cases we might not want to disconnect current pool and
+                    # lose in flight commands responses
                     if disconnect_startup_nodes_pools:
                         # Disconnect the connection pool to avoid keeping the connection open
-                        # For some cases we might not want to disconnect current pool and
-                        # lose in flight commands responses
                         r.connection_pool.disconnect()
                 except ResponseError:
                     raise RedisClusterException(

redis/connection.py

@@ -698,7 +698,13 @@ def update_current_socket_timeout(self, relaxed_timeout: Optional[float] = None)
         conn_socket = self._get_socket()
         if conn_socket:
             timeout = relaxed_timeout if relaxed_timeout != -1 else self.socket_timeout
-            conn_socket.settimeout(timeout)
+            # if the current timeout is 0 it means we are in the middle of a can_read call
+            # in this case we don't want to change the timeout because the operation
+            # is non-blocking and should return immediately
+            # Changing the state from non-blocking to blocking in the middle of a read operation
+            # will lead to a deadlock
+            if conn_socket.gettimeout() != 0:
+                conn_socket.settimeout(timeout)
             self.update_parser_timeout(timeout)
 
     def update_parser_timeout(self, timeout: Optional[float] = None):

tests/conftest.py

@@ -163,6 +163,13 @@ def pytest_addoption(parser):
         help="Name of the Redis endpoint the tests should be executed on",
     )
 
+    parser.addoption(
+        "--cluster-endpoint-name",
+        action="store",
+        default=None,
+        help="Name of the Redis endpoint with OSS API the tests should be executed on",
+    )
+
 
 def _get_info(redis_url):
     client = redis.Redis.from_url(redis_url)

tests/maint_notifications/proxy_server_helpers.py

@@ -1,7 +1,6 @@
 import base64
 from dataclasses import dataclass
 import logging
-import re
 from typing import Union
 
 from redis.http.http_client import HttpClient, HttpError
@@ -10,6 +9,25 @@
 class RespTranslator:
     """Helper class to translate between RESP and other encodings."""
 
+    @staticmethod
+    def re_cluster_maint_notification_to_resp(txt: str) -> str:
+        """Convert query to RESP format."""
+        parts = txt.split()
+
+        match parts:
+            case ["MOVING", seq_id, time, new_host]:
+                return f">4\r\n+MOVING\r\n:{seq_id}\r\n:{time}\r\n+{new_host}\r\n"
+            case ["MIGRATING", seq_id, time, shards]:
+                return f">4\r\n+MIGRATING\r\n:{seq_id}\r\n:{time}\r\n+{shards}\r\n"
+            case ["MIGRATED", seq_id, shards]:
+                return f">3\r\n+MIGRATED\r\n:{seq_id}\r\n+{shards}\r\n"
+            case ["FAILING_OVER", seq_id, time, shards]:
+                return f">4\r\n+FAILING_OVER\r\n:{seq_id}\r\n:{time}\r\n+{shards}\r\n"
+            case ["FAILED_OVER", seq_id, shards]:
+                return f">3\r\n+FAILED_OVER\r\n:{seq_id}\r\n+{shards}\r\n"
+            case _:
+                raise NotImplementedError(f"Unknown notification: {txt}")
+
     @staticmethod
     def oss_maint_notification_to_resp(txt: str) -> str:
         """Convert query to RESP format."""
@@ -232,8 +250,7 @@ def send_notification(
 
         if not conn_ids:
             raise RuntimeError(
-                f"No connections found for node {node_port}. "
-                f"Available nodes: {list(set(c.get('node') for c in stats.get('connections', {}).values()))}"
+                f"No connections found for node {connected_to_port}. \nStats: {stats}"
             )
 
         # Send notification to each connection

tests/test_asyncio/test_scenario/conftest.py

@@ -17,7 +17,7 @@
 from redis.event import AsyncEventListenerInterface, EventDispatcher
 from redis.multidb.failure_detector import DEFAULT_MIN_NUM_FAILURES
 from tests.test_scenario.conftest import get_endpoints_config, extract_cluster_fqdn
-from tests.test_scenario.fault_injector_client import FaultInjectorClient
+from tests.test_scenario.fault_injector_client import REFaultInjector
 
 
 class CheckActiveDatabaseChangedListener(AsyncEventListenerInterface):
@@ -31,7 +31,7 @@ async def listen(self, event: AsyncActiveDatabaseChanged):
 @pytest.fixture()
 def fault_injector_client():
     url = os.getenv("FAULT_INJECTION_API_URL", "http://127.0.0.1:20324")
-    return FaultInjectorClient(url)
+    return REFaultInjector(url)
 
 
 @pytest_asyncio.fixture()

tests/test_scenario/conftest.py

@@ -6,6 +6,7 @@
 from urllib.parse import urlparse
 
 import pytest
+from redis import RedisCluster
 
 from redis.backoff import NoBackoff, ExponentialBackoff
 from redis.event import EventDispatcher, EventListenerInterface
@@ -22,12 +23,16 @@
 from redis.client import Redis
 from redis.maint_notifications import EndpointType, MaintNotificationsConfig
 from redis.retry import Retry
-from tests.test_scenario.fault_injector_client import FaultInjectorClient
+from tests.test_scenario.fault_injector_client import (
+    ProxyServerFaultInjector,
+    REFaultInjector,
+)
 
 RELAXED_TIMEOUT = 30
 CLIENT_TIMEOUT = 5
 
 DEFAULT_ENDPOINT_NAME = "m-standard"
+DEFAULT_OSS_API_ENDPOINT_NAME = "oss-api"
 
 
 class CheckActiveDatabaseChangedListener(EventListenerInterface):
@@ -38,13 +43,24 @@ def listen(self, event: ActiveDatabaseChanged):
         self.is_changed_flag = True
 
 
+def use_mock_proxy():
+    return os.getenv("REDIS_ENTERPRISE_TESTS", "true").lower() == "false"
+
+
 @pytest.fixture()
 def endpoint_name(request):
     return request.config.getoption("--endpoint-name") or os.getenv(
         "REDIS_ENDPOINT_NAME", DEFAULT_ENDPOINT_NAME
     )
 
 
+@pytest.fixture()
+def cluster_endpoint_name(request):
+    return request.config.getoption("--cluster-endpoint-name") or os.getenv(
+        "REDIS_CLUSTER_ENDPOINT_NAME", DEFAULT_OSS_API_ENDPOINT_NAME
+    )
+
+
 def get_endpoints_config(endpoint_name: str):
     endpoints_config = os.getenv("REDIS_ENDPOINTS_CONFIG_PATH", None)
 
@@ -67,10 +83,27 @@ def endpoints_config(endpoint_name: str):
     return get_endpoints_config(endpoint_name)
 
 
+@pytest.fixture()
+def cluster_endpoints_config(cluster_endpoint_name: str):
+    return get_endpoints_config(cluster_endpoint_name)
+
+
 @pytest.fixture()
 def fault_injector_client():
-    url = os.getenv("FAULT_INJECTION_API_URL", "http://127.0.0.1:20324")
-    return FaultInjectorClient(url)
+    if use_mock_proxy():
+        return ProxyServerFaultInjector(oss_cluster=False)
+    else:
+        url = os.getenv("FAULT_INJECTION_API_URL", "http://127.0.0.1:20324")
+        return REFaultInjector(url)
+
+
+@pytest.fixture()
+def fault_injector_client_oss_api():
+    if use_mock_proxy():
+        return ProxyServerFaultInjector(oss_cluster=True)
+    else:
+        url = os.getenv("FAULT_INJECTION_API_URL", "http://127.0.0.1:20324")
+        return REFaultInjector(url)
 
 
 @pytest.fixture()
@@ -208,8 +241,6 @@ def _get_client_maint_notifications(
         endpoint_type=endpoint_type,
     )
 
-    # Create Redis client with maintenance notifications config
-    # This will automatically create the MaintNotificationsPoolHandler
     if disable_retries:
         retry = Retry(NoBackoff(), 0)
     else:
@@ -218,6 +249,8 @@ def _get_client_maint_notifications(
     tls_enabled = True if parsed.scheme == "rediss" else False
     logging.info(f"TLS enabled: {tls_enabled}")
 
+    # Create Redis client with maintenance notifications config
+    # This will automatically create the MaintNotificationsPoolHandler
     client = Redis(
         host=host,
         port=port,
@@ -235,3 +268,76 @@ def _get_client_maint_notifications(
     logging.info(f"Client uses Protocol: {client.connection_pool.get_protocol()}")
 
     return client
+
+
+@pytest.fixture()
+def cluster_client_maint_notifications(cluster_endpoints_config):
+    return _get_cluster_client_maint_notifications(cluster_endpoints_config)
+
+
+def _get_cluster_client_maint_notifications(
+    endpoints_config,
+    protocol: int = 3,
+    enable_maintenance_notifications: bool = True,
+    endpoint_type: Optional[EndpointType] = None,
+    enable_relaxed_timeout: bool = True,
+    enable_proactive_reconnect: bool = True,
+    disable_retries: bool = False,
+    socket_timeout: Optional[float] = None,
+    host_config: Optional[str] = None,
+):
+    """Create Redis cluster client with maintenance notifications enabled."""
+    # Get credentials from the configuration
+    username = endpoints_config.get("username")
+    password = endpoints_config.get("password")
+
+    # Parse host and port from endpoints URL
+    endpoints = endpoints_config.get("endpoints", [])
+    if not endpoints:
+        raise ValueError("No endpoints found in configuration")
+
+    parsed = urlparse(endpoints[0])
+    host = parsed.hostname
+    port = parsed.port
+
+    if not host:
+        raise ValueError(f"Could not parse host from endpoint URL: {endpoints[0]}")
+
+    logging.info(f"Connecting to Redis Enterprise: {host}:{port} with user: {username}")
+
+    if disable_retries:
+        retry = Retry(NoBackoff(), 0)
+    else:
+        retry = Retry(backoff=ExponentialWithJitterBackoff(base=1, cap=10), retries=3)
+
+    tls_enabled = True if parsed.scheme == "rediss" else False
+    logging.info(f"TLS enabled: {tls_enabled}")
+
+    # Configure maintenance notifications
+    maintenance_config = MaintNotificationsConfig(
+        enabled=enable_maintenance_notifications,
+        proactive_reconnect=enable_proactive_reconnect,
+        relaxed_timeout=RELAXED_TIMEOUT if enable_relaxed_timeout else -1,
+        endpoint_type=endpoint_type,
+    )
+
+    # Create Redis cluster client with maintenance notifications config
+    client = RedisCluster(
+        host=host,
+        port=port,
+        socket_timeout=CLIENT_TIMEOUT if socket_timeout is None else socket_timeout,
+        username=username,
+        password=password,
+        ssl=tls_enabled,
+        ssl_cert_reqs="none",
+        ssl_check_hostname=False,
+        protocol=protocol,  # RESP3 required for push notifications
+        maint_notifications_config=maintenance_config,
+        retry=retry,
+    )
+    logging.info("Redis cluster client created with maintenance notifications enabled")
+    logging.info(
+        f"Cluster working with the following nodes: {[(node.name, node.server_type) for node in client.get_nodes()]}"
+    )
+
+    return client