Repository: redis/redis-py
Pull request #2734 (2023-05-01 14:43:04)

Extract abstract async connection class
### Pull Request check-list

_Please make sure to review and check all of these items:_

- [x] Does `$ tox` pass with this change (including linting)?
- [x] Do the CI tests pass with this change (enable it first in your forked repo and wait for the github action build to finish)?
- [x] Is the new or changed code fully tested?
- [ ] Is a documentation update included (if this change modifies existing APIs, or introduces new ones)?
- [ ] Is there an example added to the examples folder (if applicable)?
- [ ] Was the change added to CHANGES file?

_NOTE: these things are not required to open a PR and can be done
afterwards / while the PR is open._

### Description of change

Abstract asyncio connection class in analogy with the synchronous variant (https://github.com/redis/redis-py/pull/2588 and https://github.com/redis/redis-py/pull/2630).

Also this MR adds `socket_connect_timeout` to UDS connections just like TCP and asyncio UDS connections.

woutdenolf: Best to merge this PR first (adds unit tests): https://github.com/redis/redis-py/pull/2631

Patch:
diff --git a/CHANGES b/CHANGES
index 76b3393a..040e4452 100644
--- a/CHANGES
+++ b/CHANGES
@@ -1,3 +1,4 @@
+    * Fix dead weakref in sentinel connection causing ReferenceError (#2767)
     * Fix #2768, Fix KeyError: 'first-entry' in parse_xinfo_stream.
     * Fix #2749, remove unnecessary __del__ logic to close connections.
     * Fix #2754, adding a missing argument to SentinelManagedConnection
diff --git a/docs/opentelemetry.rst b/docs/opentelemetry.rst
index 96781028..d006a604 100644
--- a/docs/opentelemetry.rst
+++ b/docs/opentelemetry.rst
@@ -4,7 +4,7 @@ Integrating OpenTelemetry
 What is OpenTelemetry?
 ----------------------
 
-`OpenTelemetry <https://opentelemetry.io>`_ is an open-source observability framework for traces, metrics, and logs.
+`OpenTelemetry <https://opentelemetry.io>`_ is an open-source observability framework for traces, metrics, and logs. It is a merger of OpenCensus and OpenTracing projects hosted by Cloud Native Computing Foundation.
 
 OpenTelemetry allows developers to collect and export telemetry data in a vendor agnostic way. With OpenTelemetry, you can instrument your application once and then add or change vendors without changing the instrumentation, for example, here is a list of `popular DataDog competitors <https://uptrace.dev/get/compare/datadog-competitors.html>`_ that support OpenTelemetry.
 
@@ -97,7 +97,7 @@ See `OpenTelemetry Python Tracing API <https://uptrace.dev/opentelemetry/python-
 Uptrace
 -------
 
-Uptrace is an `open-source APM <https://uptrace.dev/get/open-source-apm.html>`_ that supports distributed tracing, metrics, and logs. You can use it to monitor applications and set up automatic alerts to receive notifications via email, Slack, Telegram, and more.
+Uptrace is an `open source APM <https://uptrace.dev/get/open-source-apm.html>`_ that supports distributed tracing, metrics, and logs. You can use it to monitor applications and set up automatic alerts to receive notifications via email, Slack, Telegram, and more.
 
 You can use Uptrace to monitor redis-py using this `GitHub example <https://github.com/redis/redis-py/tree/master/docs/examples/opentelemetry>`_ as a starting point.
 
@@ -111,9 +111,9 @@ Monitoring Redis Server performance
 
 In addition to monitoring redis-py client, you can also monitor Redis Server performance using OpenTelemetry Collector Agent.
 
-OpenTelemetry Collector is a proxy/middleman between your application and a `distributed tracing tool <https://uptrace.dev/get/compare/distributed-tracing-tools.html>`_ such as Uptrace or Jaeger. Collector receives telemetry data, processes it, and then exports the data to APM tools that can store it permanently.
+OpenTelemetry Collector is a proxy/middleman between your application and a `distributed tracing tool <https://uptrace.dev/blog/distributed-tracing-tools.html>`_ such as Uptrace or Jaeger. Collector receives telemetry data, processes it, and then exports the data to APM tools that can store it permanently.
 
-For example, you can use the Redis receiver provided by Otel Collector to `monitor Redis performance <https://uptrace.dev/opentelemetry/redis-monitoring.html>`_:
+For example, you can use the `OpenTelemetry Redis receiver <https://uptrace.dev/get/monitor/opentelemetry-redis.html>` provided by Otel Collector to monitor Redis performance:
 
 .. image:: images/opentelemetry/redis-metrics.png
   :alt: Redis metrics
@@ -123,55 +123,50 @@ See introduction to `OpenTelemetry Collector <https://uptrace.dev/opentelemetry/
 Alerting and notifications
 --------------------------
 
-Uptrace also allows you to monitor `OpenTelemetry metrics <https://uptrace.dev/opentelemetry/metrics.html>`_ using alerting rules. For example, the following rule uses the group by node expression to create an alert whenever an individual Redis shard is down:
+Uptrace also allows you to monitor `OpenTelemetry metrics <https://uptrace.dev/opentelemetry/metrics.html>`_ using alerting rules. For example, the following monitor uses the group by node expression to create an alert whenever an individual Redis shard is down:
 
 .. code-block:: python
 
-   # /etc/uptrace/uptrace.yml
-
-   alerting:
-     rules:
-       - name: Redis shard is down
-         metrics:
-           - redis_up as $redis_up
-         query:
-           - group by cluster # monitor each cluster,
-           - group by bdb # each database,
-           - group by node # and each shard
-           - $redis_up == 0
-         # shard should be down for 5 minutes to trigger an alert
-         for: 5m
+   monitors:
+     - name: Redis shard is down
+       metrics:
+         - redis_up as $redis_up
+       query:
+         - group by cluster # monitor each cluster,
+         - group by bdb # each database,
+         - group by node # and each shard
+         - $redis_up
+       min_allowed_value: 1
+       # shard should be down for 5 minutes to trigger an alert
+       for_duration: 5m
 
 You can also create queries with more complex expressions. For example, the following rule creates an alert when the keyspace hit rate is lower than 75%:
 
 .. code-block:: python
 
-   # /etc/uptrace/uptrace.yml
-
-   alerting:
-     rules:
-       - name: Redis read hit rate < 75%
-         metrics:
-           - redis_keyspace_read_hits as $hits
-           - redis_keyspace_read_misses as $misses
-         query:
-           - group by cluster
-           - group by bdb
-           - group by node
-           - $hits / ($hits + $misses) < 0.75
-         for: 5m
+   monitors:
+     - name: Redis read hit rate < 75%
+       metrics:
+         - redis_keyspace_read_hits as $hits
+         - redis_keyspace_read_misses as $misses
+       query:
+         - group by cluster
+         - group by bdb
+         - group by node
+         - $hits / ($hits + $misses) as hit_rate
+       min_allowed_value: 0.75
+       for_duration: 5m
 
 See `Alerting and Notifications <https://uptrace.dev/get/alerting.html>`_ for details.
 
 What's next?
 ------------
 
-Next, you can learn how to configure `uptrace-python <https://uptrace.dev/get/uptrace-python.html>`_ to export spans, metrics, and logs to Uptrace.
+Next, you can learn how to configure `uptrace-python <https://uptrace.dev/get/opentelemetry-python.html>`_ to export spans, metrics, and logs to Uptrace.
 
 You may also be interested in the following guides:
 
-- `OpenTelemetry Django <https://uptrace.dev/opentelemetry/instrumentations/python-django.html>`_
-- `OpenTelemetry Flask <https://uptrace.dev/opentelemetry/instrumentations/python-flask.html>`_
-- `OpenTelemetry FastAPI <https://uptrace.dev/opentelemetry/instrumentations/python-fastapi.html>`_
-- `OpenTelemetry SQLAlchemy <https://uptrace.dev/opentelemetry/instrumentations/python-sqlalchemy.html>`_
-- `OpenTelemetry instrumentations <https://uptrace.dev/opentelemetry/instrumentations/>`_
+- `OpenTelemetry Django <https://uptrace.dev/get/instrument/opentelemetry-django.html>`_
+- `OpenTelemetry Flask <https://uptrace.dev/get/instrument/instrument/opentelemetry-flask.html>`_
+- `OpenTelemetry FastAPI <https://uptrace.dev/get/instrument/opentelemetry-fastapi.html>`_
+- `OpenTelemetry SQLAlchemy <https://uptrace.dev/get/instrument/opentelemetry-sqlalchemy.html>`_
diff --git a/redis/__init__.py b/redis/__init__.py
index b8850add..d7b74edf 100644
--- a/redis/__init__.py
+++ b/redis/__init__.py
@@ -19,6 +19,7 @@ from redis.exceptions import (
     ConnectionError,
     DataError,
     InvalidResponse,
+    OutOfMemoryError,
     PubSubError,
     ReadOnlyError,
     RedisError,
@@ -72,6 +73,7 @@ __all__ = [
     "from_url",
     "default_backoff",
     "InvalidResponse",
+    "OutOfMemoryError",
     "PubSubError",
     "ReadOnlyError",
     "Redis",
diff --git a/redis/asyncio/__init__.py b/redis/asyncio/__init__.py
index bf90dde5..2a82df25 100644
--- a/redis/asyncio/__init__.py
+++ b/redis/asyncio/__init__.py
@@ -24,6 +24,7 @@ from redis.exceptions import (
     ConnectionError,
     DataError,
     InvalidResponse,
+    OutOfMemoryError,
     PubSubError,
     ReadOnlyError,
     RedisError,
@@ -47,6 +48,7 @@ __all__ = [
     "default_backoff",
     "InvalidResponse",
     "PubSubError",
+    "OutOfMemoryError",
     "ReadOnlyError",
     "Redis",
     "RedisCluster",
diff --git a/redis/asyncio/connection.py b/redis/asyncio/connection.py
index 4ba177ef..1bc3aa38 100644
--- a/redis/asyncio/connection.py
+++ b/redis/asyncio/connection.py
@@ -8,6 +8,7 @@ import ssl
 import sys
 import threading
 import weakref
+from abc import abstractmethod
 from itertools import chain
 from types import MappingProxyType
 from typing import (
@@ -49,6 +50,7 @@ from redis.exceptions import (
     ModuleError,
     NoPermissionError,
     NoScriptError,
+    OutOfMemoryError,
     ReadOnlyError,
     RedisError,
     ResponseError,
@@ -174,6 +176,7 @@ class BaseParser:
         "READONLY": ReadOnlyError,
         "NOAUTH": AuthenticationError,
         "NOPERM": NoPermissionError,
+        "OOM": OutOfMemoryError,
     }
 
     def __init__(self, socket_read_size: int):
@@ -196,7 +199,7 @@ class BaseParser:
     def on_disconnect(self):
         raise NotImplementedError()
 
-    def on_connect(self, connection: "Connection"):
+    def on_connect(self, connection: "AbstractConnection"):
         raise NotImplementedError()
 
     async def can_read_destructive(self) -> bool:
@@ -224,7 +227,7 @@ class PythonParser(BaseParser):
         self._buffer = b""
         self._chunks.clear()
 
-    def on_connect(self, connection: "Connection"):
+    def on_connect(self, connection: "AbstractConnection"):
         """Called when the stream connects"""
         self._stream = connection._reader
         if self._stream is None:
@@ -358,7 +361,7 @@ class HiredisParser(BaseParser):
         super().__init__(socket_read_size=socket_read_size)
         self._reader: Optional[hiredis.Reader] = None
 
-    def on_connect(self, connection: "Connection"):
+    def on_connect(self, connection: "AbstractConnection"):
         self._stream = connection._reader
         kwargs: _HiredisReaderArgs = {
             "protocolError": InvalidResponse,
@@ -430,25 +433,23 @@ else:
 
 
 class ConnectCallbackProtocol(Protocol):
-    def __call__(self, connection: "Connection"):
+    def __call__(self, connection: "AbstractConnection"):
         ...
 
 
 class AsyncConnectCallbackProtocol(Protocol):
-    async def __call__(self, connection: "Connection"):
+    async def __call__(self, connection: "AbstractConnection"):
         ...
 
 
 ConnectCallbackT = Union[ConnectCallbackProtocol, AsyncConnectCallbackProtocol]
 
 
-class Connection:
-    """Manages TCP communication to and from a Redis server"""
+class AbstractConnection:
+    """Manages communication to and from a Redis server"""
 
     __slots__ = (
         "pid",
-        "host",
-        "port",
         "db",
         "username",
         "client_name",
@@ -456,9 +457,6 @@ class Connection:
         "password",
         "socket_timeout",
         "socket_connect_timeout",
-        "socket_keepalive",
-        "socket_keepalive_options",
-        "socket_type",
         "redis_connect_func",
         "retry_on_timeout",
         "retry_on_error",
@@ -480,15 +478,10 @@ class Connection:
     def __init__(
         self,
         *,
-        host: str = "localhost",
-        port: Union[str, int] = 6379,
         db: Union[str, int] = 0,
         password: Optional[str] = None,
         socket_timeout: Optional[float] = None,
         socket_connect_timeout: Optional[float] = None,
-        socket_keepalive: bool = False,
-        socket_keepalive_options: Optional[Mapping[int, Union[int, bytes]]] = None,
-        socket_type: int = 0,
         retry_on_timeout: bool = False,
         retry_on_error: Union[list, _Sentinel] = SENTINEL,
         encoding: str = "utf-8",
@@ -512,18 +505,15 @@ class Connection:
                 "2. 'credential_provider'"
             )
         self.pid = os.getpid()
-        self.host = host
-        self.port = int(port)
         self.db = db
         self.client_name = client_name
         self.credential_provider = credential_provider
         self.password = password
         self.username = username
         self.socket_timeout = socket_timeout
-        self.socket_connect_timeout = socket_connect_timeout or socket_timeout or None
-        self.socket_keepalive = socket_keepalive
-        self.socket_keepalive_options = socket_keepalive_options or {}
-        self.socket_type = socket_type
+        if socket_connect_timeout is None:
+            socket_connect_timeout = socket_timeout
+        self.socket_connect_timeout = socket_connect_timeout
         self.retry_on_timeout = retry_on_timeout
         if retry_on_error is SENTINEL:
             retry_on_error = []
@@ -544,7 +534,6 @@ class Connection:
             self.retry = Retry(NoBackoff(), 0)
         self.health_check_interval = health_check_interval
         self.next_health_check: float = -1
-        self.ssl_context: Optional[RedisSSLContext] = None
         self.encoder = encoder_class(encoding, encoding_errors, decode_responses)
         self.redis_connect_func = redis_connect_func
         self._reader: Optional[asyncio.StreamReader] = None
@@ -558,11 +547,9 @@ class Connection:
         repr_args = ",".join((f"{k}={v}" for k, v in self.repr_pieces()))
         return f"{self.__class__.__name__}<{repr_args}>"
 
+    @abstractmethod
     def repr_pieces(self):
-        pieces = [("host", self.host), ("port", self.port), ("db", self.db)]
-        if self.client_name:
-            pieces.append(("client_name", self.client_name))
-        return pieces
+        pass
 
     @property
     def is_connected(self):
@@ -621,51 +608,17 @@ class Connection:
             if task and inspect.isawaitable(task):
                 await task
 
+    @abstractmethod
     async def _connect(self):
-        """Create a TCP socket connection"""
-        async with async_timeout(self.socket_connect_timeout):
-            reader, writer = await asyncio.open_connection(
-                host=self.host,
-                port=self.port,
-                ssl=self.ssl_context.get() if self.ssl_context else None,
-            )
-        self._reader = reader
-        self._writer = writer
-        sock = writer.transport.get_extra_info("socket")
-        if sock:
-            sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
-            try:
-                # TCP_KEEPALIVE
-                if self.socket_keepalive:
-                    sock.setsockopt(socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1)
-                    for k, v in self.socket_keepalive_options.items():
-                        sock.setsockopt(socket.SOL_TCP, k, v)
+        pass
 
-            except (OSError, TypeError):
-                # `socket_keepalive_options` might contain invalid options
-                # causing an error. Do not leave the connection open.
-                writer.close()
-                raise
+    @abstractmethod
+    def _host_error(self) -> str:
+        pass
 
-    def _error_message(self, exception):
-        # args for socket.error can either be (errno, "message")
-        # or just "message"
-        if not exception.args:
-            # asyncio has a bug where on Connection reset by peer, the
-            # exception is not instanciated, so args is empty. This is the
-            # workaround.
-            # See: https://github.com/redis/redis-py/issues/2237
-            # See: https://github.com/python/cpython/issues/94061
-            return (
-                f"Error connecting to {self.host}:{self.port}. Connection reset by peer"
-            )
-        elif len(exception.args) == 1:
-            return f"Error connecting to {self.host}:{self.port}. {exception.args[0]}."
-        else:
-            return (
-                f"Error {exception.args[0]} connecting to {self.host}:{self.port}. "
-                f"{exception.args[0]}."
-            )
+    @abstractmethod
+    def _error_message(self, exception: BaseException) -> str:
+        pass
 
     async def on_connect(self) -> None:
         """Initialize the connection, authenticate and select a database"""
@@ -806,9 +759,8 @@ class Connection:
             return await self._parser.can_read_destructive()
         except OSError as e:
             await self.disconnect(nowait=True)
-            raise ConnectionError(
-                f"Error while reading from {self.host}:{self.port}: {e.args}"
-            )
+            host_error = self._host_error()
+            raise ConnectionError(f"Error while reading from {host_error}: {e.args}")
 
     async def read_response(
         self,
@@ -819,6 +771,7 @@ class Connection:
     ):
         """Read the response from a previously sent command"""
         read_timeout = timeout if timeout is not None else self.socket_timeout
+        host_error = self._host_error()
         try:
             if read_timeout is not None:
                 async with async_timeout(read_timeout):
@@ -836,13 +789,11 @@ class Connection:
             # it was a self.socket_timeout error.
             if disconnect_on_error:
                 await self.disconnect(nowait=True)
-            raise TimeoutError(f"Timeout reading from {self.host}:{self.port}")
+            raise TimeoutError(f"Timeout reading from {host_error}")
         except OSError as e:
             if disconnect_on_error:
                 await self.disconnect(nowait=True)
-            raise ConnectionError(
-                f"Error while reading from {self.host}:{self.port} : {e.args}"
-            )
+            raise ConnectionError(f"Error while reading from {host_error} : {e.args}")
         except BaseException:
             # Also by default close in case of BaseException.  A lot of code
             # relies on this behaviour when doing Command/Response pairs.
@@ -936,7 +887,90 @@ class Connection:
         return output
 
 
+class Connection(AbstractConnection):
+    "Manages TCP communication to and from a Redis server"
+
+    def __init__(
+        self,
+        *,
+        host: str = "localhost",
+        port: Union[str, int] = 6379,
+        socket_keepalive: bool = False,
+        socket_keepalive_options: Optional[Mapping[int, Union[int, bytes]]] = None,
+        socket_type: int = 0,
+        **kwargs,
+    ):
+        self.host = host
+        self.port = int(port)
+        self.socket_keepalive = socket_keepalive
+        self.socket_keepalive_options = socket_keepalive_options or {}
+        self.socket_type = socket_type
+        super().__init__(**kwargs)
+
+    def repr_pieces(self):
+        pieces = [("host", self.host), ("port", self.port), ("db", self.db)]
+        if self.client_name:
+            pieces.append(("client_name", self.client_name))
+        return pieces
+
+    def _connection_arguments(self) -> Mapping:
+        return {"host": self.host, "port": self.port}
+
+    async def _connect(self):
+        """Create a TCP socket connection"""
+        async with async_timeout(self.socket_connect_timeout):
+            reader, writer = await asyncio.open_connection(
+                **self._connection_arguments()
+            )
+        self._reader = reader
+        self._writer = writer
+        sock = writer.transport.get_extra_info("socket")
+        if sock:
+            sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
+            try:
+                # TCP_KEEPALIVE
+                if self.socket_keepalive:
+                    sock.setsockopt(socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1)
+                    for k, v in self.socket_keepalive_options.items():
+                        sock.setsockopt(socket.SOL_TCP, k, v)
+
+            except (OSError, TypeError):
+                # `socket_keepalive_options` might contain invalid options
+                # causing an error. Do not leave the connection open.
+                writer.close()
+                raise
+
+    def _host_error(self) -> str:
+        return f"{self.host}:{self.port}"
+
+    def _error_message(self, exception: BaseException) -> str:
+        # args for socket.error can either be (errno, "message")
+        # or just "message"
+
+        host_error = self._host_error()
+
+        if not exception.args:
+            # asyncio has a bug where on Connection reset by peer, the
+            # exception is not instanciated, so args is empty. This is the
+            # workaround.
+            # See: https://github.com/redis/redis-py/issues/2237
+            # See: https://github.com/python/cpython/issues/94061
+            return f"Error connecting to {host_error}. Connection reset by peer"
+        elif len(exception.args) == 1:
+            return f"Error connecting to {host_error}. {exception.args[0]}."
+        else:
+            return (
+                f"Error {exception.args[0]} connecting to {host_error}. "
+                f"{exception.args[0]}."
+            )
+
+
 class SSLConnection(Connection):
+    """Manages SSL connections to and from the Redis server(s).
+    This class extends the Connection class, adding SSL functionality, and making
+    use of ssl.SSLContext (https://docs.python.org/3/library/ssl.html#ssl.SSLContext)
+    """
+
     def __init__(
         self,
         ssl_keyfile: Optional[str] = None,
@@ -947,7 +981,6 @@ class SSLConnection(Connection):
         ssl_check_hostname: bool = False,
         **kwargs,
     ):
-        super().__init__(**kwargs)
         self.ssl_context: RedisSSLContext = RedisSSLContext(
             keyfile=ssl_keyfile,
             certfile=ssl_certfile,
@@ -956,6 +989,12 @@ class SSLConnection(Connection):
             ca_data=ssl_ca_data,
             check_hostname=ssl_check_hostname,
         )
+        super().__init__(**kwargs)
+
+    def _connection_arguments(self) -> Mapping:
+        kwargs = super()._connection_arguments()
+        kwargs["ssl"] = self.ssl_context.get()
+        return kwargs
 
     @property
     def keyfile(self):
@@ -1035,77 +1074,12 @@ class RedisSSLContext:
         return self.context
 
 
-class UnixDomainSocketConnection(Connection):  # lgtm [py/missing-call-to-init]
-    def __init__(
-        self,
-        *,
-        path: str = "",
-        db: Union[str, int] = 0,
-        username: Optional[str] = None,
-        password: Optional[str] = None,
-        socket_timeout: Optional[float] = None,
-        socket_connect_timeout: Optional[float] = None,
-        encoding: str = "utf-8",
-        encoding_errors: str = "strict",
-        decode_responses: bool = False,
-        retry_on_timeout: bool = False,
-        retry_on_error: Union[list, _Sentinel] = SENTINEL,
-        parser_class: Type[BaseParser] = DefaultParser,
-        socket_read_size: int = 65536,
-        health_check_interval: float = 0.0,
-        client_name: str = None,
-        retry: Optional[Retry] = None,
-        redis_connect_func=None,
-        credential_provider: Optional[CredentialProvider] = None,
-    ):
-        """
-        Initialize a new UnixDomainSocketConnection.
-        To specify a retry policy, first set `retry_on_timeout` to `True`
-        then set `retry` to a valid `Retry` object
-        """
-        if (username or password) and credential_provider is not None:
-            raise DataError(
-                "'username' and 'password' cannot be passed along with 'credential_"
-                "provider'. Please provide only one of the following arguments: \n"
-                "1. 'password' and (optional) 'username'\n"
-                "2. 'credential_provider'"
-            )
-        self.pid = os.getpid()
+class UnixDomainSocketConnection(AbstractConnection):
+    "Manages UDS communication to and from a Redis server"
+
+    def __init__(self, *, path: str = "", **kwargs):
         self.path = path
-        self.db = db
-        self.client_name = client_name
-        self.credential_provider = credential_provider
-        self.password = password
-        self.username = username
-        self.socket_timeout = socket_timeout
-        self.socket_connect_timeout = socket_connect_timeout or socket_timeout or None
-        self.retry_on_timeout = retry_on_timeout
-        if retry_on_error is SENTINEL:
-            retry_on_error = []
-        if retry_on_timeout:
-            retry_on_error.append(TimeoutError)
-        self.retry_on_error = retry_on_error
-        if retry_on_error:
-            if retry is None:
-                self.retry = Retry(NoBackoff(), 1)
-            else:
-                # deep-copy the Retry object as it is mutable
-                self.retry = copy.deepcopy(retry)
-            # Update the retry's supported errors with the specified errors
-            self.retry.update_supported_errors(retry_on_error)
-        else:
-            self.retry = Retry(NoBackoff(), 0)
-        self.health_check_interval = health_check_interval
-        self.next_health_check = -1
-        self.redis_connect_func = redis_connect_func
-        self.encoder = Encoder(encoding, encoding_errors, decode_responses)
-        self._sock = None
-        self._reader = None
-        self._writer = None
-        self._socket_read_size = socket_read_size
-        self.set_parser(parser_class)
-        self._connect_callbacks = []
-        self._buffer_cutoff = 6000
+        super().__init__(**kwargs)
 
     def repr_pieces(self) -> Iterable[Tuple[str, Union[str, int]]]:
         pieces = [("path", self.path), ("db", self.db)]
@@ -1120,15 +1094,21 @@ class UnixDomainSocketConnection(Connection):  # lgtm [py/missing-call-to-init]
         self._writer = writer
         await self.on_connect()
 
-    def _error_message(self, exception):
+    def _host_error(self) -> str:
+        return self.host
+
+    def _error_message(self, exception: BaseException) -> str:
         # args for socket.error can either be (errno, "message")
         # or just "message"
+        host_error = self._host_error()
         if len(exception.args) == 1:
-            return f"Error connecting to unix socket: {self.path}. {exception.args[0]}."
+            return (
+                f"Error connecting to unix socket: {host_error}. {exception.args[0]}."
+            )
         else:
             return (
                 f"Error {exception.args[0]} connecting to unix socket: "
-                f"{self.path}. {exception.args[1]}."
+                f"{host_error}. {exception.args[1]}."
             )
 
 
@@ -1160,7 +1140,7 @@ URL_QUERY_ARGUMENT_PARSERS: Mapping[str, Callable[..., object]] = MappingProxyTy
 class ConnectKwargs(TypedDict, total=False):
     username: str
     password: str
-    connection_class: Type[Connection]
+    connection_class: Type[AbstractConnection]
     host: str
     port: int
     db: int
@@ -1282,7 +1262,7 @@ class ConnectionPool:
 
     def __init__(
         self,
-        connection_class: Type[Connection] = Connection,
+        connection_class: Type[AbstractConnection] = Connection,
         max_connections: Optional[int] = None,
         **connection_kwargs,
     ):
@@ -1305,8 +1285,8 @@ class ConnectionPool:
         self._fork_lock = threading.Lock()
         self._lock = asyncio.Lock()
         self._created_connections: int
-        self._available_connections: List[Connection]
-        self._in_use_connections: Set[Connection]
+        self._available_connections: List[AbstractConnection]
+        self._in_use_connections: Set[AbstractConnection]
         self.reset()  # lgtm [py/init-calls-subclass]
         self.encoder_class = self.connection_kwargs.get("encoder_class", Encoder)
 
@@ -1429,7 +1409,7 @@ class ConnectionPool:
         self._created_connections += 1
         return self.connection_class(**self.connection_kwargs)
 
-    async def release(self, connection: Connection):
+    async def release(self, connection: AbstractConnection):
         """Releases the connection back to the pool"""
         self._checkpid()
         async with self._lock:
@@ -1450,7 +1430,7 @@ class ConnectionPool:
                 await connection.disconnect()
                 return
 
-    def owns_connection(self, connection: Connection):
+    def owns_connection(self, connection: AbstractConnection):
         return connection.pid == self.pid
 
     async def disconnect(self, inuse_connections: bool = True):
@@ -1464,7 +1444,7 @@ class ConnectionPool:
         self._checkpid()
         async with self._lock:
             if inuse_connections:
-                connections: Iterable[Connection] = chain(
+                connections: Iterable[AbstractConnection] = chain(
                     self._available_connections, self._in_use_connections
                 )
             else:
@@ -1522,14 +1502,14 @@ class BlockingConnectionPool(ConnectionPool):
         self,
         max_connections: int = 50,
         timeout: Optional[int] = 20,
-        connection_class: Type[Connection] = Connection,
+        connection_class: Type[AbstractConnection] = Connection,
         queue_class: Type[asyncio.Queue] = asyncio.LifoQueue,
         **connection_kwargs,
     ):
 
         self.queue_class = queue_class
         self.timeout = timeout
-        self._connections: List[Connection]
+        self._connections: List[AbstractConnection]
         super().__init__(
             connection_class=connection_class,
             max_connections=max_connections,
@@ -1619,7 +1599,7 @@ class BlockingConnectionPool(ConnectionPool):
 
         return connection
 
-    async def release(self, connection: Connection):
+    async def release(self, connection: AbstractConnection):
         """Releases the connection back to the pool."""
         # Make sure we haven't changed process.
         self._checkpid()
diff --git a/redis/cluster.py b/redis/cluster.py
index 2ab173de..be8e4623 100644
--- a/redis/cluster.py
+++ b/redis/cluster.py
@@ -237,6 +237,7 @@ class AbstractRedisCluster:
                 "SLOWLOG LEN",
                 "SLOWLOG RESET",
                 "WAIT",
+                "WAITAOF",
                 "SAVE",
                 "MEMORY PURGE",
                 "MEMORY MALLOC-STATS",
diff --git a/redis/commands/core.py b/redis/commands/core.py
index af5384e9..392ddb54 100644
--- a/redis/commands/core.py
+++ b/redis/commands/core.py
@@ -1341,6 +1341,21 @@ class ManagementCommands(CommandsProtocol):
         """
         return self.execute_command("WAIT", num_replicas, timeout, **kwargs)
 
+    def waitaof(
+        self, num_local: int, num_replicas: int, timeout: int, **kwargs
+    ) -> ResponseT:
+        """
+        This command blocks the current client until all previous write
+        commands by that client are acknowledged as having been fsynced
+        to the AOF of the local Redis and/or at least the specified number
+        of replicas.
+
+        For more information see https://redis.io/commands/waitaof
+        """
+        return self.execute_command(
+            "WAITAOF", num_local, num_replicas, timeout, **kwargs
+        )
+
     def hello(self):
         """
         This function throws a NotImplementedError since it is intentionally
diff --git a/redis/connection.py b/redis/connection.py
index 5af8928a..bec456c9 100644
--- a/redis/connection.py
+++ b/redis/connection.py
@@ -28,6 +28,7 @@ from redis.exceptions import (
     ModuleError,
     NoPermissionError,
     NoScriptError,
+    OutOfMemoryError,
     ReadOnlyError,
     RedisError,
     ResponseError,
@@ -149,6 +150,7 @@ class BaseParser:
             MODULE_UNLOAD_NOT_POSSIBLE_ERROR: ModuleError,
             **NO_AUTH_SET_ERROR,
         },
+        "OOM": OutOfMemoryError,
         "WRONGPASS": AuthenticationError,
         "EXECABORT": ExecAbortError,
         "LOADING": BusyLoadingError,
@@ -592,6 +594,8 @@ class AbstractConnection:
         self,
         db=0,
         password=None,
+        socket_timeout=None,
+        socket_connect_timeout=None,
         retry_on_timeout=False,
         retry_on_error=SENTINEL,
         encoding="utf-8",
@@ -627,6 +631,10 @@ class AbstractConnection:
         self.credential_provider = credential_provider
         self.password = password
         self.username = username
+        self.socket_timeout = socket_timeout
+        if socket_connect_timeout is None:
+            socket_connect_timeout = socket_timeout
+        self.socket_connect_timeout = socket_connect_timeout
         self.retry_on_timeout = retry_on_timeout
         if retry_on_error is SENTINEL:
             retry_on_error = []
@@ -939,8 +947,6 @@ class Connection(AbstractConnection):
         self,
         host="localhost",
         port=6379,
-        socket_timeout=None,
-        socket_connect_timeout=None,
         socket_keepalive=False,
         socket_keepalive_options=None,
         socket_type=0,
@@ -948,8 +954,6 @@ class Connection(AbstractConnection):
     ):
         self.host = host
         self.port = int(port)
-        self.socket_timeout = socket_timeout
-        self.socket_connect_timeout = socket_connect_timeout or socket_timeout
         self.socket_keepalive = socket_keepalive
         self.socket_keepalive_options = socket_keepalive_options or {}
         self.socket_type = socket_type
@@ -1170,9 +1174,8 @@ class SSLConnection(Connection):
 class UnixDomainSocketConnection(AbstractConnection):
     "Manages UDS communication to and from a Redis server"
 
-    def __init__(self, path="", socket_timeout=None, **kwargs):
+    def __init__(self, path="", **kwargs):
         self.path = path
-        self.socket_timeout = socket_timeout
         super().__init__(**kwargs)
 
     def repr_pieces(self):
@@ -1184,8 +1187,9 @@ class UnixDomainSocketConnection(AbstractConnection):
     def _connect(self):
         "Create a Unix domain socket connection"
         sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
-        sock.settimeout(self.socket_timeout)
+        sock.settimeout(self.socket_connect_timeout)
         sock.connect(self.path)
+        sock.settimeout(self.socket_timeout)
         return sock
 
     def _host_error(self):
diff --git a/redis/exceptions.py b/redis/exceptions.py
index 8a8bf423..7cf15a7d 100644
--- a/redis/exceptions.py
+++ b/redis/exceptions.py
@@ -49,6 +49,18 @@ class NoScriptError(ResponseError):
     pass
 
 
+class OutOfMemoryError(ResponseError):
+    """
+    Indicates the database is full. Can only occur when either:
+      * Redis maxmemory-policy=noeviction
+      * Redis maxmemory-policy=volatile* and there are no evictable keys
+
+    For more information see `Memory optimization in Redis <https://redis.io/docs/management/optimization/memory-optimization/#memory-allocation>`_. # noqa
+    """
+
+    pass
+
+
 class ExecAbortError(ResponseError):
     pass
 
@@ -131,6 +143,7 @@ class AskError(ResponseError):
     pertain to this hash slot, but only if the key in question exists,
     otherwise the query is forwarded using a -ASK redirection to the node that
     is target of the migration.
+
     src node: MIGRATING to dst node
         get > ASK error
         ask dst node > ASKING command
diff --git a/redis/sentinel.py b/redis/sentinel.py
index f9f8f1c3..0ba179b9 100644
--- a/redis/sentinel.py
+++ b/redis/sentinel.py
@@ -78,6 +78,54 @@ class SentinelManagedSSLConnection(SentinelManagedConnection, SSLConnection):
     pass
 
 
+class SentinelConnectionPoolProxy:
+    def __init__(
+        self,
+        connection_pool,
+        is_master,
+        check_connection,
+        service_name,
+        sentinel_manager,
+    ):
+        self.connection_pool_ref = weakref.ref(connection_pool)
+        self.is_master = is_master
+        self.check_connection = check_connection
+        self.service_name = service_name
+        self.sentinel_manager = sentinel_manager
+        self.reset()
+
+    def reset(self):
+        self.master_address = None
+        self.slave_rr_counter = None
+
+    def get_master_address(self):
+        master_address = self.sentinel_manager.discover_master(self.service_name)
+        if self.is_master and self.master_address != master_address:
+            self.master_address = master_address
+            # disconnect any idle connections so that they reconnect
+            # to the new master the next time that they are used.
+            connection_pool = self.connection_pool_ref()
+            if connection_pool is not None:
+                connection_pool.disconnect(inuse_connections=False)
+        return master_address
+
+    def rotate_slaves(self):
+        slaves = self.sentinel_manager.discover_slaves(self.service_name)
+        if slaves:
+            if self.slave_rr_counter is None:
+                self.slave_rr_counter = random.randint(0, len(slaves) - 1)
+            for _ in range(len(slaves)):
+                self.slave_rr_counter = (self.slave_rr_counter + 1) % len(slaves)
+                slave = slaves[self.slave_rr_counter]
+                yield slave
+        # Fallback to the master connection
+        try:
+            yield self.get_master_address()
+        except MasterNotFoundError:
+            pass
+        raise SlaveNotFoundError(f"No slave found for {self.service_name!r}")
+
+
 class SentinelConnectionPool(ConnectionPool):
     """
     Sentinel backed connection pool.
@@ -95,8 +143,15 @@ class SentinelConnectionPool(ConnectionPool):
         )
         self.is_master = kwargs.pop("is_master", True)
         self.check_connection = kwargs.pop("check_connection", False)
+        self.proxy = SentinelConnectionPoolProxy(
+            connection_pool=self,
+            is_master=self.is_master,
+            check_connection=self.check_connection,
+            service_name=service_name,
+            sentinel_manager=sentinel_manager,
+        )
         super().__init__(**kwargs)
-        self.connection_kwargs["connection_pool"] = weakref.proxy(self)
+        self.connection_kwargs["connection_pool"] = self.proxy
         self.service_name = service_name
         self.sentinel_manager = sentinel_manager
 
@@ -106,8 +161,11 @@ class SentinelConnectionPool(ConnectionPool):
 
     def reset(self):
         super().reset()
-        self.master_address = None
-        self.slave_rr_counter = None
+        self.proxy.reset()
+
+    @property
+    def master_address(self):
+        return self.proxy.master_address
 
     def owns_connection(self, connection):
         check = not self.is_master or (
@@ -117,31 +175,11 @@ class SentinelConnectionPool(ConnectionPool):
         return check and parent.owns_connection(connection)
 
     def get_master_address(self):
-        master_address = self.sentinel_manager.discover_master(self.service_name)
-        if self.is_master:
-            if self.master_address != master_address:
-                self.master_address = master_address
-                # disconnect any idle connections so that they reconnect
-                # to the new master the next time that they are used.
-                self.disconnect(inuse_connections=False)
-        return master_address
+        return self.proxy.get_master_address()
 
     def rotate_slaves(self):
         "Round-robin slave balancer"
-        slaves = self.sentinel_manager.discover_slaves(self.service_name)
-        if slaves:
-            if self.slave_rr_counter is None:
-                self.slave_rr_counter = random.randint(0, len(slaves) - 1)
-            for _ in range(len(slaves)):
-                self.slave_rr_counter = (self.slave_rr_counter + 1) % len(slaves)
-                slave = slaves[self.slave_rr_counter]
-                yield slave
-        # Fallback to the master connection
-        try:
-            yield self.get_master_address()
-        except MasterNotFoundError:
-            pass
-        raise SlaveNotFoundError(f"No slave found for {self.service_name!r}")
+        return self.proxy.rotate_slaves()
 
 
 class Sentinel(SentinelCommands):


Test patch:
diff --git a/tests/ssl_utils.py b/tests/ssl_utils.py
new file mode 100644
index 00000000..50937638
--- /dev/null
+++ b/tests/ssl_utils.py
@@ -0,0 +1,14 @@
+import os
+
+
+def get_ssl_filename(name):
+    root = os.path.join(os.path.dirname(__file__), "..")
+    cert_dir = os.path.abspath(os.path.join(root, "docker", "stunnel", "keys"))
+    if not os.path.isdir(cert_dir):  # github actions package validation case
+        cert_dir = os.path.abspath(
+            os.path.join(root, "..", "docker", "stunnel", "keys")
+        )
+        if not os.path.isdir(cert_dir):
+            raise IOError(f"No SSL certificates found. They should be in {cert_dir}")
+
+    return os.path.join(cert_dir, name)
diff --git a/tests/test_asyncio/test_cluster.py b/tests/test_asyncio/test_cluster.py
index 17aa879b..c41d4a21 100644
--- a/tests/test_asyncio/test_cluster.py
+++ b/tests/test_asyncio/test_cluster.py
@@ -1,7 +1,6 @@
 import asyncio
 import binascii
 import datetime
-import os
 import warnings
 from typing import Any, Awaitable, Callable, Dict, List, Optional, Type, Union
 from urllib.parse import urlparse
@@ -36,6 +35,7 @@ from tests.conftest import (
     skip_unless_arch_bits,
 )
 
+from ..ssl_utils import get_ssl_filename
 from .compat import mock
 
 pytestmark = pytest.mark.onlycluster
@@ -2744,17 +2744,8 @@ class TestSSL:
     appropriate port.
     """
 
-    ROOT = os.path.join(os.path.dirname(__file__), "../..")
-    CERT_DIR = os.path.abspath(os.path.join(ROOT, "docker", "stunnel", "keys"))
-    if not os.path.isdir(CERT_DIR):  # github actions package validation case
-        CERT_DIR = os.path.abspath(
-            os.path.join(ROOT, "..", "docker", "stunnel", "keys")
-        )
-        if not os.path.isdir(CERT_DIR):
-            raise IOError(f"No SSL certificates found. They should be in {CERT_DIR}")
-
-    SERVER_CERT = os.path.join(CERT_DIR, "server-cert.pem")
-    SERVER_KEY = os.path.join(CERT_DIR, "server-key.pem")
+    SERVER_CERT = get_ssl_filename("server-cert.pem")
+    SERVER_KEY = get_ssl_filename("server-key.pem")
 
     @pytest_asyncio.fixture()
     def create_client(self, request: FixtureRequest) -> Callable[..., RedisCluster]:
diff --git a/tests/test_asyncio/test_commands.py b/tests/test_asyncio/test_commands.py
index 89a85258..c0259680 100644
--- a/tests/test_asyncio/test_commands.py
+++ b/tests/test_asyncio/test_commands.py
@@ -461,6 +461,20 @@ class TestRedisCommands:
         with pytest.raises(TypeError):
             await r.client_no_touch()
 
+    @skip_if_server_version_lt("7.2.0")
+    @pytest.mark.onlycluster
+    async def test_waitaof(self, r):
+        # must return a list of 2 elements
+        assert len(await r.waitaof(0, 0, 0)) == 2
+        assert len(await r.waitaof(1, 0, 0)) == 2
+        assert len(await r.waitaof(1, 0, 1000)) == 2
+
+        # value is out of range, value must between 0 and 1
+        with pytest.raises(exceptions.ResponseError):
+            await r.waitaof(2, 0, 0)
+        with pytest.raises(exceptions.ResponseError):
+            await r.waitaof(-1, 0, 0)
+
     async def test_config_get(self, r: redis.Redis):
         data = await r.config_get()
         assert "maxmemory" in data
diff --git a/tests/test_asyncio/test_connect.py b/tests/test_asyncio/test_connect.py
new file mode 100644
index 00000000..8e3209fd
--- /dev/null
+++ b/tests/test_asyncio/test_connect.py
@@ -0,0 +1,145 @@
+import asyncio
+import logging
+import re
+import socket
+import ssl
+
+import pytest
+
+from redis.asyncio.connection import (
+    Connection,
+    SSLConnection,
+    UnixDomainSocketConnection,
+)
+
+from ..ssl_utils import get_ssl_filename
+
+_logger = logging.getLogger(__name__)
+
+
+_CLIENT_NAME = "test-suite-client"
+_CMD_SEP = b"\r\n"
+_SUCCESS_RESP = b"+OK" + _CMD_SEP
+_ERROR_RESP = b"-ERR" + _CMD_SEP
+_SUPPORTED_CMDS = {f"CLIENT SETNAME {_CLIENT_NAME}": _SUCCESS_RESP}
+
+
+@pytest.fixture
+def tcp_address():
+    with socket.socket() as sock:
+        sock.bind(("127.0.0.1", 0))
+        return sock.getsockname()
+
+
+@pytest.fixture
+def uds_address(tmpdir):
+    return tmpdir / "uds.sock"
+
+
+async def test_tcp_connect(tcp_address):
+    host, port = tcp_address
+    conn = Connection(host=host, port=port, client_name=_CLIENT_NAME, socket_timeout=10)
+    await _assert_connect(conn, tcp_address)
+
+
+async def test_uds_connect(uds_address):
+    path = str(uds_address)
+    conn = UnixDomainSocketConnection(
+        path=path, client_name=_CLIENT_NAME, socket_timeout=10
+    )
+    await _assert_connect(conn, path)
+
+
+@pytest.mark.ssl
+async def test_tcp_ssl_connect(tcp_address):
+    host, port = tcp_address
+    certfile = get_ssl_filename("server-cert.pem")
+    keyfile = get_ssl_filename("server-key.pem")
+    conn = SSLConnection(
+        host=host,
+        port=port,
+        client_name=_CLIENT_NAME,
+        ssl_ca_certs=certfile,
+        socket_timeout=10,
+    )
+    await _assert_connect(conn, tcp_address, certfile=certfile, keyfile=keyfile)
+
+
+async def _assert_connect(conn, server_address, certfile=None, keyfile=None):
+    stop_event = asyncio.Event()
+    finished = asyncio.Event()
+
+    async def _handler(reader, writer):
+        try:
+            return await _redis_request_handler(reader, writer, stop_event)
+        finally:
+            finished.set()
+
+    if isinstance(server_address, str):
+        server = await asyncio.start_unix_server(_handler, path=server_address)
+    elif certfile:
+        host, port = server_address
+        context = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH)
+        context.minimum_version = ssl.TLSVersion.TLSv1_2
+        context.load_cert_chain(certfile=certfile, keyfile=keyfile)
+        server = await asyncio.start_server(_handler, host=host, port=port, ssl=context)
+    else:
+        host, port = server_address
+        server = await asyncio.start_server(_handler, host=host, port=port)
+
+    async with server as aserver:
+        await aserver.start_serving()
+        try:
+            await conn.connect()
+            await conn.disconnect()
+        finally:
+            stop_event.set()
+            aserver.close()
+            await aserver.wait_closed()
+            await finished.wait()
+
+
+async def _redis_request_handler(reader, writer, stop_event):
+    buffer = b""
+    command = None
+    command_ptr = None
+    fragment_length = None
+    while not stop_event.is_set() or buffer:
+        _logger.info(str(stop_event.is_set()))
+        try:
+            buffer += await asyncio.wait_for(reader.read(1024), timeout=0.5)
+        except TimeoutError:
+            continue
+        if not buffer:
+            continue
+        parts = re.split(_CMD_SEP, buffer)
+        buffer = parts[-1]
+        for fragment in parts[:-1]:
+            fragment = fragment.decode()
+            _logger.info("Command fragment: %s", fragment)
+
+            if fragment.startswith("*") and command is None:
+                command = [None for _ in range(int(fragment[1:]))]
+                command_ptr = 0
+                fragment_length = None
+                continue
+
+            if fragment.startswith("$") and command[command_ptr] is None:
+                fragment_length = int(fragment[1:])
+                continue
+
+            assert len(fragment) == fragment_length
+            command[command_ptr] = fragment
+            command_ptr += 1
+
+            if command_ptr < len(command):
+                continue
+
+            command = " ".join(command)
+            _logger.info("Command %s", command)
+            resp = _SUPPORTED_CMDS.get(command, _ERROR_RESP)
+            _logger.info("Response from %s", resp)
+            writer.write(resp)
+            await writer.drain()
+            command = None
+    _logger.info("Exit handler")
diff --git a/tests/test_asyncio/test_connection_pool.py b/tests/test_asyncio/test_connection_pool.py
index 92499e2c..24d99021 100644
--- a/tests/test_asyncio/test_connection_pool.py
+++ b/tests/test_asyncio/test_connection_pool.py
@@ -606,10 +606,18 @@ class TestConnection:
     @skip_if_server_version_lt("2.8.8")
     @skip_if_redis_enterprise()
     async def test_read_only_error(self, r):
-        """READONLY errors get turned in ReadOnlyError exceptions"""
+        """READONLY errors get turned into ReadOnlyError exceptions"""
         with pytest.raises(redis.ReadOnlyError):
             await r.execute_command("DEBUG", "ERROR", "READONLY blah blah")
 
+    @skip_if_redis_enterprise()
+    async def test_oom_error(self, r):
+        """OOM errors get turned into OutOfMemoryError exceptions"""
+        with pytest.raises(redis.OutOfMemoryError):
+            # note: don't use the DEBUG OOM command since it's not the same
+            # as the db being full
+            await r.execute_command("DEBUG", "ERROR", "OOM blah blah")
+
     def test_connect_from_url_tcp(self):
         connection = redis.Redis.from_url("redis://localhost")
         pool = connection.connection_pool
diff --git a/tests/test_commands.py b/tests/test_commands.py
index 4e2ac5b0..2213e81f 100644
--- a/tests/test_commands.py
+++ b/tests/test_commands.py
@@ -704,6 +704,20 @@ class TestRedisCommands:
         with pytest.raises(TypeError):
             r.client_no_touch()
 
+    @pytest.mark.onlycluster
+    @skip_if_server_version_lt("7.2.0")
+    def test_waitaof(self, r):
+        # must return a list of 2 elements
+        assert len(r.waitaof(0, 0, 0)) == 2
+        assert len(r.waitaof(1, 0, 0)) == 2
+        assert len(r.waitaof(1, 0, 1000)) == 2
+
+        # value is out of range, value must between 0 and 1
+        with pytest.raises(exceptions.ResponseError):
+            r.waitaof(2, 0, 0)
+        with pytest.raises(exceptions.ResponseError):
+            r.waitaof(-1, 0, 0)
+
     @pytest.mark.onlynoncluster
     @skip_if_server_version_lt("3.2.0")
     def test_client_reply(self, r, r_timeout):
diff --git a/tests/test_connect.py b/tests/test_connect.py
new file mode 100644
index 00000000..b4ec7020
--- /dev/null
+++ b/tests/test_connect.py
@@ -0,0 +1,185 @@
+import logging
+import re
+import socket
+import socketserver
+import ssl
+import threading
+
+import pytest
+
+from redis.connection import Connection, SSLConnection, UnixDomainSocketConnection
+
+from .ssl_utils import get_ssl_filename
+
+_logger = logging.getLogger(__name__)
+
+
+_CLIENT_NAME = "test-suite-client"
+_CMD_SEP = b"\r\n"
+_SUCCESS_RESP = b"+OK" + _CMD_SEP
+_ERROR_RESP = b"-ERR" + _CMD_SEP
+_SUPPORTED_CMDS = {f"CLIENT SETNAME {_CLIENT_NAME}": _SUCCESS_RESP}
+
+
+@pytest.fixture
+def tcp_address():
+    with socket.socket() as sock:
+        sock.bind(("127.0.0.1", 0))
+        return sock.getsockname()
+
+
+@pytest.fixture
+def uds_address(tmpdir):
+    return tmpdir / "uds.sock"
+
+
+def test_tcp_connect(tcp_address):
+    host, port = tcp_address
+    conn = Connection(host=host, port=port, client_name=_CLIENT_NAME, socket_timeout=10)
+    _assert_connect(conn, tcp_address)
+
+
+def test_uds_connect(uds_address):
+    path = str(uds_address)
+    conn = UnixDomainSocketConnection(path, client_name=_CLIENT_NAME, socket_timeout=10)
+    _assert_connect(conn, path)
+
+
+@pytest.mark.ssl
+def test_tcp_ssl_connect(tcp_address):
+    host, port = tcp_address
+    certfile = get_ssl_filename("server-cert.pem")
+    keyfile = get_ssl_filename("server-key.pem")
+    conn = SSLConnection(
+        host=host,
+        port=port,
+        client_name=_CLIENT_NAME,
+        ssl_ca_certs=certfile,
+        socket_timeout=10,
+    )
+    _assert_connect(conn, tcp_address, certfile=certfile, keyfile=keyfile)
+
+
+def _assert_connect(conn, server_address, certfile=None, keyfile=None):
+    if isinstance(server_address, str):
+        server = _RedisUDSServer(server_address, _RedisRequestHandler)
+    else:
+        server = _RedisTCPServer(
+            server_address, _RedisRequestHandler, certfile=certfile, keyfile=keyfile
+        )
+    with server as aserver:
+        t = threading.Thread(target=aserver.serve_forever)
+        t.start()
+        try:
+            aserver.wait_online()
+            conn.connect()
+            conn.disconnect()
+        finally:
+            aserver.stop()
+            t.join(timeout=5)
+
+
+class _RedisTCPServer(socketserver.TCPServer):
+    def __init__(self, *args, certfile=None, keyfile=None, **kw) -> None:
+        self._ready_event = threading.Event()
+        self._stop_requested = False
+        self._certfile = certfile
+        self._keyfile = keyfile
+        super().__init__(*args, **kw)
+
+    def service_actions(self):
+        self._ready_event.set()
+
+    def wait_online(self):
+        self._ready_event.wait()
+
+    def stop(self):
+        self._stop_requested = True
+        self.shutdown()
+
+    def is_serving(self):
+        return not self._stop_requested
+
+    def get_request(self):
+        if self._certfile is None:
+            return super().get_request()
+        newsocket, fromaddr = self.socket.accept()
+        connstream = ssl.wrap_socket(
+            newsocket,
+            server_side=True,
+            certfile=self._certfile,
+            keyfile=self._keyfile,
+            ssl_version=ssl.PROTOCOL_TLSv1_2,
+        )
+        return connstream, fromaddr
+
+
+class _RedisUDSServer(socketserver.UnixStreamServer):
+    def __init__(self, *args, **kw) -> None:
+        self._ready_event = threading.Event()
+        self._stop_requested = False
+        super().__init__(*args, **kw)
+
+    def service_actions(self):
+        self._ready_event.set()
+
+    def wait_online(self):
+        self._ready_event.wait()
+
+    def stop(self):
+        self._stop_requested = True
+        self.shutdown()
+
+    def is_serving(self):
+        return not self._stop_requested
+
+
+class _RedisRequestHandler(socketserver.StreamRequestHandler):
+    def setup(self):
+        _logger.info("%s connected", self.client_address)
+
+    def finish(self):
+        _logger.info("%s disconnected", self.client_address)
+
+    def handle(self):
+        buffer = b""
+        command = None
+        command_ptr = None
+        fragment_length = None
+        while self.server.is_serving() or buffer:
+            try:
+                buffer += self.request.recv(1024)
+            except socket.timeout:
+                continue
+            if not buffer:
+                continue
+            parts = re.split(_CMD_SEP, buffer)
+            buffer = parts[-1]
+            for fragment in parts[:-1]:
+                fragment = fragment.decode()
+                _logger.info("Command fragment: %s", fragment)
+
+                if fragment.startswith("*") and command is None:
+                    command = [None for _ in range(int(fragment[1:]))]
+                    command_ptr = 0
+                    fragment_length = None
+                    continue
+
+                if fragment.startswith("$") and command[command_ptr] is None:
+                    fragment_length = int(fragment[1:])
+                    continue
+
+                assert len(fragment) == fragment_length
+                command[command_ptr] = fragment
+                command_ptr += 1
+
+                if command_ptr < len(command):
+                    continue
+
+                command = " ".join(command)
+                _logger.info("Command %s", command)
+                resp = _SUPPORTED_CMDS.get(command, _ERROR_RESP)
+                _logger.info("Response %s", resp)
+                self.request.sendall(resp)
+                command = None
+        _logger.info("Exit handler")
diff --git a/tests/test_connection_pool.py b/tests/test_connection_pool.py
index e8a42692..155bffe5 100644
--- a/tests/test_connection_pool.py
+++ b/tests/test_connection_pool.py
@@ -528,10 +528,17 @@ class TestConnection:
     @skip_if_server_version_lt("2.8.8")
     @skip_if_redis_enterprise()
     def test_read_only_error(self, r):
-        "READONLY errors get turned in ReadOnlyError exceptions"
+        "READONLY errors get turned into ReadOnlyError exceptions"
         with pytest.raises(redis.ReadOnlyError):
             r.execute_command("DEBUG", "ERROR", "READONLY blah blah")
 
+    def test_oom_error(self, r):
+        "OOM errors get turned into OutOfMemoryError exceptions"
+        with pytest.raises(redis.OutOfMemoryError):
+            # note: don't use the DEBUG OOM command since it's not the same
+            # as the db being full
+            r.execute_command("DEBUG", "ERROR", "OOM blah blah")
+
     def test_connect_from_url_tcp(self):
         connection = redis.Redis.from_url("redis://localhost")
         pool = connection.connection_pool
diff --git a/tests/test_sentinel.py b/tests/test_sentinel.py
index 8542a0bf..e5e3d26f 100644
--- a/tests/test_sentinel.py
+++ b/tests/test_sentinel.py
@@ -98,6 +98,15 @@ def test_discover_master_error(sentinel):
         sentinel.discover_master("xxx")
 
 
+@pytest.mark.onlynoncluster
+def test_dead_pool(sentinel):
+    master = sentinel.master_for("mymaster", db=9)
+    conn = master.connection_pool.get_connection("_")
+    conn.disconnect()
+    del master
+    conn.connect()
+
+
 @pytest.mark.onlynoncluster
 def test_discover_master_sentinel_down(cluster, sentinel, master_ip):
     # Put first sentinel 'foo' down
diff --git a/tests/test_ssl.py b/tests/test_ssl.py
index ed38a316..c1a981d3 100644
--- a/tests/test_ssl.py
+++ b/tests/test_ssl.py
@@ -1,4 +1,3 @@
-import os
 import socket
 import ssl
 from urllib.parse import urlparse
@@ -9,6 +8,7 @@ import redis
 from redis.exceptions import ConnectionError, RedisError
 
 from .conftest import skip_if_cryptography, skip_if_nocryptography
+from .ssl_utils import get_ssl_filename
 
 
 @pytest.mark.ssl
@@ -19,17 +19,8 @@ class TestSSL:
     and connecting to the appropriate port.
     """
 
-    ROOT = os.path.join(os.path.dirname(__file__), "..")
-    CERT_DIR = os.path.abspath(os.path.join(ROOT, "docker", "stunnel", "keys"))
-    if not os.path.isdir(CERT_DIR):  # github actions package validation case
-        CERT_DIR = os.path.abspath(
-            os.path.join(ROOT, "..", "docker", "stunnel", "keys")
-        )
-        if not os.path.isdir(CERT_DIR):
-            raise IOError(f"No SSL certificates found. They should be in {CERT_DIR}")
-
-    SERVER_CERT = os.path.join(CERT_DIR, "server-cert.pem")
-    SERVER_KEY = os.path.join(CERT_DIR, "server-key.pem")
+    SERVER_CERT = get_ssl_filename("server-cert.pem")
+    SERVER_KEY = get_ssl_filename("server-key.pem")
 
     def test_ssl_with_invalid_cert(self, request):
         ssl_url = request.config.option.redis_ssl_url