Support SYNC and PSYNC (#1741)

Co-authored-by: Chayim <chayim@users.noreply.github.com>

3 files changed, 38 insertions, 1 deletions

diff --git a/redis/commands/core.py b/redis/commands/core.py
index 62e3ba8..835ea61 100644
--- a/redis/commands/core.py
+++ b/redis/commands/core.py
@@ -637,6 +637,31 @@ class ManagementCommands:
             args.append(b"ASYNC")
         return self.execute_command("FLUSHDB", *args, **kwargs)
 
+    def sync(self):
+        """
+        Initiates a replication stream from the master.
+
+        For more information check https://redis.io/commands/sync
+        """
+        from redis.client import NEVER_DECODE
+
+        options = {}
+        options[NEVER_DECODE] = []
+        return self.execute_command("SYNC", **options)
+
+    def psync(self, replicationid, offset):
+        """
+        Initiates a replication stream from the master.
+        Newer version for `sync`.
+
+        For more information check https://redis.io/commands/sync
+        """
+        from redis.client import NEVER_DECODE
+
+        options = {}
+        options[NEVER_DECODE] = []
+        return self.execute_command("PSYNC", replicationid, offset, **options)
+
     def swapdb(self, first, second, **kwargs):
         """
         Swap two databases
diff --git a/redis/connection.py b/redis/connection.py
index 4ba58fe..6c4494b 100755
--- a/redis/connection.py
+++ b/redis/connection.py
@@ -382,7 +382,7 @@ class HiredisParser(BaseParser):
         except Exception:
             pass
 
-    def on_connect(self, connection):
+    def on_connect(self, connection, **kwargs):
         self._sock = connection._sock
         self._socket_timeout = connection.socket_timeout
         kwargs = {
diff --git a/tests/test_commands.py b/tests/test_commands.py
index eab9072..b8dc69f 100644
--- a/tests/test_commands.py
+++ b/tests/test_commands.py
@@ -4151,6 +4151,18 @@ class TestRedisCommands:
             assert r.replicaof("NO ONE")
         assert r.replicaof("NO", "ONE")
 
+    @skip_if_server_version_lt("2.8.0")
+    def test_sync(self, r):
+        r2 = redis.Redis(port=6380, decode_responses=False)
+        res = r2.sync()
+        assert b"REDIS" in res
+
+    @skip_if_server_version_lt("2.8.0")
+    def test_psync(self, r):
+        r2 = redis.Redis(port=6380, decode_responses=False)
+        res = r2.psync(r2.client_id(), 1)
+        assert b"FULLRESYNC" in res
+
 
 @pytest.mark.onlynoncluster
 class TestBinarySave: