Repository: redis/redis-py (python)

Issue:
Getting ReadOnly error when writing data to AWS elastic cache while the cluster is vertically scaled
I have a python application that writes some data to AWS elastic cache cluster at a regular interval.

Here's an example script that simulates the functionality of my application. It also replicates the error that I have been facing.

```py
from datetime import datetime
import redis
import time
import sys
import random

class Config:
    default_host = "localhost"
    master_host = "xxx.xx.0001.xxxx.cache.amazonaws.com"
    replica_host = "xxx.xx.0001.xxxx.cache.amazonaws.com"
    redis_db = 8
    socket_conn_timeout = 10
    request_delay_sec = 0.1

def get_redis_client():
    return redis.Redis(
        host=Config.master_host,
        db=Config.redis_db,
        socket_connect_timeout=Config.socket_conn_timeout,
    )

def get_random_key_value():
    val = time.time()
    key = "test_key_" + str(random.randint(0, 100))
    return key, val

r = get_redis_client()

r.flushdb()

flag = False

while True:
    try:
        if flag:
            print("beat:", time.time())
        r.set(*get_random_key_value())
        time.sleep(Config.request_delay_sec)
    except redis.RedisError as re:
        print(datetime.now(), "Error:", type(re), re)
        flag = True
        # sys.exit()
    except KeyboardInterrupt:
        print("Stopping loop execution")
        sys.exit()
```

Here are the environment details of my application

 - OS(Ubuntu 18.04)
 - Python(v 3.7.0)
 - Redis-py(v 3.5.3)
 - AWS elastic cache(cluster mode disabled, 1 master node, 1 read replica)

When I scale my AWS elastic cache cluster vertically and the above script is running, I get the following error for few seconds while the cluster scaling-up is in process and then it goes away.

```bash
<class 'redis.exceptions.ReadOnlyError'> You can't write against a read only replica.
```

