Repository: redis/redis-py (python)

Issue:
Add support for new redis command WAITAOF
7.2.0 adds support for the new redis command WAITAOF. We need to add support, documented [here](https://redis.io/commands/WAITAOF).

Pull request:
Add waitaof
### Pull Request check-list

_Please make sure to review and check all of these items:_

- [ ] Does `$ tox` pass with this change (including linting)?
- [ ] Do the CI tests pass with this change (enable it first in your forked repo and wait for the github action build to finish)?
- [ ] 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

closes #2684 


Patch:
diff --git a/CHANGES b/CHANGES
index 7fd510bd..040e4452 100644
--- a/CHANGES
+++ b/CHANGES
@@ -1,3 +1,5 @@
+    * 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
     * Fix `xadd` command to accept non-negative `maxlen` including 0
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..efe3a3e1 100644
--- a/redis/asyncio/connection.py
+++ b/redis/asyncio/connection.py
@@ -49,6 +49,7 @@ from redis.exceptions import (
     ModuleError,
     NoPermissionError,
     NoScriptError,
+    OutOfMemoryError,
     ReadOnlyError,
     RedisError,
     ResponseError,
@@ -174,6 +175,7 @@ class BaseParser:
         "READONLY": ReadOnlyError,
         "NOAUTH": AuthenticationError,
         "NOPERM": NoPermissionError,
+        "OOM": OutOfMemoryError,
     }
 
     def __init__(self, socket_read_size: int):
diff --git a/redis/client.py b/redis/client.py
index 9fd5b7ca..ab626ccd 100755
--- a/redis/client.py
+++ b/redis/client.py
@@ -320,7 +320,7 @@ def parse_xautoclaim(response, **options):
 def parse_xinfo_stream(response, **options):
     data = pairs_to_dict(response, decode_keys=True)
     if not options.get("full", False):
-        first = data["first-entry"]
+        first = data.get("first-entry")
         if first is not None:
             data["first-entry"] = (first[0], pairs_to_dict(first[1]))
         last = data["last-entry"]
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/commands/json/commands.py b/redis/commands/json/commands.py
index 5da9245a..f91ef918 100644
--- a/redis/commands/json/commands.py
+++ b/redis/commands/json/commands.py
@@ -261,7 +261,8 @@ class JSONCommands:
         decode_keys: Optional[bool] = False,
     ) -> Optional[str]:
         """
-        Sets or updates the JSON value at a path..
+        Merges a given JSON value into matching paths. Consequently, JSON values
+        at matching paths are updated, deleted, or expanded with new children
 
         ``decode_keys`` If set to True, the keys of ``obj`` will be decoded
         with utf-8.
diff --git a/redis/connection.py b/redis/connection.py
index 5af8928a..bf0d6dea 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,
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