Add s3 tests

Author: vshepard

s3/__init__.py

@@ -0,0 +1,8 @@
+import configparser
+
+
+def read_config(s3_config_file):
+    config = configparser.ConfigParser()
+    config.read_string('[fake-section]\n' + open(s3_config_file).read())
+
+    return config['fake-section']

s3/test_utils/__init__.py

@@ -0,0 +1,208 @@
+import os
+import io
+import sys
+
+import minio
+from minio import Minio
+from minio.deleteobjects import DeleteObject
+import urllib3
+from pg_probackup2.storage.fs_backup import TestBackupDir
+from pg_probackup2.init_helpers import init_params
+from s3.test_utils import config_provider
+
+root = os.path.realpath(os.path.join(os.path.dirname(__file__), '../..'))
+if root not in sys.path:
+    sys.path.append(root)
+
+status_forcelist = [413,  # RequestBodyTooLarge
+                    429,  # TooManyRequests
+                    500,  # InternalError
+                    503,  # ServerBusy
+                    ]
+
+DEFAULT_CONF_FILE = 's3/tests/s3.conf'
+
+
+class S3TestBackupDir(TestBackupDir):
+    is_file_based = False
+
+    def __init__(self, *, rel_path, backup):
+        self.access_key = None
+        self.secret_key = None
+        self.s3_type = None
+        self.tmp_path = None
+        self.host = None
+        self.port = None
+        self.bucket_name = None
+        self.region = None
+        self.bucket = None
+        self.path_suffix = None
+        self.https = None
+        self.s3_config_file = None
+        self.ca_certificate = None
+
+        self.set_s3_config_file()
+        self.setup_s3_env()
+
+        path = "pg_probackup"
+        if self.path_suffix:
+            path += "_" + self.path_suffix
+        if self.tmp_path == '' or os.path.isabs(self.tmp_path):
+            self.path = f"{path}{self.tmp_path}/{rel_path}/{backup}"
+        else:
+            self.path = f"{path}/{self.tmp_path}/{rel_path}/{backup}"
+
+        secure: bool = False
+        self.versioning: bool = False
+        if self.https in ['ON', 'HTTPS']:
+            secure = True
+        if self.https and self.ca_certificate:
+            http_client = urllib3.PoolManager(cert_reqs='CERT_REQUIRED',
+                                              ca_certs=self.ca_certificate,
+                                              retries=urllib3.Retry(total=5,
+                                                                    backoff_factor=1,
+                                                                    status_forcelist=status_forcelist))
+        else:
+            http_client = urllib3.PoolManager(retries=urllib3.Retry(total=5,
+                                                                    backoff_factor=1,
+                                                                    status_forcelist=status_forcelist))
+
+        self.conn = Minio(self.host + ":" + self.port, secure=secure, access_key=self.access_key,
+                          secret_key=self.secret_key, http_client=http_client)
+        if not self.conn.bucket_exists(self.bucket):
+            raise Exception(f"Test bucket {self.bucket} does not exist.")
+
+        try:
+            config = self.conn.get_bucket_versioning(self.bucket)
+            if config.status.lower() == "enabled" or config.status.lower() == "suspended":
+                self.versioning = True
+            else:
+                self.versioning = False
+        except Exception as e:
+            if "NotImplemented" in repr(e):
+                self.versioning = False
+            else:
+                raise e
+        self.pb_args = ('-B', '/' + self.path, f'--s3={init_params.s3_type}')
+        if self.s3_config_file:
+            self.pb_args += (f'--s3-config-file={self.s3_config_file}',)
+        return
+
+    def setup_s3_env(self, s3_config=None):
+        self.tmp_path = os.environ.get('PGPROBACKUP_TMP_DIR', default='')
+        self.host = os.environ.get('PG_PROBACKUP_S3_HOST', default='')
+
+        # If environment variables are not setup, use from config
+        if self.s3_config_file or s3_config:
+            minio_config = config_provider.read_config(self.s3_config_file or s3_config)
+            self.access_key = minio_config['access-key']
+            self.secret_key = minio_config['secret-key']
+            self.host = minio_config['s3-host']
+            self.port = minio_config['s3-port']
+            self.bucket = minio_config['s3-bucket']
+            self.region = minio_config['s3-region']
+            self.https = minio_config['s3-secure']
+            init_params.s3_type = 'minio'
+        else:
+            self.access_key = os.environ.get('PG_PROBACKUP_S3_ACCESS_KEY')
+            self.secret_key = os.environ.get('PG_PROBACKUP_S3_SECRET_ACCESS_KEY')
+            self.host = os.environ.get('PG_PROBACKUP_S3_HOST')
+            self.port = os.environ.get('PG_PROBACKUP_S3_PORT')
+            self.bucket = os.environ.get('PG_PROBACKUP_S3_BUCKET_NAME')
+            self.region = os.environ.get('PG_PROBACKUP_S3_REGION')
+            self.https = os.environ.get('PG_PROBACKUP_S3_HTTPS')
+            self.ca_certificate = os.environ.get('PG_PROBACKUP_S3_CA_CERTIFICATE')
+            init_params.s3_type = os.environ.get('PG_PROBACKUP_S3_TEST')
+
+        # multi-url case
+        # remove all urls from string except the first one
+        if ';' in self.host:
+            self.host = self.host[:self.host.find(';')]
+            if ':' in self.host:  # also change port if it was overridden in multihost string
+                self.port = self.host[self.host.find(':') + 1:]
+                self.host = self.host[:self.host.find(':')]
+
+    def set_s3_config_file(self):
+        s3_config = os.environ.get('PG_PROBACKUP_S3_CONFIG_FILE')
+        if s3_config is not None and s3_config.strip().lower() == "true":
+            self.s3_config_file = DEFAULT_CONF_FILE
+        else:
+            self.s3_config_file = s3_config
+
+    def list_instance_backups(self, instance):
+        full_path = os.path.join(self.path, 'backups', instance)
+        candidates = self.conn.list_objects(self.bucket, prefix=full_path, recursive=True)
+        return [os.path.basename(os.path.dirname(x.object_name))
+                for x in candidates if x.object_name.endswith('backup.control')]
+
+    def list_files(self, sub_dir, recursive=False):
+        full_path = os.path.join(self.path, sub_dir)
+        #  Need '/' in the end to find inside the folder
+        full_path_dir = full_path if full_path[-1] == '/' else full_path + '/'
+        object_list = self.conn.list_objects(self.bucket, prefix=full_path_dir, recursive=recursive)
+        return [obj.object_name.replace(full_path_dir, '', 1)
+                for obj in object_list
+                if not obj.is_dir]
+
+    def list_dirs(self, sub_dir):
+        full_path = os.path.join(self.path, sub_dir)
+        #  Need '/' in the end to find inside the folder
+        full_path_dir = full_path if full_path[-1] == '/' else full_path + '/'
+        object_list = self.conn.list_objects(self.bucket, prefix=full_path_dir, recursive=False)
+        return [obj.object_name.replace(full_path_dir, '', 1).rstrip('\\/')
+                for obj in object_list
+                if obj.is_dir]
+
+    def read_file(self, sub_path, *, text=True):
+        full_path = os.path.join(self.path, sub_path)
+        bytes = self.conn.get_object(self.bucket, full_path).read()
+        if not text:
+            return bytes
+        return bytes.decode('utf-8')
+
+    def write_file(self, sub_path, data, *, text=True):
+        full_path = os.path.join(self.path, sub_path)
+        if text:
+            data = data.encode('utf-8')
+        self.conn.put_object(self.bucket, full_path, io.BytesIO(data), length=len(data))
+
+    def cleanup(self, dir=''):
+        self.remove_dir(dir)
+
+    def remove_file(self, sub_path):
+        full_path = os.path.join(self.path, sub_path)
+        self.conn.remove_object(self.bucket, full_path)
+
+    def remove_dir(self, sub_path):
+        if sub_path:
+            full_path = os.path.join(self.path, sub_path)
+        else:
+            full_path = self.path
+        objs = self.conn.list_objects(self.bucket, prefix=full_path, recursive=True,
+                                      include_version=self.versioning)
+        delobjs = (DeleteObject(o.object_name, o.version_id) for o in objs)
+        errs = list(self.conn.remove_objects(self.bucket, delobjs))
+        if errs:
+            strerrs = "; ".join(str(err) for err in errs)
+            raise Exception("There were errors: {0}".format(strerrs))
+
+    def exists(self, sub_path):
+        full_path = os.path.join(self.path, sub_path)
+        try:
+            self.conn.stat_object(self.bucket, full_path)
+            return True
+        except minio.error.S3Error as s3err:
+            if s3err.code == 'NoSuchKey':
+                return False
+            raise s3err
+        except Exception as err:
+            raise err
+
+    def __str__(self):
+        return '/' + self.path
+
+    def __repr__(self):
+        return "S3TestBackupDir" + str(self.path)
+
+    def __fspath__(self):
+        return self.path