AWS docs also states that during vertical scaling process some inconsistencies may occur because of data syncing(https://docs.aws.amazon.com/AmazonElastiCache/latest/red-ug/redis-cluster-vertical-scaling.html)

why this error occurs during the scale up process? how can it be fixed?

I tried the same thing with a golang script and it works perfectly fine.

Pull request description:
Support for specifying error types with retry
### 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?
- [ X] Is a documentation update included (if this change modifies existing APIs, or introduces new ones)?
- [X ] Is there an example added to the examples folder (if applicable)?

_NOTE: these things are not required to open a PR and can be done
afterwards / while the PR is open._

### Description of change

The current client implementation supports only retry on TimeoutError.
Added support for passing specific error types to use with the retry mechanism.
The 'retry_on_timeout' argument has not changed for backward compatibility.
Closes #1764 

Patch:
diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md
index 04f989a3..0f9ca0b1 100644
--- a/CONTRIBUTING.md
+++ b/CONTRIBUTING.md
@@ -32,7 +32,7 @@ Here's how to get started with your code contribution:
 
 1.  Create your own fork of redis-py
 2.  Do the changes in your fork
-3.  
+3.
     *Create a virtualenv and install the development dependencies from the dev_requirements.txt file:*
 
         a.  python -m venv .venv
@@ -69,12 +69,12 @@ configuration](https://redis.io/topics/sentinel).
 
 ## Testing
 
-Call `invoke tests` to run all tests, or `invoke all-tests` to run linters 
-tests as well. With the 'tests' and 'all-tests' targets, all Redis and 
-RedisCluster tests will be run. 
+Call `invoke tests` to run all tests, or `invoke all-tests` to run linters
+tests as well. With the 'tests' and 'all-tests' targets, all Redis and
+RedisCluster tests will be run.
 
-It is possible to run only Redis client tests (with cluster mode disabled) by 
-using `invoke standalone-tests`; similarly, RedisCluster tests can be run by using 
+It is possible to run only Redis client tests (with cluster mode disabled) by
+using `invoke standalone-tests`; similarly, RedisCluster tests can be run by using
 `invoke cluster-tests`.
 
 Each run of tox starts and stops the various dockers required. Sometimes
@@ -85,9 +85,7 @@ tests against multiple versions of python. Feel free to test your
 changes against all the python versions supported, as declared by the
 tox.ini file (eg: tox -e py39). If you have the various python versions
 on your desktop, you can run *tox* by itself, to test all supported
-versions. Alternatively, as your source code is mounted in the
-**lots-of-pythons** docker, you can start exploring from there, with all
-supported python versions!
+versions.
 
 ### Docker Tips
 
diff --git a/docs/redismodules.rst b/docs/redismodules.rst
index 0cb8c496..53a50081 100644
--- a/docs/redismodules.rst
+++ b/docs/redismodules.rst
@@ -44,7 +44,7 @@ These are the commands for interacting with the `RedisJSON module <https://redis
 RediSearch Commands
 *******************
 
-These are the commands for interacting with the `RediSearch module <https://redisjson.io>`_. Below is a brief example, as well as documentation on the commands themselves.
+These are the commands for interacting with the `RediSearch module <https://redisearch.io>`_. Below is a brief example, as well as documentation on the commands themselves.
 
 **Create a search index, and display its information**
 
@@ -64,7 +64,7 @@ These are the commands for interacting with the `RediSearch module <https://redi
 RedisGraph Commands
 *******************
 
-These are the commands for interacting with the `RedisGraph module <https://redisjson.io>`_. Below is a brief example, as well as documentation on the commands themselves.
+These are the commands for interacting with the `RedisGraph module <https://redisgraph.io>`_. Below is a brief example, as well as documentation on the commands themselves.
 
 **Create a graph, adding two nodes**
 
@@ -97,7 +97,7 @@ These are the commands for interacting with the `RedisGraph module <https://redi
 RedisBloom Commands
 *******************
 
-These are the commands for interacting with the `RedisBloom module <https://redisjson.io>`_. Below is a brief example, as well as documentation on the commands themselves.
+These are the commands for interacting with the `RedisBloom module <https://redisbloom.io>`_. Below is a brief example, as well as documentation on the commands themselves.
 
 **Create and add to a bloom filter**
 
diff --git a/redis/client.py b/redis/client.py
index 3ae133c3..0236f20a 100755
--- a/redis/client.py
+++ b/redis/client.py
@@ -768,6 +768,7 @@ class Redis(RedisModuleCommands, CoreCommands, SentinelCommands):
         "STRALGO": parse_stralgo,
         "PUBSUB NUMSUB": parse_pubsub_numsub,
         "RANDOMKEY": lambda r: r and r or None,
+        "RESET": str_if_bytes,
         "SCAN": parse_scan,
         "SCRIPT EXISTS": lambda r: list(map(bool, r)),
         "SCRIPT FLUSH": bool_ok,
@@ -868,6 +869,7 @@ class Redis(RedisModuleCommands, CoreCommands, SentinelCommands):
         errors=None,
         decode_responses=False,
         retry_on_timeout=False,
+        retry_on_error=[],
         ssl=False,
         ssl_keyfile=None,
         ssl_certfile=None,
@@ -886,8 +888,10 @@ class Redis(RedisModuleCommands, CoreCommands, SentinelCommands):
     ):
         """
         Initialize a new Redis client.
-        To specify a retry policy, first set `retry_on_timeout` to `True`
-        then set `retry` to a valid `Retry` object
+        To specify a retry policy for specific errors, first set
+        `retry_on_error` to a list of the error/s to retry on, then set
+        `retry` to a valid `Retry` object.
+        To retry on TimeoutError, `retry_on_timeout` can also be set to `True`.
         """
         if not connection_pool:
             if charset is not None:
@@ -904,7 +908,8 @@ class Redis(RedisModuleCommands, CoreCommands, SentinelCommands):
                     )
                 )
                 encoding_errors = errors
