CLIENT REPLY support, available since redis 3.2.0 (#1581)

Author: chayim

redis/commands.py

@@ -376,6 +376,25 @@ def client_getname(self):
         "Returns the current connection name"
         return self.execute_command('CLIENT GETNAME')
 
+    def client_reply(self, reply):
+        """Enable and disable redis server replies.
+        ``reply`` Must be ON OFF or SKIP,
+            ON - The default most with server replies to commands
+            OFF - Disable server responses to commands
+            SKIP - Skip the response of the immediately following command.
+
+        Note: When setting OFF or SKIP replies, you will need a client object
+        with a timeout specified in seconds, and will need to catch the
+        TimeoutError.
+              The test_client_reply unit test illustrates this, and
+              conftest.py has a client with a timeout.
+        See https://redis.io/commands/client-reply
+        """
+        replies = ['ON', 'OFF', 'SKIP']
+        if reply not in replies:
+            raise DataError('CLIENT REPLY must be one of %r' % replies)
+        return self.execute_command("CLIENT REPLY", reply)
+
     def client_id(self):
         "Returns the current connection id"
         return self.execute_command('CLIENT ID')

tests/conftest.py

@@ -100,6 +100,12 @@ def r(request):
         yield client
 
 
+@pytest.fixture()
+def r_timeout(request):
+    with _get_client(redis.Redis, request, socket_timeout=1) as client:
+        yield client
+
+
 @pytest.fixture()
 def r2(request):
     "A second client for tests that need multiple"

tests/test_commands.py

@@ -478,6 +478,19 @@ def test_client_pause(self, r):
     def test_client_unpause(self, r):
         assert r.client_unpause() == b'OK'
 
+    @skip_if_server_version_lt('3.2.0')
+    def test_client_reply(self, r, r_timeout):
+        assert r_timeout.client_reply('ON') == b'OK'
+        with pytest.raises(exceptions.TimeoutError):
+            r_timeout.client_reply('OFF')
+
+            r_timeout.client_reply('SKIP')
+
+        assert r_timeout.set('foo', 'bar')
+
+        # validate it was set
+        assert r.get('foo') == b'bar'
+
     def test_config_get(self, r):
         data = r.config_get()
         assert 'maxmemory' in data