Function get_cluster_client_maint_notifications

tests/test_scenario/conftest.py:337–409

Create Redis cluster client with maintenance notifications enabled.

(
    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,
    auth_ssl_client_certs: bool = False,
    socket_timeout: Optional[float] = None,
)

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,
    auth_ssl_client_certs: bool = False,
    socket_timeout: Optional[float] = 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=0.01, cap=1), retries=10
        )

    tls_enabled = True if parsed.scheme == "rediss" else False
    logging.info(f"TLS enabled: {tls_enabled}")

    tls_kwargs = {"ssl": tls_enabled}

    if tls_enabled:
        # Prepare SSL certificate configuration
        ssl_config = _prepare_ssl_certificates(auth_ssl_client_certs)
        tls_kwargs.update(ssl_config)

    # 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,