-
+            if retry_on_timeout is True:
+                retry_on_error.append(TimeoutError)
             kwargs = {
                 "db": db,
                 "username": username,
@@ -913,7 +918,7 @@ class Redis(RedisModuleCommands, CoreCommands, SentinelCommands):
                 "encoding": encoding,
                 "encoding_errors": encoding_errors,
                 "decode_responses": decode_responses,
-                "retry_on_timeout": retry_on_timeout,
+                "retry_on_error": retry_on_error,
                 "retry": copy.deepcopy(retry),
                 "max_connections": max_connections,
                 "health_check_interval": health_check_interval,
@@ -950,6 +955,7 @@ class Redis(RedisModuleCommands, CoreCommands, SentinelCommands):
                             "ssl_ca_certs": ssl_ca_certs,
                             "ssl_check_hostname": ssl_check_hostname,
                             "ssl_password": ssl_password,
+                            "ssl_ca_path": ssl_ca_path,
                         }
                     )
             connection_pool = ConnectionPool(**kwargs)
@@ -1144,11 +1150,14 @@ class Redis(RedisModuleCommands, CoreCommands, SentinelCommands):
     def _disconnect_raise(self, conn, error):
         """
         Close the connection and raise an exception
-        if retry_on_timeout is not set or the error
-        is not a TimeoutError
+        if retry_on_error is not set or the error
+        is not one of the specified error types
         """
         conn.disconnect()
-        if not (conn.retry_on_timeout and isinstance(error, TimeoutError)):
+        if (
+            conn.retry_on_error is None
+            or isinstance(error, tuple(conn.retry_on_error)) is False
+        ):
             raise error
 
     # COMMAND EXECUTION AND PROTOCOL PARSING
@@ -1529,6 +1538,8 @@ class PubSub:
         with a message handler, the handler is invoked instead of a parsed
         message being returned.
         """
+        if response is None:
+            return None
         message_type = str_if_bytes(response[0])
         if message_type == "pmessage":
             message = {
diff --git a/redis/cluster.py b/redis/cluster.py
index b1adeb73..0c2fc715 100644
--- a/redis/cluster.py
+++ b/redis/cluster.py
@@ -996,9 +996,11 @@ class RedisCluster(RedisClusterCommands):
                 self.reinitialize_counter += 1
                 if self._should_reinitialized():
                     self.nodes_manager.initialize()
+                    # Reset the counter
+                    self.reinitialize_counter = 0
                 else:
                     self.nodes_manager.update_moved_exception(e)
-                    moved = True
+                moved = True
             except TryAgainError:
                 log.exception("TryAgainError")
 
@@ -1320,6 +1322,7 @@ class NodesManager:
         tmp_slots = {}
         disagreements = []
         startup_nodes_reachable = False
+        fully_covered = False
         kwargs = self.connection_kwargs
         for startup_node in self.startup_nodes.values():
             try:
@@ -1431,6 +1434,12 @@ class NodesManager:
                                     f'slots cache: {", ".join(disagreements)}'
                                 )
 
+            fully_covered = self.check_slots_coverage(tmp_slots)
+            if fully_covered:
+                # Don't need to continue to the next startup node if all
+                # slots are covered
+                break
+
         if not startup_nodes_reachable:
             raise RedisClusterException(
                 "Redis Cluster cannot be connected. Please provide at least "
@@ -1440,7 +1449,6 @@ class NodesManager:
         # Create Redis connections to all nodes
         self.create_redis_connections(list(tmp_nodes_cache.values()))
 
-        fully_covered = self.check_slots_coverage(tmp_slots)
         # Check if the slots are not fully covered
         if not fully_covered and self._require_full_coverage:
             # Despite the requirement that the slots be covered, there
@@ -1478,6 +1486,8 @@ class NodesManager:
         self.default_node = self.get_nodes_by_server_type(PRIMARY)[0]
         # Populate the startup nodes with all discovered nodes
         self.populate_startup_nodes(self.nodes_cache.values())
+        # If initialize was called after a MovedError, clear it
+        self._moved_exception = None
 
     def close(self):
         self.default_node = None
diff --git a/redis/commands/core.py b/redis/commands/core.py
index 835ea612..0823315a 100644
--- a/redis/commands/core.py
+++ b/redis/commands/core.py
@@ -479,6 +479,104 @@ class ManagementCommands:
         """
         return self.execute_command("CLIENT ID", **kwargs)
 
+    def client_tracking_on(
+        self,
+        clientid=None,
+        prefix=[],
+        bcast=False,
+        optin=False,
+        optout=False,
+        noloop=False,
+    ):
+        """
+        Turn on the tracking mode.
+        For more information about the options look at client_tracking func.
+
+        See https://redis.io/commands/client-tracking
+        """
+        return self.client_tracking(
+            True, clientid, prefix, bcast, optin, optout, noloop
+        )
+
+    def client_tracking_off(
+        self,
+        clientid=None,
+        prefix=[],
+        bcast=False,
+        optin=False,
+        optout=False,
+        noloop=False,
+    ):
+        """
+        Turn off the tracking mode.
+        For more information about the options look at client_tracking func.
+
+        See https://redis.io/commands/client-tracking
+        """
+        return self.client_tracking(
+            False, clientid, prefix, bcast, optin, optout, noloop
+        )
+
+    def client_tracking(
+        self,
+        on=True,
+        clientid=None,
+        prefix=[],
+        bcast=False,
+        optin=False,
+        optout=False,
+        noloop=False,
+        **kwargs,
+    ):
+        """
+        Enables the tracking feature of the Redis server, that is used
+        for server assisted client side caching.
+
+        ``on`` indicate for tracking on or tracking off. The dafualt is on.
+
+        ``clientid`` send invalidation messages to the connection with
+        the specified ID.
+
+        ``bcast`` enable tracking in broadcasting mode. In this mode
+        invalidation messages are reported for all the prefixes
+        specified, regardless of the keys requested by the connection.
+
+        ``optin``  when broadcasting is NOT active, normally don't track
+        keys in read only commands, unless they are called immediately
+        after a CLIENT CACHING yes command.
+
+        ``optout`` when broadcasting is NOT active, normally track keys in
+        read only commands, unless they are called immediately after a
+        CLIENT CACHING no command.
+
+        ``noloop`` don't send notifications about keys modified by this
+        connection itself.
+
+        ``prefix``  for broadcasting, register a given key prefix, so that
+        notifications will be provided only for keys starting with this string.
+
+        See https://redis.io/commands/client-tracking
+        """
+
+        if len(prefix) != 0 and bcast is False:
+            raise DataError("Prefix can only be used with bcast")
+
+        pieces = ["ON"] if on else ["OFF"]
+        if clientid is not None:
+            pieces.extend(["REDIRECT", clientid])
+        for p in prefix:
+            pieces.extend(["PREFIX", p])
+        if bcast:
+            pieces.append("BCAST")
+        if optin:
+            pieces.append("OPTIN")
+        if optout:
+            pieces.append("OPTOUT")
+        if noloop:
+            pieces.append("NOLOOP")
+
+        return self.execute_command("CLIENT TRACKING", *pieces)
+
     def client_trackinginfo(self, **kwargs):
         """
         Returns the information about the current client connection's
@@ -510,16 +608,28 @@ class ManagementCommands:
             args.append(b"ERROR")
         return self.execute_command(*args, **kwargs)
 
-    def client_pause(self, timeout, **kwargs):
+    def client_pause(self, timeout, all=True, **kwargs):
         """
         Suspend all the Redis clients for the specified amount of time
         :param timeout: milliseconds to pause clients
 
         For more information check https://redis.io/commands/client-pause
-        """
+        :param all: If true (default) all client commands are blocked.
+             otherwise, clients are only blocked if they attempt to execute
+             a write command.
+             For the WRITE mode, some commands have special behavior:
+                 EVAL/EVALSHA: Will block client for all scripts.
+                 PUBLISH: Will block client.
+                 PFCOUNT: Will block client.
+                 WAIT: Acknowledgments will be delayed, so this command will
+                 appear blocked.
+        """
+        args = ["CLIENT PAUSE", str(timeout)]
         if not isinstance(timeout, int):
             raise DataError("CLIENT PAUSE timeout must be an integer")
