Replication slots

Author: zilder

testgres/backup.py

@@ -154,7 +154,7 @@ def spawn_primary(self, name=None, destroy=True):
 
         return node
 
-    def spawn_replica(self, name=None, destroy=True):
+    def spawn_replica(self, name=None, destroy=True, slot_name=None):
         """
         Create a replica of the original node from a backup.
 

testgres/node.py

@@ -179,7 +179,7 @@ def _assign_master(self, master):
         # now this node has a master
         self._master = master
 
-    def _create_recovery_conf(self, username):
+    def _create_recovery_conf(self, username, slot_name=None):
         """NOTE: this is a private method!"""
 
         # fetch master of this node
@@ -207,6 +207,9 @@ def _create_recovery_conf(self, username):
             "standby_mode=on\n"
         ).format(conninfo)
 
+        if slot_name:
+            line += "primary_slot_name={}".format()
+
         self.append_conf(RECOVERY_CONF_FILE, line)
 
     def _maybe_start_logger(self):
@@ -856,7 +859,22 @@ def backup(self, **kwargs):
 
         return NodeBackup(node=self, **kwargs)
 
-    def replicate(self, name=None, **kwargs):
+    def create_replication_slot(self, slot_name, dbname=None, username=None):
+        """
+        Create a physical replication slot.
+
+        Args:
+            slot_name: slot name
+            dbname: database name
+            username: database user name
+        """
+        query = "select pg_create_physical_replication_slot('{}')".format(slot_name)
+
+        self.execute(query=query,
+                     dbname=dbname or default_dbname(),
+                     username=username or default_username())
+
+    def replicate(self, name=None, slot_name=None, **kwargs):
         """
         Create a binary replica of this node.
 
@@ -870,7 +888,9 @@ def replicate(self, name=None, **kwargs):
         backup = self.backup(**kwargs)
 
         # transform backup into a replica
-        return backup.spawn_replica(name=name, destroy=True)
+        return backup.spawn_replica(name=name,
+                                    destroy=True,
+                                    slot_name=slot_name)
 
     def catchup(self, dbname=None, username=None):
         """

tests/test_simple.py

@@ -382,6 +382,18 @@ def test_replicate(self):
                 res = node.execute('select * from test')
                 self.assertListEqual(res, [])
 
+    def test_replication_slots(self):
+        query_create = 'create table test as select generate_series(1, 2) as val'
+
+        with get_new_node() as node:
+            node.init(allow_streaming=True).start()
+            node.create_replication_slot('slot1')
+            node.execute(query_create)
+
+            with node.replicate(slot_name='slot1').start() as replica:
+                res = replica.execute('select * from test')
+                self.assertListEqual(res, [(1, ), (2, )])
+
     def test_incorrect_catchup(self):
         with get_new_node() as node:
             node.init(allow_streaming=True).start()