Replication slots

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.
 
@@ -171,7 +171,7 @@ def spawn_replica(self, name=None, destroy=True):
 
         # Assign it a master and a recovery file (private magic)
         node._assign_master(self.original_node)
-        node._create_recovery_conf(username=self.username)
+        node._create_recovery_conf(username=self.username, slot_name=slot_name)
 
         return node
 

testgres/consts.py

@@ -24,3 +24,6 @@
 PG_LOG_FILE = "postgresql.log"
 UTILS_LOG_FILE = "utils.log"
 BACKUP_LOG_FILE = "backup.log"
+
+# default replication slots number
+REPLICATION_SLOTS = 10

testgres/node.py

@@ -32,7 +32,8 @@
     RECOVERY_CONF_FILE, \
     PG_LOG_FILE, \
     UTILS_LOG_FILE, \
-    PG_PID_FILE
+    PG_PID_FILE, \
+    REPLICATION_SLOTS
 
 from .decorators import \
     method_decorator, \
@@ -179,7 +180,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 +208,9 @@ def _create_recovery_conf(self, username):
             "standby_mode=on\n"
         ).format(conninfo)
 
+        if slot_name:
+            line += "primary_slot_name={}\n".format(slot_name)
+
         self.append_conf(RECOVERY_CONF_FILE, line)
 
     def _maybe_start_logger(self):
@@ -340,11 +344,14 @@ def get_auth_method(t):
                 conf.write(u"fsync = off\n")
 
             # yapf: disable
-            conf.write(u"log_statement = {}\n"
-                       u"listen_addresses = '{}'\n"
-                       u"port = {}\n".format(log_statement,
-                                             self.host,
-                                             self.port))
+            conf.write(
+                u"log_statement = {}\n"
+                u"listen_addresses = '{}'\n"
+                u"port = {}\n"
+                u"max_replication_slots = {}\n".format(log_statement,
+                                                       self.host,
+                                                       self.port,
+                                                       REPLICATION_SLOTS))
 
             # replication-related settings
             if allow_streaming:
@@ -856,7 +863,24 @@ 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 +894,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()