-        return self.execute_command("CLIENT PAUSE", str(timeout), **kwargs)
+        if not all:
+            args.append("WRITE")
+        return self.execute_command(*args, **kwargs)
 
     def client_unpause(self, **kwargs):
         """
@@ -707,6 +817,13 @@ class ManagementCommands:
         else:
             return self.execute_command("LOLWUT", **kwargs)
 
+    def reset(self):
+        """Perform a full reset on the connection's server side contenxt.
+
+        See: https://redis.io/commands/reset
+        """
+        return self.execute_command("RESET")
+
     def migrate(
         self,
         host,
diff --git a/redis/commands/json/commands.py b/redis/commands/json/commands.py
index e7f07b61..a132b8ee 100644
--- a/redis/commands/json/commands.py
+++ b/redis/commands/json/commands.py
@@ -1,3 +1,6 @@
+import os
+from json import JSONDecodeError, loads
+
 from deprecated import deprecated
 
 from redis.exceptions import DataError
@@ -213,6 +216,54 @@ class JSONCommands:
             pieces.append("XX")
         return self.execute_command("JSON.SET", *pieces)
 
+    def set_file(self, name, path, file_name, nx=False, xx=False, decode_keys=False):
+        """
+        Set the JSON value at key ``name`` under the ``path`` to the content
+        of the json file ``file_name``.
+
+        ``nx`` if set to True, set ``value`` only if it does not exist.
+        ``xx`` if set to True, set ``value`` only if it exists.
+        ``decode_keys`` If set to True, the keys of ``obj`` will be decoded
+        with utf-8.
+
+        """
+
+        with open(file_name, "r") as fp:
+            file_content = loads(fp.read())
+
+        return self.set(name, path, file_content, nx=nx, xx=xx, decode_keys=decode_keys)
+
+    def set_path(self, json_path, root_folder, nx=False, xx=False, decode_keys=False):
+        """
+        Iterate over ``root_folder`` and set each JSON file to a value
+        under ``json_path`` with the file name as the key.
+
+        ``nx`` if set to True, set ``value`` only if it does not exist.
+        ``xx`` if set to True, set ``value`` only if it exists.
+        ``decode_keys`` If set to True, the keys of ``obj`` will be decoded
+        with utf-8.
+
+        """
+        set_files_result = {}
+        for root, dirs, files in os.walk(root_folder):
+            for file in files:
+                file_path = os.path.join(root, file)
+                try:
+                    file_name = file_path.rsplit(".")[0]
+                    self.set_file(
+                        file_name,
+                        json_path,
+                        file_path,
+                        nx=nx,
+                        xx=xx,
+                        decode_keys=decode_keys,
+                    )
+                    set_files_result[file_path] = True
+                except JSONDecodeError:
+                    set_files_result[file_path] = False
+
+        return set_files_result
+
     def strlen(self, name, path=None):
         """Return the length of the string JSON value under ``path`` at key
         ``name``.
diff --git a/redis/connection.py b/redis/connection.py
index 3fe8543c..a349a0f2 100755
--- a/redis/connection.py
+++ b/redis/connection.py
@@ -513,6 +513,7 @@ class Connection:
         socket_keepalive_options=None,
         socket_type=0,
         retry_on_timeout=False,
+        retry_on_error=[],
         encoding="utf-8",
         encoding_errors="strict",
         decode_responses=False,
@@ -526,8 +527,10 @@ class Connection:
     ):
         """
         Initialize a new Connection.
-        To specify a retry policy, first set `retry_on_timeout` to `True`
-        then set `retry` to a valid `Retry` object
+        To specify a retry policy for specific errors, first set
+        `retry_on_error` to a list of the error/s to retry on, then set
+        `retry` to a valid `Retry` object.
+        To retry on TimeoutError, `retry_on_timeout` can also be set to `True`.
         """
         self.pid = os.getpid()
         self.host = host
@@ -543,11 +546,17 @@ class Connection:
         self.socket_type = socket_type
         self.retry_on_timeout = retry_on_timeout
         if retry_on_timeout:
+            # Add TimeoutError to the errors list to retry on
+            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_erros(retry_on_error)
         else:
             self.retry = Retry(NoBackoff(), 0)
         self.health_check_interval = health_check_interval
@@ -969,6 +978,7 @@ class UnixDomainSocketConnection(Connection):
         encoding_errors="strict",
         decode_responses=False,
         retry_on_timeout=False,
+        retry_on_error=[],
         parser_class=DefaultParser,
         socket_read_size=65536,
         health_check_interval=0,