s3/test_utils/config_provider.py

@@ -0,0 +1,17 @@
+import unittest
+import os
+
+from . import auth_test, param_test
+
+
+def load_tests(loader, tests, pattern):
+    suite = unittest.TestSuite()
+
+    if 'PG_PROBACKUP_TEST_BASIC' in os.environ:
+        if os.environ['PG_PROBACKUP_TEST_BASIC'] == 'ON':
+            loader.testMethodPrefix = 'test_basic'
+
+    suite.addTests(loader.loadTestsFromModule(auth_test))
+    suite.addTests(loader.loadTestsFromModule(param_test))
+
+    return suite

s3/test_utils/s3_backup.py

@@ -0,0 +1,36 @@
+import os
+import sys
+
+root = os.path.realpath(os.path.join(os.path.dirname(__file__), '../..'))
+if root not in sys.path:
+    sys.path.append(root)
+
+from tests.helpers.ptrack_helpers import ProbackupTest
+
+
+class AuthorizationTest(ProbackupTest):
+    """
+    Check connect to S3 via pre_start_checks() function
+    calling pg_probackup init --s3
+
+    test that s3 keys allow to connect to all types of storages
+    """
+
+    def s3_auth_test(self):
+        console_output = self.pb.init(options=["--log-level-console=VERBOSE"])
+
+        self.assertNotIn(': 403', console_output)  # Because we can have just '403' substring in timestamp
+        self.assertMessage(console_output, contains='S3_pre_start_check successful')
+        self.assertMessage(console_output, contains='HTTP response: 200')
+        self.assertIn(
+            f"INFO: Backup catalog '{self.backup_dir}' successfully initialized",
+            console_output)
+
+    def test_log_level_file_requires_log_directory(self):
+        console_output = self.pb.init(options=["--log-level-file=VERBOSE"],
+                                      skip_log_directory=True,
+                                      expect_error=True)
+
+        self.assertMessage(console_output,
+                           contains='ERROR: Cannot save S3 logs to a file. You must specify --log-directory option when'
+                                    ' running backup with --log-level-file option enabled')

