diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml new file mode 100644 index 0000000..6b10df9 --- /dev/null +++ b/.github/workflows/ci.yml @@ -0,0 +1,68 @@ +# This workflow will install Python dependencies, run tests and lint with a variety of Python versions +# For more information see: https://docs.github.com/en/actions/automating-builds-and-tests/building-and-testing-python + +name: CI + +on: + push: + branches: [ "main" ] + paths-ignore: + - "**/*.md" + + pull_request: + branches: [ "main" ] + paths-ignore: + - "**/*.md" + +jobs: + lint: + runs-on: ubuntu-latest + strategy: + fail-fast: false + matrix: + python-version: ["3.8", "3.9", "3.10", "3.11", "3.12", "3.13", "3.14"] + + steps: + - name: Checkout + uses: actions/checkout@v7 + - name: Set up Python ${{ matrix.python-version }} + uses: actions/setup-python@v7 + with: + python-version: ${{ matrix.python-version }} + cache: 'pip' + - name: Install dependencies + run: | + python -m pip install --upgrade pip + python -m pip install flake8 flake8-pyproject ruff==0.15.22 + - name: Lint with flake8 + run: | + flake8 . + - name: Lint with ruff + run: | + ruff check . + + build-check: + runs-on: ubuntu-latest + needs: lint + steps: + - name: Checkout + uses: actions/checkout@v7 + + - name: Set up Python 3.12 + uses: actions/setup-python@v7 + with: + python-version: "3.12" + cache: 'pip' + + - name: Install build tools + run: | + python -m pip install --upgrade pip + python -m pip install build twine + + - name: Build package + run: | + python -m build + + - name: Check package metadata + run: | + twine check dist/* diff --git a/__init__.py b/__init__.py index 8b13789..e69de29 100644 --- a/__init__.py +++ b/__init__.py @@ -1 +0,0 @@ - diff --git a/pg_probackup2/app.py b/pg_probackup2/app.py deleted file mode 100644 index 7026463..0000000 --- a/pg_probackup2/app.py +++ /dev/null @@ -1,955 +0,0 @@ -import contextlib -import importlib -import json -import logging -import os -import re -import subprocess -import threading -import time -import unittest - -import testgres - -from .storage.fs_backup import TestBackupDir, FSTestBackupDir -from .gdb import GDBobj -from .init_helpers import init_params - -warning = """ -Wrong splint in show_pb -Original Header:f -{header} -Original Body: -{body} -Splitted Header -{header_split} -Splitted Body -{body_split} -""" - - -class ProbackupException(Exception): - def __init__(self, message, cmd): - self.message = message - self.cmd = cmd - - def __str__(self): - return '\n ERROR: {0}\n CMD: {1}'.format(repr(self.message), self.cmd) - - -# Local backup control -fs_backup_class = FSTestBackupDir - - -class _OutputContainer: - """Internal container for output and cmd when test_class is not provided""" - def __init__(self): - self.output = None - self.cmd = None - - -class ProbackupApp: - - def __init__(self, test_class: unittest.TestCase = None, - pg_node=None, pb_log_path=None, test_env=None, auto_compress_alg=None, - backup_dir=None, probackup_path=None): - self.process = None - # Use provided test_class or create internal container - self.test_class = test_class if test_class is not None else _OutputContainer() - self.pg_node = pg_node - self.pb_log_path = pb_log_path - self.test_env = test_env if test_env is not None else os.environ.copy() - self.auto_compress_alg = auto_compress_alg - self.backup_dir = backup_dir - self.probackup_path = probackup_path or init_params.probackup_path - self.probackup_old_path = init_params.probackup_old_path - self.remote = init_params.remote - self.wal_tree_enabled = init_params.wal_tree_enabled - self.verbose = init_params.verbose - self.archive_compress = init_params.archive_compress - self.test_class.output = None - self.execution_time = None - self.valgrind = init_params.valgrind - self.valgrind_sup_path = init_params.valgrind_sup_path - - def form_daemon_process(self, cmdline, env): - if self.test_class.output is None: - self.test_class.output = str() - - def stream_output(stream) -> None: - try: - for line in iter(stream.readline, ''): - print(line) - self.test_class.output += line - finally: - stream.close() - - self.process = subprocess.Popen( - cmdline, - stdout=subprocess.PIPE, - stderr=subprocess.PIPE, - text=True, - env=env - ) - logging.info(f"Process started in background with PID: {self.process.pid}") - - if self.process.stdout and self.process.stderr: - self.stdout_thread = threading.Thread(target=stream_output, args=(self.process.stdout,), daemon=True) - self.stderr_thread = threading.Thread(target=stream_output, args=(self.process.stderr,), daemon=True) - - self.stdout_thread.start() - self.stderr_thread.start() - - return self.process.pid - - # ---- Start run function ---- # - def run(self, command, gdb=False, old_binary=False, return_id=True, env=None, - skip_log_directory=False, expect_error=False, use_backup_dir=True, daemonize=False): - """ - Run pg_probackup - backup_dir: target directory for making backup - command: commandline options - expect_error: option for ignoring errors and getting error message as a result of running the function - gdb: when True it returns GDBObj(), when tuple('suspend', port) it runs probackup - in suspended gdb mode with attachable gdb port, for local debugging - """ - command = self._add_backup_dir_to_cmd(command, use_backup_dir) - # Old bin or regular one - binary_path = self._get_binary_path(old_binary) - - if not env: - env = self.test_env - # Add additional options if needed - command, strcommand = self._add_options(command, skip_log_directory) - - self.test_class.cmd = f"{binary_path} {strcommand}" - if self.verbose: - print(self.test_class.cmd) - - cmdline = self._form_cmdline(binary_path, command, use_valgrind=gdb is False) - - if gdb is True: - # general test flow for using GDBObj - return GDBobj(cmdline, self.test_class) - - return self._execute_command(cmdline, env, command, gdb, expect_error, return_id, daemonize) - - def _add_backup_dir_to_cmd(self, command: list, use_backup_dir: TestBackupDir): - if isinstance(use_backup_dir, TestBackupDir): - return [command[0], *use_backup_dir.pb_args, *command[1:]] - elif use_backup_dir: - return [command[0], *self.backup_dir.pb_args, *command[1:]] - else: - return [command[0], *self.backup_dir.pb_args[2:], *command[1:]] - - def _get_binary_path(self, old_binary): - if old_binary: - if not self.probackup_old_path: - logging.error('PGPROBACKUPBIN_OLD is not set') - exit(1) - return self.probackup_old_path - return self.probackup_path - - def _add_options(self, command: list, skip_log_directory: bool): - strcommand = ' '.join(str(p) for p in command) - - if '--log-level-file' in strcommand and \ - '--log-directory' not in strcommand and \ - not skip_log_directory: - command += ['--log-directory=' + self.pb_log_path] - strcommand += ' ' + command[-1] - - if init_params.major_version == 2: - if 'pglz' in strcommand and \ - ' -j' not in strcommand and \ - '--thread' not in strcommand: - command += ['-j', '1'] - strcommand += ' -j 1' - - return command, strcommand - - def _form_cmdline(self, binary_path, command, use_valgrind=True): - cmdline = [binary_path, *command] - - if not use_valgrind or command[0] == "--version": - return cmdline - if self.valgrind != 'y' and not self.valgrind_sup_path: - return cmdline - - os.makedirs(self.pb_log_path, exist_ok=True) - valgrind_cmd = [ - "valgrind", - "--gen-suppressions=all", - "--leak-check=full", - "--show-reachable=yes", - "--error-limit=no", - "--show-leak-kinds=all", - "--errors-for-leak-kinds=all", - "--error-exitcode=0", - f"--log-file={os.path.join(self.pb_log_path, f'valgrind-{command[0]}-%p.log')}" - ] - if self.valgrind_sup_path: - if os.path.isfile(self.valgrind_sup_path): - valgrind_cmd += [f"--suppressions={self.valgrind_sup_path}"] - else: - raise FileNotFoundError(f"PG_PROBACKUP_VALGRIND_SUP must be the path to a valgrind suppression file, " - f"but found: {self.valgrind_sup_path}") - else: - # Assume the tests are started from the root of pg_probackup repository - valgrind_cmd += [f"--suppressions=.valgrind.supp"] - - return valgrind_cmd + cmdline - - def _execute_command(self, cmdline, env, command, gdb, expect_error, return_id, daemonize): - try: - if isinstance(gdb, tuple) and gdb[0] == 'suspend': - gdb_port = gdb[1] - cmdline = ['gdbserver'] + ['localhost:' + str(gdb_port)] + cmdline - logging.warning("pg_probackup gdb suspended, waiting gdb connection on localhost:{0}".format(gdb_port)) - - # Execute command - start_time = time.time() - if daemonize: - pid = self.form_daemon_process(cmdline, env) - if expect_error: - _timeout = 120 - if self.stdout_thread: - self.stdout_thread.join(_timeout) - if self.stderr_thread: - self.stderr_thread.join(_timeout) - return pid - else: - self.test_class.output = subprocess.check_output( - cmdline, - stderr=subprocess.STDOUT, - env=env - ).decode('utf-8', errors='replace') - end_time = time.time() - self.execution_time = end_time - start_time - - if command[0] == 'backup' and return_id: - result = self.get_backup_id() - else: - result = self.test_class.output - if expect_error is True: - assert False, f"Exception was expected, but run finished successful with result: `{result}`\n" \ - f"CMD: {self.test_class.cmd}" - elif expect_error: - assert False, f"Exception was expected {expect_error}, but run finished successful with result: `{result}`\n" \ - f"CMD: {self.test_class.cmd}" - return result - except subprocess.CalledProcessError as e: - self.test_class.output = e.output.decode('utf-8').replace("\r", "") - if expect_error: - return self.test_class.output - else: - raise ProbackupException(self.test_class.output, self.test_class.cmd) - # ---- End run function ---- # - - def get_backup_id(self): - if init_params.major_version > 2: - pattern = re.compile(r"Backup (.*) completed successfully.") - for line in self.test_class.output.splitlines(): - match = pattern.search(line) - if match: - return match.group(1) - else: - for line in self.test_class.output.splitlines(): - if 'INFO: Backup' and 'completed' in line: - return line.split()[2] - return None - - def init(self, options=None, old_binary=False, skip_log_directory=False, expect_error=False, use_backup_dir=True): - if options is None: - options = [] - return self.run(['init'] + options, - old_binary=old_binary, - skip_log_directory=skip_log_directory, - expect_error=expect_error, - use_backup_dir=use_backup_dir - ) - - def add_instance(self, instance, node, old_binary=False, options=None, expect_error=False, datname=False): - if options is None: - options = [] - - if not datname: - datname = 'postgres' - - cmd = [ - 'add-instance', - '--instance={0}'.format(instance), - '-D', node.data_dir, - '--pgport', '%i' % node.port, - '--pgdatabase', datname - ] - - # don`t forget to kill old_binary after remote ssh release - if self.remote and not old_binary: - options = options + [ - '--remote-proto=ssh', - '--remote-host=localhost'] - - if self.wal_tree_enabled: - options = options + ['--wal-tree'] - - return self.run(cmd + options, old_binary=old_binary, expect_error=expect_error) - - def set_config(self, instance, old_binary=False, options=None, expect_error=False): - if options is None: - options = [] - cmd = [ - 'set-config', - '--instance={0}'.format(instance), - ] - - return self.run(cmd + options, old_binary=old_binary, expect_error=expect_error) - - def set_backup(self, instance, backup_id=False, - old_binary=False, options=None, expect_error=False): - if options is None: - options = [] - cmd = [ - 'set-backup', - ] - - if instance: - cmd = cmd + ['--instance={0}'.format(instance)] - - if backup_id: - cmd = cmd + ['-i', backup_id] - - return self.run(cmd + options, old_binary=old_binary, expect_error=expect_error) - - def del_instance(self, instance, options=None, old_binary=False, expect_error=False): - if options is None: - options = [] - cmd = ['del-instance', '--instance={0}'.format(instance)] + options - return self.run(cmd, - old_binary=old_binary, - expect_error=expect_error) - - def backup_node( - self, instance, node, data_dir=False, - backup_type='full', datname=False, options=None, - gdb=False, - old_binary=False, return_id=True, no_remote=False, - env=None, - expect_error=False, - sync=False - ): - if options is None: - options = [] - if not node and not data_dir: - logging.error('You must provide ether node or data_dir for backup') - exit(1) - - if not datname: - datname = 'postgres' - - cmd_list = [ - 'backup', - '--instance={0}'.format(instance), - # "-D", pgdata, - '-p', '%i' % node.port, - '-d', datname - ] - - if data_dir: - cmd_list += ['-D', self._node_dir(data_dir)] - - # don`t forget to kill old_binary after remote ssh release - if self.remote and not old_binary and not no_remote: - options = options + [ - '--remote-proto=ssh', - '--remote-host=localhost'] - - if self.auto_compress_alg and '--compress' in options and \ - self.archive_compress and self.archive_compress != 'zlib': - options = [o if o != '--compress' else f'--compress-algorithm={self.archive_compress}' - for o in options] - - if backup_type: - cmd_list += ['-b', backup_type] - - if not (old_binary or sync): - cmd_list += ['--no-sync'] - - return self.run(cmd_list + options, gdb, old_binary, return_id, env=env, - expect_error=expect_error) - - def backup_replica_node(self, instance, node, data_dir=False, *, - master, backup_type='full', datname=False, - options=None, env=None): - """ - Try to reliably run backup on replica by switching wal at master - at the moment pg_probackup is waiting for archived wal segment - """ - if options is None: - options = [] - assert '--stream' not in options or backup_type == 'page', \ - "backup_replica_node should be used with one of archive-mode or " \ - "page-stream mode" - - options = options.copy() - if not any('--log-level-file' in x for x in options): - options.append('--log-level-file=INFO') - - gdb = self.backup_node( - instance, node, data_dir, - backup_type=backup_type, - datname=datname, - options=options, - env=env, - gdb=True) - gdb.set_breakpoint('wait_wal_lsn') - # we need to break on wait_wal_lsn in pg_stop_backup - gdb.run_until_break() - if backup_type == 'page': - self.switch_wal_segment(master) - if '--stream' not in options: - gdb.continue_execution_until_break() - self.switch_wal_segment(master) - gdb.continue_execution_until_exit() - - 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] - return (backup_id, output) - - def checkdb_node( - self, use_backup_dir=False, instance=False, data_dir=False, - options=None, gdb=False, old_binary=False, - skip_log_directory=False, - expect_error=False - ): - if options is None: - options = [] - cmd_list = ["checkdb"] - - if instance: - cmd_list += ["--instance={0}".format(instance)] - - if data_dir: - cmd_list += ["-D", self._node_dir(data_dir)] - - return self.run(cmd_list + options, gdb, old_binary, - skip_log_directory=skip_log_directory, expect_error=expect_error, - use_backup_dir=use_backup_dir) - - def merge_backup( - self, instance, backup_id, - gdb=False, old_binary=False, options=None, expect_error=False): - if options is None: - options = [] - cmd_list = [ - 'merge', - '--instance={0}'.format(instance), - '-i', backup_id - ] - - return self.run(cmd_list + options, gdb, old_binary, expect_error=expect_error) - - def restore_node( - self, instance, node=None, restore_dir=None, - backup_id=None, old_binary=False, options=None, - gdb=False, - expect_error=False, - sync=False - ): - if options is None: - options = [] - if node: - if isinstance(node, str): - data_dir = node - else: - data_dir = node.data_dir - elif restore_dir: - data_dir = self._node_dir(restore_dir) - else: - raise ValueError("You must provide ether node or base_dir for backup") - - cmd_list = [ - 'restore', - '-D', data_dir, - '--instance={0}'.format(instance) - ] - - # don`t forget to kill old_binary after remote ssh release - if self.remote and not old_binary: - options = options + [ - '--remote-proto=ssh', - '--remote-host=localhost'] - - if backup_id: - cmd_list += ['-i', backup_id] - - if not (old_binary or sync): - cmd_list += ['--no-sync'] - - return self.run(cmd_list + options, gdb=gdb, old_binary=old_binary, expect_error=expect_error) - - def catchup_node( - self, - backup_mode, source_pgdata, destination_node, - options=None, - remote_host='localhost', - remote_port=None, - expect_error=False, - gdb=False - ): - - if options is None: - options = [] - cmd_list = [ - 'catchup', - '--backup-mode={0}'.format(backup_mode), - '--source-pgdata={0}'.format(source_pgdata), - '--destination-pgdata={0}'.format(destination_node.data_dir) - ] - if self.remote: - cmd_list += ['--remote-proto=ssh', f'--remote-host={remote_host}'] - if remote_port: - cmd_list.append(f'--remote-port={remote_port}') - if self.verbose: - if init_params.major_version > 2: - cmd_list += [ - '--log-level-file=trace', - '--log-directory={0}'.format(destination_node.logs_dir) - ] - else: - cmd_list += [ - '--log-level-file=VERBOSE', - '--log-directory={0}'.format(destination_node.logs_dir) - ] - - return self.run(cmd_list + options, gdb=gdb, expect_error=expect_error, use_backup_dir=False) - - def show( - self, instance=None, backup_id=None, - options=None, as_text=False, as_json=True, old_binary=False, - env=None, - expect_error=False, - gdb=False - ): - - if options is None: - options = [] - backup_list = [] - specific_record = {} - cmd_list = [ - 'show', - ] - if instance: - cmd_list += ['--instance={0}'.format(instance)] - - if backup_id: - cmd_list += ['-i', backup_id] - - # AHTUNG, WARNING will break json parsing - if as_json: - cmd_list += ['--format=json', '--log-level-console=error'] - - if as_text: - # You should print it when calling as_text=true - return self.run(cmd_list + options, old_binary=old_binary, env=env, - expect_error=expect_error, gdb=gdb) - - # get show result as list of lines - if as_json: - text_json = str(self.run(cmd_list + options, old_binary=old_binary, env=env, - expect_error=expect_error, gdb=gdb)) - try: - if expect_error: - return text_json - data = json.loads(text_json) - except ValueError: - assert False, f"Couldn't parse {text_json} as json. " \ - f"Check that you don't have additional messages inside the log or use 'as_text=True'" - - for instance_data in data: - # find specific instance if requested - if instance and instance_data['instance'] != instance: - continue - - for backup in reversed(instance_data['backups']): - # find specific backup if requested - if backup_id: - if backup['id'] == backup_id: - return backup - else: - backup_list.append(backup) - - if backup_id is not None: - raise RuntimeError("Failed to find backup with ID: {0}".format(backup_id)) - - return backup_list - else: - show_splitted = self.run(cmd_list + options, old_binary=old_binary, env=env, - expect_error=expect_error).splitlines() - if instance is not None and backup_id is None: - # cut header(ID, Mode, etc) from show as single string - header = show_splitted[1:2][0] - # cut backup records from show as single list - # with string for every backup record - body = show_splitted[3:] - # inverse list so oldest record come first - body = body[::-1] - # split string in list with string for every header element - header_split = re.split(' +', header) - # Remove empty items - for i in header_split: - if i == '': - header_split.remove(i) - continue - header_split = [ - header_element.rstrip() for header_element in header_split - ] - for backup_record in body: - backup_record = backup_record.rstrip() - # split list with str for every backup record element - backup_record_split = re.split(' +', backup_record) - # Remove empty items - for i in backup_record_split: - if i == '': - backup_record_split.remove(i) - if len(header_split) != len(backup_record_split): - logging.error(warning.format( - header=header, body=body, - header_split=header_split, - body_split=backup_record_split) - ) - exit(1) - new_dict = dict(zip(header_split, backup_record_split)) - backup_list.append(new_dict) - return backup_list - else: - # cut out empty lines and lines started with # - # and other garbage then reconstruct it as dictionary - # print show_splitted - sanitized_show = [item for item in show_splitted if item] - sanitized_show = [ - item for item in sanitized_show if not item.startswith('#') - ] - # print sanitized_show - for line in sanitized_show: - name, var = line.partition(' = ')[::2] - var = var.strip('"') - var = var.strip("'") - specific_record[name.strip()] = var - - if not specific_record: - raise RuntimeError("Failed to find backup with ID: {0}".format(backup_id)) - - return specific_record - - def show_archive( - self, instance=None, options=None, - as_text=False, as_json=True, old_binary=False, - tli=0, - expect_error=False - ): - if options is None: - options = [] - cmd_list = [ - 'show', - '--archive', - ] - if instance: - cmd_list += ['--instance={0}'.format(instance)] - - # AHTUNG, WARNING will break json parsing - if as_json: - cmd_list += ['--format=json', '--log-level-console=error'] - - if as_text: - # You should print it when calling as_text=true - return self.run(cmd_list + options, old_binary=old_binary, expect_error=expect_error) - - if as_json: - if as_text: - data = self.run(cmd_list + options, old_binary=old_binary, expect_error=expect_error) - else: - data = json.loads(self.run(cmd_list + options, old_binary=old_binary, expect_error=expect_error)) - - if instance: - instance_timelines = None - for instance_name in data: - if instance_name['instance'] == instance: - instance_timelines = instance_name['timelines'] - break - - if tli > 0: - for timeline in instance_timelines: - if timeline['tli'] == tli: - return timeline - - return {} - - if instance_timelines: - return instance_timelines - - return data - else: - show_splitted = self.run(cmd_list + options, old_binary=old_binary, - expect_error=expect_error).splitlines() - logging.error(show_splitted) - exit(1) - - def validate( - self, instance=None, backup_id=None, - options=None, old_binary=False, gdb=False, expect_error=False - ): - if options is None: - options = [] - cmd_list = [ - 'validate', - ] - if instance: - cmd_list += ['--instance={0}'.format(instance)] - if backup_id: - cmd_list += ['-i', backup_id] - - return self.run(cmd_list + options, old_binary=old_binary, gdb=gdb, - expect_error=expect_error) - - def delete( - self, instance, backup_id=None, - options=None, old_binary=False, gdb=False, expect_error=False): - if options is None: - options = [] - cmd_list = [ - 'delete', - ] - - cmd_list += ['--instance={0}'.format(instance)] - if backup_id: - cmd_list += ['-i', backup_id] - - return self.run(cmd_list + options, old_binary=old_binary, gdb=gdb, - expect_error=expect_error) - - def delete_expired( - self, instance, options=None, old_binary=False, expect_error=False): - if options is None: - options = [] - cmd_list = [ - 'delete', - '--instance={0}'.format(instance) - ] - return self.run(cmd_list + options, old_binary=old_binary, expect_error=expect_error) - - def retention( - self, instance, options=None, old_binary=False, gdb=False, expect_error=False): - if init_params.major_version < 3: - logging.warning("retention command is only available for pg_probackup 3") - return None - if options is None: - options = [] - cmd_list = [ - 'retention', - '--instance={0}'.format(instance), - ] - - return self.run(cmd_list + options, old_binary=old_binary, gdb=gdb, - expect_error=expect_error) - - def show_config(self, instance, old_binary=False, expect_error=False, gdb=False): - out_dict = {} - cmd_list = [ - 'show-config', - '--instance={0}'.format(instance) - ] - - res = self.run(cmd_list, old_binary=old_binary, expect_error=expect_error, gdb=gdb).splitlines() - for line in res: - if not line.startswith('#'): - name, var = line.partition(' = ')[::2] - out_dict[name] = var - return out_dict - - def run_binary(self, command, asynchronous=False, env=None): - - if not env: - env = self.test_env - - if self.verbose: - print([' '.join(map(str, command))]) - try: - if asynchronous: - return subprocess.Popen( - command, - stdin=subprocess.PIPE, - stdout=subprocess.PIPE, - stderr=subprocess.PIPE, - env=env - ) - else: - self.test_class.output = subprocess.check_output( - command, - stderr=subprocess.STDOUT, - env=env - ).decode('utf-8') - return self.test_class.output - except subprocess.CalledProcessError as e: - raise ProbackupException(e.output.decode('utf-8'), command) - - def _node_dir(self, base_dir): - return os.path.join(self.pg_node.test_path, base_dir) - - def set_archiving( - self, instance, node, replica=False, - overwrite=False, compress=True, old_binary=False, - log_level=False, archive_timeout=False, - custom_archive_command=None): - - # parse postgresql.auto.conf - options = {} - if replica: - options['archive_mode'] = 'always' - options['hot_standby'] = 'on' - else: - options['archive_mode'] = 'on' - - if custom_archive_command is None: - archive_command = " ".join([f'"{init_params.probackup_path}"', - 'archive-push', *self.backup_dir.pb_args]) - if os.name == 'posix': - # Dash produces a core dump when it gets a SIGQUIT from its - # child process so replace the shell with pg_probackup - archive_command = 'exec ' + archive_command - elif os.name == "nt": - archive_command = archive_command.replace("\\", "\\\\") - archive_command += f' --instance={instance}' - - # don`t forget to kill old_binary after remote ssh release - if init_params.remote and not old_binary: - archive_command += ' --remote-proto=ssh --remote-host=localhost' - - if init_params.archive_compress and compress: - archive_command += ' --compress-algorithm=' + init_params.archive_compress - - if overwrite: - archive_command += ' --overwrite' - - if init_params.major_version > 2: - archive_command += ' --log-level-console=trace' - else: - archive_command += ' --log-level-console=VERBOSE' - archive_command += ' -j 5' - archive_command += ' --batch-size 10' - - archive_command += ' --no-sync' - - if archive_timeout: - archive_command += f' --archive-timeout={archive_timeout}' - - if os.name == 'posix': - archive_command += ' --wal-file-path=%p --wal-file-name=%f' - - elif os.name == 'nt': - archive_command += ' --wal-file-path="%p" --wal-file-name="%f"' - - if log_level: - archive_command += f' --log-level-console={log_level}' - else: # custom_archive_command is not None - archive_command = custom_archive_command - 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 - - logging.info("Using ", implementation) - module_name, class_name = implementation.rsplit(sep='.', maxsplit=1) - - module = importlib.import_module(module_name) - - return getattr(module, class_name) - - def archive_push(self, instance, node, wal_file_name, wal_file_path=None, options=None, expect_error=False): - if options is None: - options = [] - cmd = [ - 'archive-push', - '--instance={0}'.format(instance), - '--wal-file-name={0}'.format(wal_file_name), - ] - if wal_file_path is None: - cmd = cmd + ['--wal-file-path={0}'.format(os.path.join(node.data_dir, 'pg_wal'))] - else: - cmd = cmd + ['--wal-file-path={0}'.format(wal_file_path)] - return self.run(cmd + options, expect_error=expect_error) - - def archive_get(self, instance, wal_file_name, wal_file_path, options=None, expect_error=False): - if options is None: - options = [] - cmd = [ - 'archive-get', - '--instance={0}'.format(instance), - '--wal-file-name={0}'.format(wal_file_name), - '--wal-file-path={0}'.format(wal_file_path), - ] - return self.run(cmd + options, expect_error=expect_error) - - def maintain( - self, instance=None, backup_id=None, - options=None, old_binary=False, gdb=False, expect_error=False - ): - if options is None: - options = [] - cmd_list = [ - 'maintain', - ] - if instance: - cmd_list += ['--instance={0}'.format(instance)] - if backup_id: - cmd_list += ['-i', backup_id] - - return self.run(cmd_list + options, old_binary=old_binary, gdb=gdb, - expect_error=expect_error) - - def build_backup_dir(self, backup='backup'): - return fs_backup_class(rel_path=self.rel_path, backup=backup) diff --git a/pg_probackup2/gdb.py b/pg_probackup2/gdb.py deleted file mode 100644 index 08c1f2e..0000000 --- a/pg_probackup2/gdb.py +++ /dev/null @@ -1,368 +0,0 @@ -import functools -import os -import subprocess -import sys -import unittest -from time import sleep - - -class GdbException(Exception): - def __init__(self, message="False"): - self.message = message - - def __str__(self): - return '\n ERROR: {0}\n'.format(repr(self.message)) - - -class GDBobj: - _gdb_enabled = False - _gdb_ok = False - _gdb_ptrace_ok = False - - def __init__(self, cmd, env, attach=False): - self.verbose = env.verbose - self.output = '' - self._did_quit = False - self.has_breakpoint = False - - # Check gdb flag is set up - if not hasattr(env, "_gdb_decorated") or not env._gdb_decorated: - raise GdbException("Test should be decorated with @needs_gdb") - if not self._gdb_enabled: - raise GdbException("No `PGPROBACKUP_GDB=on` is set.") - if not self._gdb_ok: - if not self._gdb_ptrace_ok: - raise GdbException("set /proc/sys/kernel/yama/ptrace_scope to 0" - " to run GDB tests") - raise GdbException("No gdb usage possible.") - - # Check gdb presence - try: - gdb_version, _ = subprocess.Popen( - ['gdb', '--version'], - stdout=subprocess.PIPE - ).communicate() - except OSError: - raise GdbException("Couldn't find gdb on the path") - - self.base_cmd = [ - 'gdb', - '--interpreter', - 'mi2', - ] - - if attach: - self.cmd = self.base_cmd + ['--pid'] + cmd - else: - self.cmd = self.base_cmd + ['--args'] + cmd - - if self.verbose: - print([' '.join(map(str, self.cmd))]) - - self.proc = subprocess.Popen( - self.cmd, - stdin=subprocess.PIPE, - stdout=subprocess.PIPE, - stderr=subprocess.STDOUT, - bufsize=0, - text=True, - errors='replace', - ) - self.gdb_pid = self.proc.pid - - while True: - line = self.get_line() - - if 'No such process' in line: - raise GdbException(line) - - if not line.startswith('(gdb)'): - pass - else: - break - - def __del__(self): - if not self._did_quit and hasattr(self, "proc"): - try: - self.quit() - except subprocess.TimeoutExpired: - self.kill() - - def get_line(self): - line = self.proc.stdout.readline() - self.output += line - return line - - def kill(self): - self._did_quit = True - self.proc.kill() - self.proc.wait(3) - self.proc.stdin.close() - self.proc.stdout.close() - - def terminate_subprocess(self): - self._execute('kill') - - def set_breakpoint(self, location): - - result = self._execute('break ' + location) - self.has_breakpoint = True - for line in result: - if line.startswith('~"Breakpoint'): - return - - elif line.startswith('=breakpoint-created'): - return - - elif line.startswith('^error'): # or line.startswith('(gdb)'): - break - - elif line.startswith('&"break'): - pass - - elif line.startswith('&"Function'): - raise GdbException(line) - - elif line.startswith('&"No line'): - raise GdbException(line) - - elif line.startswith('~"Make breakpoint pending on future shared'): - raise GdbException(line) - - raise GdbException( - 'Failed to set breakpoint.\n Output:\n {0}'.format(result) - ) - - def remove_all_breakpoints(self): - if self._did_quit: - return - if not self.has_breakpoint: - return - - result = self._execute('delete') - self.has_breakpoint = False - for line in result: - - if line.startswith('^done'): - return - - raise GdbException( - 'Failed to remove breakpoints.\n Output:\n {0}'.format(result) - ) - - def run_until_break(self): - if self._did_quit: - return - result = self._execute('run', False) - for line in result: - if line.startswith('*stopped,reason="breakpoint-hit"'): - return - raise GdbException( - 'Failed to run until breakpoint.\n' - ) - - def continue_execution_until_running(self): - if self._did_quit: - return - result = self._execute('continue') - - for line in result: - if line.startswith('*running') or line.startswith('^running'): - return - if line.startswith('*stopped,reason="breakpoint-hit"'): - continue - if line.startswith('*stopped,reason="exited-normally"'): - continue - - raise GdbException( - 'Failed to continue execution until running.\n' - ) - - def signal(self, sig): - if 'KILL' in sig: - self.remove_all_breakpoints() - # Avoid GDB 15.x SIGABRT: kill inferior directly, not via `signal`. - try: - pids = subprocess.check_output( - ['pgrep', '-P', str(self.gdb_pid)], - text=True - ).strip().split() - for pid in pids: - os.kill(int(pid), 9) - except (subprocess.CalledProcessError, ValueError, OSError): - pass - self.kill() - return - self._execute(f'signal {sig}') - - def continue_execution_until_exit(self): - if self._did_quit: - return - self.remove_all_breakpoints() - result = self._execute('continue', False) - - for line in result: - if line.startswith('*running'): - continue - if line.startswith('*stopped,reason="breakpoint-hit"'): - continue - if line.startswith('*stopped,reason="exited') or line == '*stopped\n': - self.quit() - return - - raise GdbException( - 'Failed to continue execution until exit.\n' - ) - - def continue_execution_until_error(self): - if self._did_quit: - return - self.remove_all_breakpoints() - result = self._execute('continue', False) - - for line in result: - if line.startswith('^error'): - return - if line.startswith('*stopped,reason="exited'): - return - if line.startswith( - '*stopped,reason="signal-received",signal-name="SIGABRT"'): - return - - raise GdbException( - 'Failed to continue execution until error.\n') - - def continue_execution_until_break(self, ignore_count=0): - if self._did_quit: - return - if ignore_count > 0: - result = self._execute( - 'continue ' + str(ignore_count), - False - ) - else: - result = self._execute('continue', False) - - for line in result: - if line.startswith('*stopped,reason="breakpoint-hit"'): - return - if line.startswith('*stopped,reason="exited-normally"'): - break - - raise GdbException( - 'Failed to continue execution until break.\n') - - def show_backtrace(self): - return self._execute("backtrace", running=False) - - def stopped_in_breakpoint(self): - while True: - line = self.get_line() - if self.verbose: - print(line) - if line.startswith('*stopped,reason="breakpoint-hit"'): - return True - - def detach(self): - if not self._did_quit: - self._execute('detach') - - def quit(self): - if not self._did_quit: - self._did_quit = True - self.proc.terminate() - self.proc.wait(3) - self.proc.stdin.close() - self.proc.stdout.close() - - # use for breakpoint, run, continue - def _execute(self, cmd, running=True): - if self._did_quit: - return [] - output = [] - self.proc.stdin.flush() - self.proc.stdin.write(cmd + '\n') - self.proc.stdin.flush() - sleep(1) - - # look for command we just send - while True: - line = self.get_line() - if self.verbose: - print(repr(line)) - - if cmd not in line: - continue - else: - break - - while True: - line = self.get_line() - output += [line] - if self.verbose: - print(repr(line)) - if line.startswith('^done') or line.startswith('*stopped'): - break - if line.startswith('^error'): - break - if running and (line.startswith('*running') or line.startswith('^running')): - # if running and line.startswith('*running'): - break - return output - - -def _set_gdb(self): - test_env = os.environ.copy() - self._gdb_enabled = test_env.get('PGPROBACKUP_GDB') == 'ON' - self._gdb_ok = self._gdb_enabled - if not self._gdb_enabled or sys.platform != 'linux': - return - try: - with open('/proc/sys/kernel/yama/ptrace_scope') as f: - ptrace = f.read() - except FileNotFoundError: - self._gdb_ptrace_ok = True - return - self._gdb_ptrace_ok = int(ptrace) == 0 - self._gdb_ok = self._gdb_ok and self._gdb_ptrace_ok - - -def _check_gdb_flag_or_skip_test(): - if not GDBobj._gdb_enabled: - return ("skip", - "Specify PGPROBACKUP_GDB and build without " - "optimizations for run this test" - ) - if GDBobj._gdb_ok: - return None - if not GDBobj._gdb_ptrace_ok: - return ("fail", "set /proc/sys/kernel/yama/ptrace_scope to 0" - " to run GDB tests") - else: - return ("fail", "use of gdb is not possible") - - -def needs_gdb(func): - check = _check_gdb_flag_or_skip_test() - if not check: - @functools.wraps(func) - def ok_wrapped(self, *args, **kwargs): - self._gdb_decorated = True - return func(self, *args, **kwargs) - - return ok_wrapped - reason = check[1] - if check[0] == "skip": - return unittest.skip(reason)(func) - - if check[0] == "fail": - @functools.wraps(func) - def fail_wrapper(self, *args, **kwargs): - self.fail(reason) - - return fail_wrapper - - raise RuntimeError("Wrong action {0}".format(check)) - - -_set_gdb(GDBobj) diff --git a/pg_probackup2/init_helpers.py b/pg_probackup2/init_helpers.py deleted file mode 100644 index fe29482..0000000 --- a/pg_probackup2/init_helpers.py +++ /dev/null @@ -1,465 +0,0 @@ -import logging -from functools import reduce -import getpass -import os -import pefile -import re -import shutil -import subprocess -import sys -import traceback -import testgres -import typing - -try: - import lz4.frame # noqa: F401 - - HAVE_LZ4 = True -except ImportError as e: - HAVE_LZ4 = False - LZ4_error = e - -try: - import zstd # noqa: F401 - - HAVE_ZSTD = True -except ImportError as e: - HAVE_ZSTD = False - ZSTD_error = e - -delete_logs = os.getenv('KEEP_LOGS') not in ['1', 'y', 'Y'] - -try: - testgres.configure_testgres( - cache_initdb=False, - cached_initdb_dir=False, - node_cleanup_full=delete_logs) -except Exception as e: - logging.warning("Can't configure testgres: {0}".format(e)) - - -class InitData_ServerProps: - pgpro_edition: typing.Optional[str] - num_version: int - - def __init__( - self, - pgpro_edition: typing.Optional[str], - num_version: int, - ): - assert pgpro_edition is None or type(pgpro_edition) is str - assert type(num_version) is int - self.pgpro_edition = pgpro_edition - self.num_version = num_version - return - - -class InitData_BinaryName: - pg_config: str - postgres: str - initdb: str - - def __init__( - self, - pg_config: str, - initdb: str, - postgres: str, - ): - assert type(pg_config) is str - assert type(postgres) is str - assert type(initdb) is str - - self.pg_config = pg_config - self.postgres = postgres - self.initdb = initdb - return - - -class Init(object): - _server_props: InitData_ServerProps - - def __init__(self): - if '-v' in sys.argv or '--verbose' in sys.argv: - self.verbose = True - else: - self.verbose = False - - os_ops = testgres.LocalOperations() - os_name = os_ops.get_name() - - pg_bin = testgres.get_bin_dir(os_ops) - if pg_bin is None: - raise testgres.InvalidOperationException( - "Failed to determine the Postgres binary directory. Specify the path to the directory in PG_BIN or put it into the system PATH.", - ) - - if not os_ops.path_exists(pg_bin): - raise testgres.InvalidOperationException( - "Path [{}] is not found.".format(pg_bin), - ) - - pg_config_path = os_ops.build_path( - pg_bin, - __class__._get_binary_names(os_ops).pg_config, - ) - - assert type(pg_config_path) is str - - if os_ops.path_exists(pg_config_path): - self._server_props = __class__._init__get_server_props__via_pg_config( - os_ops, - pg_config_path, - ) - else: - self._server_props = __class__._init__get_server_props__via_server( - os_ops, - pg_bin, - ) - - # TODO: Always test with NLS support and remove this flag - self.is_nls_enabled = True - - postgres = os_ops.build_path(pg_bin, 'postgres.exe' if os_name == 'nt' else 'postgres') - - if os_name == 'posix': - ldd = os_ops.exec_command( - ['ldd', postgres], - encoding='utf-8') - self.is_lz4_enabled = 'liblz4.so' in ldd - elif os_name == 'nt': - pe = pefile.PE(postgres, fast_load=True) - pe.parse_data_directories(directories=[ - pefile.DIRECTORY_ENTRY['IMAGE_DIRECTORY_ENTRY_IMPORT']]) - self.is_lz4_enabled = False - for entry in pe.DIRECTORY_ENTRY_IMPORT: - if entry.dll.decode('utf-8') == 'liblz4.dll': - self.is_lz4_enabled = True - break - else: - # Fall back to trying to determine lz4 support via pg_config - pg_config_data = testgres.utils.get_pg_config2( - pg_config_path=pg_config_path, - os_ops=os_ops, - ) - self.is_lz4_enabled = '-llz4' in pg_config_data['LIBS'] - - os.environ['LANGUAGE'] = 'en' # set default locale language to en. All messages will use this locale - test_env = os.environ.copy() - envs_list = [ - 'LANGUAGE', - 'LC_ALL', - 'PGCONNECT_TIMEOUT', - 'PGDATA', - 'PGDATABASE', - 'PGHOSTADDR', - 'PGREQUIRESSL', - 'PGSERVICE', - 'PGSSLMODE', - 'PGUSER', - 'PGPORT', - 'PGHOST' - ] - - for e in envs_list: - test_env.pop(e, None) - - test_env['LC_MESSAGES'] = 'C' - test_env['LC_TIME'] = 'C' - self._test_env = test_env - - # Get the directory from which the script was executed - self.source_path = os.getcwd() - tmp_path = test_env.get('PGPROBACKUP_TMP_DIR') - if tmp_path and os.path.isabs(tmp_path): - self.tmp_path = tmp_path - else: - self.tmp_path = os.path.abspath( - os.path.join(self.source_path, tmp_path or os.path.join('tests', 'tmp_dirs')) - ) - - os.makedirs(self.tmp_path, exist_ok=True) - - self.username = getpass.getuser() - - self.probackup_path = None - if 'PGPROBACKUPBIN' in test_env: - if shutil.which(test_env["PGPROBACKUPBIN"]): - self.probackup_path = test_env["PGPROBACKUPBIN"] - else: - raise Exception( - 'pg_probackup binary not found at PGPROBACKUPBIN={0}'.format( - test_env["PGPROBACKUPBIN"])) - - if not self.probackup_path: - probackup_path_tmp = os_ops.build_path(pg_bin, 'pg_probackup') - - if os.path.isfile(probackup_path_tmp): - if not os.access(probackup_path_tmp, os.X_OK): - logging.warning('{0} is not an executable file'.format( - probackup_path_tmp)) - else: - self.probackup_path = probackup_path_tmp - - if not self.probackup_path: - probackup_path_tmp = self.source_path - - if os.path.isfile(probackup_path_tmp): - if not os.access(probackup_path_tmp, os.X_OK): - logging.warning('{0} is not an executable file'.format( - probackup_path_tmp)) - else: - self.probackup_path = probackup_path_tmp - - if not self.probackup_path: - raise Exception('pg_probackup binary is not found, add it to PATH or point PGPROBACKUPBIN') - - if os_name == 'posix': - self.EXTERNAL_DIRECTORY_DELIMITER = ':' - os.environ['PATH'] = os.path.dirname( - self.probackup_path) + ':' + os.environ['PATH'] - - elif os_name == 'nt': - self.EXTERNAL_DIRECTORY_DELIMITER = ';' - os.environ['PATH'] = os.path.dirname( - self.probackup_path) + ';' + os.environ['PATH'] - - self.probackup_old_path = None - if 'PGPROBACKUPBIN_OLD' in test_env: - if (os.path.isfile(test_env['PGPROBACKUPBIN_OLD']) and os.access(test_env['PGPROBACKUPBIN_OLD'], os.X_OK)): - self.probackup_old_path = test_env['PGPROBACKUPBIN_OLD'] - else: - if self.verbose: - print('PGPROBACKUPBIN_OLD is not an executable file') - - self.probackup_version = None - self.old_probackup_version = None - - probackup_version_output = subprocess.check_output( - [self.probackup_path, "--version"], - stderr=subprocess.STDOUT, - ).decode('utf-8') - match = re.search(r"\d+\.\d+\.\d+", - probackup_version_output) - self.probackup_version = match.group(0) if match else None - match = re.search(r"\(compressions: ([^)]*)\)", probackup_version_output) - compressions = match.group(1) if match else None - if compressions: - self.probackup_compressions = {s.strip() for s in compressions.split(',')} - else: - self.probackup_compressions = [] - - if self.probackup_old_path: - old_probackup_version_output = subprocess.check_output( - [self.probackup_old_path, "--version"], - stderr=subprocess.STDOUT, - ).decode('utf-8') - match = re.search(r"\d+\.\d+\.\d+", - old_probackup_version_output) - self.old_probackup_version = match.group(0) if match else None - - self.remote = test_env.get('PGPROBACKUP_SSH_REMOTE', None) == 'ON' - self.ptrack = test_env.get('PG_PROBACKUP_PTRACK', None) == 'ON' and self._server_props.num_version >= 110000 - self.wal_tree_enabled = test_env.get('PG_PROBACKUP_WAL_TREE_ENABLED', None) == 'ON' - - self.bckp_source = test_env.get('PG_PROBACKUP_SOURCE', 'pro').lower() - if self.bckp_source not in ('base', 'direct', 'pro'): - raise Exception("Wrong PG_PROBACKUP_SOURCE value. Available options: base|direct|pro") - - self.paranoia = test_env.get('PG_PROBACKUP_PARANOIA', None) == 'ON' - env_compress = test_env.get('ARCHIVE_COMPRESSION', None) - if env_compress: - env_compress = env_compress.lower() - if env_compress in ('on', 'zlib'): - self.compress_suffix = '.gz' - self.archive_compress = 'zlib' - elif env_compress == 'lz4': - if not HAVE_LZ4: - raise LZ4_error - if 'lz4' not in self.probackup_compressions: - raise Exception("pg_probackup is not compiled with lz4 support") - self.compress_suffix = '.lz4' - self.archive_compress = 'lz4' - elif env_compress == 'zstd': - if not HAVE_ZSTD: - raise ZSTD_error - if 'zstd' not in self.probackup_compressions: - raise Exception("pg_probackup is not compiled with zstd support") - self.compress_suffix = '.zst' - self.archive_compress = 'zstd' - else: - self.compress_suffix = '' - self.archive_compress = False - - cfs_compress = test_env.get('PG_PROBACKUP_CFS_COMPRESS', None) - if cfs_compress: - self.cfs_compress = cfs_compress.lower() - else: - self.cfs_compress = self.archive_compress - - os.environ["PGAPPNAME"] = "pg_probackup" - self.delete_logs = delete_logs - - if self.probackup_version.split('.')[0].isdigit(): - self.major_version = int(self.probackup_version.split('.')[0]) - else: - raise Exception('Can\'t process pg_probackup version \"{}\": the major version is expected to be a number'.format(self.probackup_version)) - - self.valgrind = test_env.get('PG_PROBACKUP_VALGRIND') - self.valgrind_sup_path = test_env.get('PG_PROBACKUP_VALGRIND_SUP') - - def test_env(self): - return self._test_env.copy() - - @property - def is_enterprise(self) -> bool: - assert type(self._server_props) is InitData_ServerProps - return self._server_props.pgpro_edition == 'enterprise' - - @property - def is_shardman(self) -> bool: - assert type(self._server_props) is InitData_ServerProps - return self._server_props.pgpro_edition == 'shardman' - - @property - def is_pgpro(self) -> bool: - assert type(self._server_props) is InitData_ServerProps - return self._server_props.pgpro_edition is not None - - @property - def pg_config_version(self) -> int: - assert type(self._server_props) is InitData_ServerProps - return self._server_props.num_version - - @staticmethod - def _init__get_server_props__via_pg_config( - os_ops: testgres.OsOperations, - pg_config_path: str, - ) -> InitData_ServerProps: - assert isinstance(os_ops, testgres.OsOperations) - assert type(pg_config_path) is str - - pg_config = testgres.utils.get_pg_config2( - pg_config_path=pg_config_path, - os_ops=os_ops, - ) - assert type(pg_config) is dict - - pgpro_edition = pg_config.get('PGPRO_EDITION') - assert pgpro_edition is None or type(pgpro_edition) is str - - version_str = pg_config.get('VERSION', '') - assert type(version_str) is str - - if not version_str: - raise RuntimeError("Field 'VERSION' not found in pg_config output") - - version_num = testgres.parse_pg_version(version_str) - parts = [*version_num.split('.'), '0', '0'][:3] - parts[0] = re.match(r'\d+', parts[0]).group() - - num_version = reduce(lambda v, x: v * 100 + int(x), parts, 0) - assert type(num_version) is int - - return InitData_ServerProps( - pgpro_edition=pgpro_edition, - num_version=num_version, - ) - - @staticmethod - def _init__get_server_props__via_server( - os_ops: testgres.OsOperations, - pg_bin_dir: str, - ) -> InitData_ServerProps: - assert isinstance(os_ops, testgres.OsOperations) - assert type(pg_bin_dir) is str - - tmpdir = os_ops.mkdtemp() - assert type(tmpdir) is str - assert os_ops.path_exists(tmpdir) - assert os_ops.is_abs_path(tmpdir) - - try: - initdb = os_ops.build_path( - pg_bin_dir, - __class__._get_binary_names(os_ops).initdb, - ) - - os_ops.exec_command( - [initdb, "-D", tmpdir], - encoding='utf-8', - ) - - postgres = os_ops.build_path( - pg_bin_dir, - __class__._get_binary_names(os_ops).postgres, - ) - - pgpro_edition_out: typing.Optional[str] = None - try: - exec_r = os_ops.exec_command( - [postgres, "-C", "pgpro_edition", "-D", tmpdir], - encoding='utf-8', - ) - assert type(exec_r) is str - pgpro_edition_out = exec_r.strip() - except testgres.ExecUtilException as e: - assert e.exit_code != 0 - - logging.debug("Exception ({}): {}".format( - type(e).__name__, - e, - )) - - assert pgpro_edition_out is None or type(pgpro_edition_out) is str - pgpro_edition = pgpro_edition_out - - server_version_out = os_ops.exec_command( - [postgres, "-C", "server_version", "-D", tmpdir], - encoding='utf-8' - ) - assert type(server_version_out) is str - server_version = testgres.parse_pg_version(server_version_out) - parts = [*server_version.split('.'), '0', '0'][:3] - parts[0] = re.match(r'\d+', parts[0]).group() - # Server_version consists of two fields (x.y) so num_version always ends with 00 - pg_config_version = reduce(lambda v, x: v * 100 + int(x), parts, 0) - finally: - os_ops.rmdirs(tmpdir) - - return InitData_ServerProps( - pgpro_edition=pgpro_edition, - num_version=pg_config_version, - ) - - sm_binary_names__win32 = InitData_BinaryName( - pg_config="pg_config.exe", - initdb="initdb.exe", - postgres="postgres.exe", - ) - - sm_binary_names__linux = InitData_BinaryName( - pg_config="pg_config", - initdb="initdb", - postgres="postgres", - ) - - @staticmethod - def _get_binary_names( - os_ops: testgres.OsOperations, - ) -> InitData_BinaryName: - assert isinstance(os_ops, testgres.OsOperations) - - platform_name = os_ops.get_platform() - if platform_name == "win32": - return __class__.sm_binary_names__win32 - - return __class__.sm_binary_names__linux - - -try: - init_params = Init() -except Exception as e: - traceback.print_exc(file=sys.stderr) - logging.error(str(e)) - logging.warning("testgres.plugins.probackup2.init_params is set to None.") - init_params = None diff --git a/pg_probackup2/storage/__init__.py b/pg_probackup2/storage/__init__.py deleted file mode 100644 index e69de29..0000000 diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..575fa7b --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,61 @@ +[build-system] +requires = ["setuptools>=61.0"] +build-backend = "setuptools.build_meta" + +[tool.setuptools.package-dir] +"pg_probackup2" = "src" + +[tool.flake8] +extend-ignore = ["E501"] +exclude = [".git", "__pycache__", "env", "venv"] + +[tool.pytest.ini_options] +testpaths = ["tests"] +log_file_level = "NOTSET" +log_file_format = "%(levelname)8s [%(asctime)s] %(message)s" +log_file_date_format = "%Y-%m-%d %H:%M:%S" + +[project] +name = "testgres_pg_probackup2" +version = "0.10.0" +description = "Plugin for testgres that manages pg_probackup2" +readme = "README.md" +requires-python = ">=3.7.3" + +# Формат для совместимости с Python 3.7 +license = {text = "PostgreSQL"} + +authors = [ + {name = "Postgres Professional", email = "testgres@postgrespro.ru"}, +] + +keywords = [ + "pg_probackup", + "testing", + "testgres", +] + +classifiers = [ + "Intended Audience :: Developers", + "Operating System :: Unix", + "Programming Language :: Python :: 3 :: Only", + "Programming Language :: Python :: 3.7", + "Programming Audience :: Python :: 3.8", + "Programming Language :: Python :: 3.9", + "Programming Language :: Python :: 3.10", + "Programming Language :: Python :: 3.11", + "Programming Language :: Python :: 3.12", + "Programming Language :: Python :: 3.13", + "Topic :: Software Development :: Libraries", + "Topic :: Software Development :: Testing", +] + +dependencies = [ + "lz4", + "pefile", + "testgres>=1.14.4", + "zstd", +] + +[project.urls] +"HomePage" = "https://github.com/postgrespro/testgres-pg_probackup2" diff --git a/setup.py b/setup.py deleted file mode 100644 index 3fbd50d..0000000 --- a/setup.py +++ /dev/null @@ -1,23 +0,0 @@ -try: - from setuptools import setup -except ImportError: - from distutils.core import setup - -setup( - version='0.9.2', - name='testgres_pg_probackup2', - packages=['pg_probackup2', 'pg_probackup2.storage'], - description='Plugin for testgres that manages pg_probackup2', - url='https://github.com/postgrespro/testgres-pg_probackup2', - long_description_content_type='text/markdown', - license='PostgreSQL', - author='Postgres Professional', - author_email='testgres@postgrespro.ru', - keywords=['pg_probackup', 'testing', 'testgres'], - install_requires=[ - 'lz4', - 'pefile', - 'testgres>=1.14.4', - 'zstd' - ] -) diff --git a/pg_probackup2/__init__.py b/src/storage/__init__.py similarity index 100% rename from pg_probackup2/__init__.py rename to src/storage/__init__.py diff --git a/pg_probackup2/storage/fs_backup.py b/src/storage/fs_backup.py similarity index 100% rename from pg_probackup2/storage/fs_backup.py rename to src/storage/fs_backup.py diff --git a/tests/test_basic.py b/tests/test_basic.py index 51b24cf..8c19894 100644 --- a/tests/test_basic.py +++ b/tests/test_basic.py @@ -6,9 +6,9 @@ import unittest import testgres -from pg_probackup2.app import ProbackupApp -from pg_probackup2.init_helpers import Init, init_params -from pg_probackup2.storage.fs_backup import FSTestBackupDir +from src.app import ProbackupApp +from src.init_helpers import Init, init_params +from src.storage.fs_backup import FSTestBackupDir class ProbackupTest(unittest.TestCase): diff --git a/tests/test_gdb.py b/tests/test_gdb.py index ba7a929..20643f8 100644 --- a/tests/test_gdb.py +++ b/tests/test_gdb.py @@ -1,7 +1,7 @@ import pytest from unittest import mock -from pg_probackup2 import gdb +from src import gdb def test_needs_gdb_decorator_behavior():