@@ -978,8 +988,10 @@ class UnixDomainSocketConnection(Connection):
     ):
         """
         Initialize a new UnixDomainSocketConnection.
-        To specify a retry policy, first set `retry_on_timeout` to `True`
-        then set `retry` to a valid `Retry` object
+        To specify a retry policy for specific errors, first set
+        `retry_on_error` to a list of the error/s to retry on, then set
+        `retry` to a valid `Retry` object.
+        To retry on TimeoutError, `retry_on_timeout` can also be set to `True`.
         """
         self.pid = os.getpid()
         self.path = path
@@ -990,11 +1002,17 @@ class UnixDomainSocketConnection(Connection):
         self.socket_timeout = socket_timeout
         self.retry_on_timeout = retry_on_timeout
         if retry_on_timeout:
+            # Add TimeoutError to the errors list to retry on
+            retry_on_error.append(TimeoutError)
+        self.retry_on_error = retry_on_error
+        if self.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_erros(retry_on_error)
         else:
             self.retry = Retry(NoBackoff(), 0)
         self.health_check_interval = health_check_interval
@@ -1052,6 +1070,7 @@ URL_QUERY_ARGUMENT_PARSERS = {
     "socket_connect_timeout": float,
     "socket_keepalive": to_bool,
     "retry_on_timeout": to_bool,
+    "retry_on_error": list,
     "max_connections": int,
     "health_check_interval": int,
     "ssl_check_hostname": to_bool,
diff --git a/redis/retry.py b/redis/retry.py
index 75504c77..6147fbd9 100644
--- a/redis/retry.py
+++ b/redis/retry.py
@@ -19,6 +19,14 @@ class Retry:
         self._retries = retries
         self._supported_errors = supported_errors
 
+    def update_supported_erros(self, specified_errors: list):
+        """
+        Updates the supported errors with the specified error types
+        """
+        self._supported_errors = tuple(
+            set(self._supported_errors + tuple(specified_errors))
+        )
+
     def call_with_retry(self, do, fail):
         """
         Execute an operation that might fail and returns its result, or
diff --git a/tox.ini b/tox.ini
index b574d17b..bceab0bf 100644
--- a/tox.ini
+++ b/tox.ini
@@ -73,12 +73,6 @@ ports =
     36379:6379/tcp
 healtcheck_cmd = python -c "import socket;print(True) if 0 == socket.socket(socket.AF_INET, socket.SOCK_STREAM).connect_ex(('127.0.0.1',36379)) else False"
 
-[docker:lots-of-pythons]
-name = lots-of-pythons
-image = redisfab/lots-of-pythons:latest
-volumes =
-    bind:rw:{toxinidir}:/data
-
 [docker:redis_cluster]
 name = redis_cluster
 image = redisfab/redis-py-cluster:6.2.6-buster


Test patch:
diff --git a/tests/test_commands.py b/tests/test_commands.py
index b8dc69f9..09121617 100644
--- a/tests/test_commands.py
+++ b/tests/test_commands.py
@@ -396,6 +396,28 @@ class TestRedisCommands:
         assert len(res) > 2
         assert "prefixes" in res
 
+    @pytest.mark.onlynoncluster
+    @skip_if_server_version_lt("6.0.0")
+    def test_client_tracking(self, r, r2):
+
+        # simple case
+        assert r.client_tracking_on()
+        assert r.client_tracking_off()
+
+        # id based
+        client_id = r.client_id()
+        assert r.client_tracking_on(client_id)
+        assert r.client_tracking_off(client_id)
+
+        # id exists
+        client_id = r2.client_id()
+        assert r.client_tracking_on(client_id)
+        assert r2.client_tracking_off(client_id)
+
+        # now with some prefixes
+        with pytest.raises(exceptions.DataError):
+            assert r.client_tracking_on(prefix=["foo", "bar", "blee"])
+
     @pytest.mark.onlynoncluster
     @skip_if_server_version_lt("5.0.0")
     def test_client_unblock(self, r):
@@ -557,6 +579,13 @@ class TestRedisCommands:
         with pytest.raises(exceptions.RedisError):
             r.client_pause(timeout="not an integer")
 
+    @skip_if_server_version_lt("6.2.0")
+    def test_client_pause_all(self, r, r2):
+        assert r.client_pause(1, all=False)
+        assert r2.set("foo", "bar")
+        assert r2.get("foo") == b"bar"
+        assert r.get("foo") == b"bar"
+
     @pytest.mark.onlynoncluster
     @skip_if_server_version_lt("6.2.0")
     @skip_if_redis_enterprise()
@@ -640,6 +669,11 @@ class TestRedisCommands:
         lolwut = r.lolwut(5, 6, 7, 8).decode("utf-8")
         assert "Redis ver." in lolwut
 
+    @pytest.mark.onlynoncluster
+    @skip_if_server_version_lt("6.2.0")
+    def test_reset(self, r):
+        assert r.reset() == "RESET"
+
     def test_object(self, r):
         r["a"] = "foo"
         assert isinstance(r.object("refcount", "a"), int)
diff --git a/tests/test_json.py b/tests/test_json.py
index a99547dd..6980e673 100644
--- a/tests/test_json.py
+++ b/tests/test_json.py
@@ -1392,3 +1392,41 @@ def test_custom_decoder(client):
     assert client.exists("foo") == 0
     assert not isinstance(cj.__encoder__, json.JSONEncoder)
     assert not isinstance(cj.__decoder__, json.JSONDecoder)
+
+
+@pytest.mark.redismod
+def test_set_file(client):
+    import json
+    import tempfile
+
+    obj = {"hello": "world"}
+    jsonfile = tempfile.NamedTemporaryFile(suffix=".json")
+    with open(jsonfile.name, "w+") as fp:
+        fp.write(json.dumps(obj))
+
+    nojsonfile = tempfile.NamedTemporaryFile()
+    nojsonfile.write(b"Hello World")
+
+    assert client.json().set_file("test", Path.rootPath(), jsonfile.name)
+    assert client.json().get("test") == obj
+    with pytest.raises(json.JSONDecodeError):
+        client.json().set_file("test2", Path.rootPath(), nojsonfile.name)
+
+
+@pytest.mark.redismod
+def test_set_path(client):
+    import json
+    import tempfile
+
+    root = tempfile.mkdtemp()
+    sub = tempfile.mkdtemp(dir=root)
+    jsonfile = tempfile.mktemp(suffix=".json", dir=sub)
+    nojsonfile = tempfile.mktemp(dir=root)
+
+    with open(jsonfile, "w+") as fp:
+        fp.write(json.dumps({"hello": "world"}))
+    open(nojsonfile, "a+").write("hello")
+
+    result = {jsonfile: True, nojsonfile: False}
+    assert client.json().set_path(Path.rootPath(), root) == result
+    assert client.json().get(jsonfile.rsplit(".")[0]) == {"hello": "world"}
diff --git a/tests/test_retry.py b/tests/test_retry.py
index c4650bc6..00947872 100644
--- a/tests/test_retry.py
+++ b/tests/test_retry.py
@@ -1,10 +1,20 @@
+from unittest.mock import patch
+
 import pytest
 
 from redis.backoff import NoBackoff
+from redis.client import Redis
 from redis.connection import Connection, UnixDomainSocketConnection
-from redis.exceptions import ConnectionError
+from redis.exceptions import (
+    BusyLoadingError,
+    ConnectionError,
+    ReadOnlyError,
+    TimeoutError,
+)
 from redis.retry import Retry
 
+from .conftest import _get_client
+
 
 class BackoffMock:
     def __init__(self):
@@ -39,6 +49,37 @@ class TestConnectionConstructorWithRetry:
         assert isinstance(c.retry, Retry)
         assert c.retry._retries == retries
 
+    @pytest.mark.parametrize("Class", [Connection, UnixDomainSocketConnection])
+    def test_retry_on_error(self, Class):
+        c = Class(retry_on_error=[ReadOnlyError])
+        assert c.retry_on_error == [ReadOnlyError]
+        assert isinstance(c.retry, Retry)
+        assert c.retry._retries == 1
+
+    @pytest.mark.parametrize("Class", [Connection, UnixDomainSocketConnection])
+    def test_retry_on_error_empty_value(self, Class):
+        c = Class(retry_on_error=[])
+        assert c.retry_on_error == []
+        assert isinstance(c.retry, Retry)
+        assert c.retry._retries == 0
+
+    @pytest.mark.parametrize("Class", [Connection, UnixDomainSocketConnection])
+    def test_retry_on_error_and_timeout(self, Class):
+        c = Class(
+            retry_on_error=[ReadOnlyError, BusyLoadingError], retry_on_timeout=True
+        )
+        assert c.retry_on_error == [ReadOnlyError, BusyLoadingError, TimeoutError]
+        assert isinstance(c.retry, Retry)
+        assert c.retry._retries == 1
+
+    @pytest.mark.parametrize("retries", range(10))
+    @pytest.mark.parametrize("Class", [Connection, UnixDomainSocketConnection])
+    def test_retry_on_error_retry(self, Class, retries):
+        c = Class(retry_on_error=[ReadOnlyError], retry=Retry(NoBackoff(), retries))
+        assert c.retry_on_error == [ReadOnlyError]
+        assert isinstance(c.retry, Retry)
+        assert c.retry._retries == retries
+
 
 class TestRetry:
     "Test that Retry calls backoff and retries the expected number of times"
@@ -65,3 +106,85 @@ class TestRetry:
         assert self.actual_failures == 1 + retries
         assert backoff.reset_calls == 1
         assert backoff.calls == retries
+
+
+@pytest.mark.onlynoncluster
+class TestRedisClientRetry:
+    "Test the standalone Redis client behavior with retries"
+
+    def test_client_retry_on_error_with_success(self, request):
+        with patch.object(Redis, "parse_response") as parse_response:
+
+            def mock_parse_response(connection, *args, **options):
+                def ok_response(connection, *args, **options):
+                    return "MOCK_OK"
+
+                parse_response.side_effect = ok_response
+                raise ReadOnlyError()
+
+            parse_response.side_effect = mock_parse_response
+            r = _get_client(Redis, request, retry_on_error=[ReadOnlyError])
+            assert r.get("foo") == "MOCK_OK"
+            assert parse_response.call_count == 2
+
+    def test_client_retry_on_error_raise(self, request):
+        with patch.object(Redis, "parse_response") as parse_response:
+            parse_response.side_effect = BusyLoadingError()
+            retries = 3
+            r = _get_client(
+                Redis,
+                request,
+                retry_on_error=[ReadOnlyError, BusyLoadingError],
+                retry=Retry(NoBackoff(), retries),
+            )
+            with pytest.raises(BusyLoadingError):
+                try:
+                    r.get("foo")
+                finally:
+                    assert parse_response.call_count == retries + 1
+
+    def test_client_retry_on_error_different_error_raised(self, request):
+        with patch.object(Redis, "parse_response") as parse_response:
+            parse_response.side_effect = TimeoutError()
+            retries = 3
+            r = _get_client(
+                Redis,
+                request,
+                retry_on_error=[ReadOnlyError],
+                retry=Retry(NoBackoff(), retries),
+            )
+            with pytest.raises(TimeoutError):
+                try:
+                    r.get("foo")
+                finally:
+                    assert parse_response.call_count == 1
+
+    def test_client_retry_on_error_and_timeout(self, request):
+        with patch.object(Redis, "parse_response") as parse_response:
+            parse_response.side_effect = TimeoutError()
+            retries = 3
+            r = _get_client(
+                Redis,
+                request,
+                retry_on_error=[ReadOnlyError],
+                retry_on_timeout=True,
+                retry=Retry(NoBackoff(), retries),
+            )
+            with pytest.raises(TimeoutError):
+                try:
+                    r.get("foo")
+                finally:
+                    assert parse_response.call_count == retries + 1
+
+    def test_client_retry_on_timeout(self, request):
+        with patch.object(Redis, "parse_response") as parse_response:
+            parse_response.side_effect = TimeoutError()
+            retries = 3
+            r = _get_client(
+                Redis, request, retry_on_timeout=True, retry=Retry(NoBackoff(), retries)
+            )
+            with pytest.raises(TimeoutError):
+                try:
+                    r.get("foo")
+                finally:
+                    assert parse_response.call_count == retries + 1