s3/tests/__init__.py

@@ -0,0 +1,56 @@
+import os
+import sys
+import unittest
+import subprocess
+
+root = os.path.realpath(os.path.join(os.path.dirname(__file__), '../..'))
+if root not in sys.path:
+    sys.path.append(root)
+
+from tests.helpers.ptrack_helpers import ProbackupTest, fs_backup_class
+
+MULTIHOST_CONF_FILE = 's3/tests/multihost.conf'
+
+class CustomTest(ProbackupTest):
+    """
+    Class for custom tests for checking some S3 features.
+    """
+
+    @unittest.skip("This test is meant for manual use only. Comment this line for testing")
+    @unittest.skipIf(fs_backup_class.is_file_based, "This test can only be launched under S3")
+    def test_s3_multihost_pbckp_825(self):
+        """
+        Test for checking multihost case.
+        !!! WARNING !!! For manual testing only.
+        For checking multihost working you should comment first 'unittest.skip'
+        and fill proper IP addresses in file multihost.conf.
+        Also, it is recommended to set options in enviroment variables -- just in case.
+        """
+        node = self.pg_node.make_simple('node', set_replication=True)
+
+        self.pb.init()
+        self.pb.add_instance('node', node)
+        self.pb.set_archiving('node', node)
+        node.slow_start()
+
+        node.pgbench_init(scale=2)
+        pgbench = node.pgbench(
+            stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
+        pgbench.wait()
+        pgbench.stdout.close()
+        before = node.table_checksum("pgbench_branches")
+        backup_id = self.pb.backup_node('node', node, options=["-j", "4", "--s3-config-file", MULTIHOST_CONF_FILE])
+        before_pgdata = self.pgdata_content(node.data_dir)
+
+        node.stop()
+        node.cleanup()
+
+        restore_result = self.pb.restore_node('node', node, options=["-j", "4"])
+        self.assertMessage(restore_result, contains="INFO: Restore of backup {0} completed.".format(backup_id))
+        after_pgdata = self.pgdata_content(node.data_dir)
+
+        node.slow_start()
+
+        after = node.table_checksum("pgbench_branches")
+        self.assertEqual(before, after)
+        self.compare_pgdata(before_pgdata, after_pgdata)

s3/tests/auth_test.py

@@ -0,0 +1,7 @@
+access-key=minioadmin
+secret-key=minioadmin
+s3-host=127.0.0.1:9000;192.168.1.38
+s3-port=9000
+s3-bucket=s3test
+s3-region=us-east-1
+s3-secure=OFF