Add readme and basic test for pg_probackup2

Author: vshepard

testgres/plugins/pg_probackup2/README.md

@@ -0,0 +1,65 @@
+# testgres - pg_probackup2
+
+Ccontrol and testing utility for [pg_probackup2](https://github.com/postgrespro/pg_probackup). Python 3.5+ is supported.
+
+
+## Installation
+
+To install `testgres`, run:
+
+```
+pip install testgres-pg_probackup
+```
+
+We encourage you to use `virtualenv` for your testing environment.
+The package requires testgres~=1.9.3.
+
+## Usage
+
+### Environment
+
+> Note: by default testgres runs `initdb`, `pg_ctl`, `psql` provided by `PATH`.
+
+There are several ways to specify a custom postgres installation:
+
+* export `PG_CONFIG` environment variable pointing to the `pg_config` executable;
+* export `PG_BIN` environment variable pointing to the directory with executable files.
+
+Example:
+
+```bash
+export PG_BIN=$HOME/pg/bin
+python my_tests.py
+```
+
+
+### Examples
+
+Here is an example of what you can do with `testgres-pg_probackup2`:
+
+```python
+# You can see full script here plugins/pg_probackup2/pg_probackup2/tests/basic_test.py
+def test_full_backup(self):
+    # Setting up a simple test node
+    node = self.pg_node.make_simple('node', pg_options={"fsync": "off", "synchronous_commit": "off"})
+
+    # Initialize and configure Probackup
+    self.pb.init()
+    self.pb.add_instance('node', node)
+    self.pb.set_archiving('node', node)
+
+    # Start the node and initialize pgbench
+    node.slow_start()
+    node.pgbench_init(scale=100, no_vacuum=True)
+
+    # Perform backup and validation
+    backup_id = self.pb.backup_node('node', node)
+    out = self.pb.validate('node', backup_id)
+
+    # Check if the backup is valid
+    self.assertIn(f"INFO: Backup {backup_id} is valid", out)
+```
+
+## Authors
+
+[Postgres Professional](https://postgrespro.ru/about)

testgres/plugins/pg_probackup2/pg_probackup2/app.py

@@ -1,10 +1,17 @@
+import contextlib
+import importlib
 import json
 import os
 import re
 import subprocess
+import sys
+import threading
+import time
 import unittest
 
-from .storage.fs_backup import TestBackupDir
+import testgres
+
+from .storage.fs_backup import TestBackupDir, FSTestBackupDir
 from .gdb import GDBobj
 from .init_helpers import init_params
 
@@ -271,14 +278,14 @@ def backup_replica_node(self, instance, node, data_dir=False, *,
         # we need to break on wait_wal_lsn in pg_stop_backup
         gdb.run_until_break()
         if backup_type == 'page':
-            self.test_class.switch_wal_segment(master)
+            self.switch_wal_segment(master)
         if '--stream' not in options:
             gdb.continue_execution_until_break()
-        self.test_class.switch_wal_segment(master)
+        self.switch_wal_segment(master)
         gdb.continue_execution_until_exit()
 
-        output = self.test_class.read_pb_log()
-        self.test_class.unlink_pg_log()
+        output = self.read_pb_log()
+        self.unlink_pg_log()
         parsed_output = re.compile(r'Backup \S+ completed').search(output)
         assert parsed_output, f"Expected: `Backup 'backup_id' completed`, but found `{output}`"
         backup_id = parsed_output[0].split(' ')[1]
@@ -693,3 +700,66 @@ def set_archiving(
         options['archive_command'] = archive_command
 
         node.set_auto_conf(options)
+
+    def switch_wal_segment(self, node, sleep_seconds=1, and_tx=False):
+        """
+        Execute pg_switch_wal() in given node
+
+        Args:
+            node: an instance of PostgresNode or NodeConnection class
+        """
+        if isinstance(node, testgres.PostgresNode):
+            with node.connect('postgres') as con:
+                if and_tx:
+                    con.execute('select txid_current()')
+                lsn = con.execute('select pg_switch_wal()')[0][0]
+        else:
+            lsn = node.execute('select pg_switch_wal()')[0][0]
+
+        if sleep_seconds > 0:
+            time.sleep(sleep_seconds)
+        return lsn
+
+    @contextlib.contextmanager
+    def switch_wal_after(self, node, seconds, and_tx=True):
+        tm = threading.Timer(seconds, self.switch_wal_segment, [node, 0, and_tx])
+        tm.start()
+        try:
+            yield
+        finally:
+            tm.cancel()
+            tm.join()
+
+    def read_pb_log(self):
+        with open(os.path.join(self.pb_log_path, 'pg_probackup.log')) as fl:
+            return fl.read()
+
+    def unlink_pg_log(self):
+        os.unlink(os.path.join(self.pb_log_path, 'pg_probackup.log'))
+
+    def load_backup_class(fs_type):
+        fs_type = os.environ.get('PROBACKUP_FS_TYPE')
+        implementation = f"{__package__}.fs_backup.FSTestBackupDir"
+        if fs_type:
+            implementation = fs_type
+
+        print("Using ", implementation)
+        module_name, class_name = implementation.rsplit(sep='.', maxsplit=1)
+
+        module = importlib.import_module(module_name)
+
+        return getattr(module, class_name)
+
+
+# Local or S3 backup
+fs_backup_class = FSTestBackupDir
+if os.environ.get('PG_PROBACKUP_S3_TEST', os.environ.get('PROBACKUP_S3_TYPE_FULL_TEST')):
+    root = os.path.realpath(os.path.join(os.path.dirname(__file__), '../..'))
+    if root not in sys.path:
+        sys.path.append(root)
+    from pg_probackup2.storage.s3_backup import S3TestBackupDir
+    fs_backup_class = S3TestBackupDir
+
+    def build_backup_dir(self, backup='backup'):
+        return fs_backup_class(rel_path=self.rel_path, backup=backup)
+

testgres/plugins/pg_probackup2/pg_probackup2/tests/basic_test.py

@@ -0,0 +1,79 @@
+import os
+import shutil
+import unittest
+import testgres
+from pg_probackup2.app import ProbackupApp
+from pg_probackup2.init_helpers import Init, init_params
+from pg_probackup2.app import build_backup_dir
+
+
+class TestUtils:
+    @staticmethod
+    def get_module_and_function_name(test_id):
+        try:
+            module_name = test_id.split('.')[-2]
+            fname = test_id.split('.')[-1]
+        except IndexError:
+            print(f"Couldn't get module name and function name from test_id: `{test_id}`")
+            module_name, fname = test_id.split('(')[1].split('.')[1], test_id.split('(')[0]
+        return module_name, fname
+
+
+class ProbackupTest(unittest.TestCase):
+    def setUp(self):
+        self.setup_test_environment()
+        self.setup_test_paths()
+        self.setup_backup_dir()
+        self.setup_probackup()
+
+    def setup_test_environment(self):
+        self.output = None
+        self.cmd = None
+        self.nodes_to_cleanup = []
+        self.module_name, self.fname = TestUtils.get_module_and_function_name(self.id())
+        self.test_env = Init().test_env()
+
+    def setup_test_paths(self):
+        self.rel_path = os.path.join(self.module_name, self.fname)
+        self.test_path = os.path.join(init_params.tmp_path, self.rel_path)
+        os.makedirs(self.test_path)
+        self.pb_log_path = os.path.join(self.test_path, "pb_log")
+
+    def setup_backup_dir(self):
+        self.backup_dir = build_backup_dir(self, 'backup')
+        self.backup_dir.cleanup()
+
+    def setup_probackup(self):
+        self.pg_node = testgres.NodeApp(self.test_path, self.nodes_to_cleanup)
+        self.pb = ProbackupApp(self, self.pg_node, self.pb_log_path, self.test_env,
+                               auto_compress_alg='zlib', backup_dir=self.backup_dir)
+
+    def tearDown(self):
+        if os.path.exists(self.test_path):
+            shutil.rmtree(self.test_path)
+
+
+class BasicTest(ProbackupTest):
+    def test_full_backup(self):
+        # Setting up a simple test node
+        node = self.pg_node.make_simple('node', pg_options={"fsync": "off", "synchronous_commit": "off"})
+
+        # Initialize and configure Probackup
+        self.pb.init()
+        self.pb.add_instance('node', node)
+        self.pb.set_archiving('node', node)
+
+        # Start the node and initialize pgbench
+        node.slow_start()
+        node.pgbench_init(scale=100, no_vacuum=True)
+
+        # Perform backup and validation
+        backup_id = self.pb.backup_node('node', node)
+        out = self.pb.validate('node', backup_id)
+
+        # Check if the backup is valid
+        self.assertIn(f"INFO: Backup {backup_id} is valid", out)
+
+
+if __name__ == "__main__":
+    unittest.main()