import json import os import shutil import subprocess import sys import tempfile import pymongo import pytest import ruamel.yaml def write_configuration( source_directory, config_path, repository_path, borgmatic_source_directory, postgresql_dump_format='custom', mongodb_dump_format='archive', ): ''' Write out borgmatic configuration into a file at the config path. Set the options so as to work for testing. This includes injecting the given repository path, borgmatic source directory for storing database dumps, dump format (for PostgreSQL), and encryption passphrase. ''' config_yaml = f''' source_directories: - {source_directory} repositories: - path: {repository_path} borgmatic_source_directory: {borgmatic_source_directory} encryption_passphrase: "test" postgresql_databases: - name: test hostname: postgresql username: postgres password: test format: {postgresql_dump_format} - name: all hostname: postgresql username: postgres password: test - name: all format: custom hostname: postgresql username: postgres password: test mariadb_databases: - name: test hostname: mariadb username: root password: test - name: all hostname: mariadb username: root password: test - name: all format: sql hostname: mariadb username: root password: test mysql_databases: - name: test hostname: not-actually-mysql username: root password: test - name: all hostname: not-actually-mysql username: root password: test - name: all format: sql hostname: not-actually-mysql username: root password: test mongodb_databases: - name: test hostname: mongodb username: root password: test authentication_database: admin format: {mongodb_dump_format} - name: all hostname: mongodb username: root password: test sqlite_databases: - name: sqlite_test path: /tmp/sqlite_test.db ''' with open(config_path, 'w') as config_file: config_file.write(config_yaml) return ruamel.yaml.YAML(typ='safe').load(config_yaml) def write_custom_restore_configuration( source_directory, config_path, repository_path, borgmatic_source_directory, postgresql_dump_format='custom', mongodb_dump_format='archive', ): ''' Write out borgmatic configuration into a file at the config path. Set the options so as to work for testing with custom restore options. This includes a custom restore_hostname, restore_port, restore_username, restore_password and restore_path. ''' config_yaml = f''' source_directories: - {source_directory} repositories: - path: {repository_path} borgmatic_source_directory: {borgmatic_source_directory} encryption_passphrase: "test" postgresql_databases: - name: test hostname: postgresql username: postgres password: test format: {postgresql_dump_format} restore_hostname: postgresql2 restore_port: 5433 restore_password: test2 mariadb_databases: - name: test hostname: mariadb username: root password: test restore_hostname: mariadb2 restore_port: 3307 restore_username: root restore_password: test2 mysql_databases: - name: test hostname: not-actually-mysql username: root password: test restore_hostname: not-actually-mysql2 restore_port: 3307 restore_username: root restore_password: test2 mongodb_databases: - name: test hostname: mongodb username: root password: test authentication_database: admin format: {mongodb_dump_format} restore_hostname: mongodb2 restore_port: 27018 restore_username: root2 restore_password: test2 sqlite_databases: - name: sqlite_test path: /tmp/sqlite_test.db restore_path: /tmp/sqlite_test2.db ''' with open(config_path, 'w') as config_file: config_file.write(config_yaml) return ruamel.yaml.YAML(typ='safe').load(config_yaml) def write_simple_custom_restore_configuration( source_directory, config_path, repository_path, borgmatic_source_directory, postgresql_dump_format='custom', ): ''' Write out borgmatic configuration into a file at the config path. Set the options so as to work for testing with custom restore options, but this time using CLI arguments. This includes a custom restore_hostname, restore_port, restore_username and restore_password as we only test these options for PostgreSQL. ''' config_yaml = f''' source_directories: - {source_directory} repositories: - path: {repository_path} borgmatic_source_directory: {borgmatic_source_directory} encryption_passphrase: "test" postgresql_databases: - name: test hostname: postgresql username: postgres password: test format: {postgresql_dump_format} ''' with open(config_path, 'w') as config_file: config_file.write(config_yaml) return ruamel.yaml.YAML(typ='safe').load(config_yaml) def get_connection_params(database, use_restore_options=False): hostname = (database.get('restore_hostname') if use_restore_options else None) or database.get( 'hostname' ) port = (database.get('restore_port') if use_restore_options else None) or database.get('port') username = (database.get('restore_username') if use_restore_options else None) or database.get( 'username' ) password = (database.get('restore_password') if use_restore_options else None) or database.get( 'password' ) return (hostname, port, username, password) def run_postgresql_command(command, config, use_restore_options=False): (hostname, port, username, password) = get_connection_params( config['postgresql_databases'][0], use_restore_options ) subprocess.check_call( [ '/usr/bin/psql', f'--host={hostname}', f'--port={port or 5432}', f"--username={username or 'root'}", f'--command={command}', 'test', ], env={'PGPASSWORD': password}, ) def run_mariadb_command(command, config, use_restore_options=False, binary_name='mariadb'): (hostname, port, username, password) = get_connection_params( config[f'{binary_name}_databases'][0], use_restore_options ) subprocess.check_call( [ f'/usr/bin/{binary_name}', f'--host={hostname}', f'--port={port or 3306}', f'--user={username}', f'--execute={command}', 'test', ], env={'MYSQL_PWD': password}, ) def get_mongodb_database_client(config, use_restore_options=False): (hostname, port, username, password) = get_connection_params( config['mongodb_databases'][0], use_restore_options ) return pymongo.MongoClient(f'mongodb://{username}:{password}@{hostname}:{port or 27017}').test def run_sqlite_command(command, config, use_restore_options=False): database = config['sqlite_databases'][0] path = (database.get('restore_path') if use_restore_options else None) or database.get('path') subprocess.check_call( [ '/usr/bin/sqlite3', path, command, '.exit', ], ) DEFAULT_HOOK_NAMES = {'postgresql', 'mariadb', 'mysql', 'mongodb', 'sqlite'} def create_test_tables(config, use_restore_options=False): ''' Create test tables for borgmatic to dump and backup. ''' command = 'create table test{id} (thing int); insert into test{id} values (1);' if 'postgresql_databases' in config: run_postgresql_command(command.format(id=1), config, use_restore_options) if 'mariadb_databases' in config: run_mariadb_command(command.format(id=2), config, use_restore_options) if 'mysql_databases' in config: run_mariadb_command(command.format(id=3), config, use_restore_options, binary_name='mysql') if 'mongodb_databases' in config: get_mongodb_database_client(config, use_restore_options)['test4'].insert_one({'thing': 1}) if 'sqlite_databases' in config: run_sqlite_command(command.format(id=5), config, use_restore_options) def drop_test_tables(config, use_restore_options=False): ''' Drop the test tables in preparation for borgmatic restoring them. ''' command = 'drop table if exists test{id};' if 'postgresql_databases' in config: run_postgresql_command(command.format(id=1), config, use_restore_options) if 'mariadb_databases' in config: run_mariadb_command(command.format(id=2), config, use_restore_options) if 'mysql_databases' in config: run_mariadb_command(command.format(id=3), config, use_restore_options, binary_name='mysql') if 'mongodb_databases' in config: get_mongodb_database_client(config, use_restore_options)['test4'].drop() if 'sqlite_databases' in config: run_sqlite_command(command.format(id=5), config, use_restore_options) def select_test_tables(config, use_restore_options=False): ''' Select the test tables to make sure they exist. Raise if the expected tables cannot be selected, for instance if a restore hasn't worked as expected. ''' command = 'select count(*) from test{id};' if 'postgresql_databases' in config: run_postgresql_command(command.format(id=1), config, use_restore_options) if 'mariadb_databases' in config: run_mariadb_command(command.format(id=2), config, use_restore_options) if 'mysql_databases' in config: run_mariadb_command(command.format(id=3), config, use_restore_options, binary_name='mysql') if 'mongodb_databases' in config: assert ( get_mongodb_database_client(config, use_restore_options)['test4'].count_documents( filter={} ) > 0 ) if 'sqlite_databases' in config: run_sqlite_command(command.format(id=5), config, use_restore_options) def test_database_dump_and_restore(): # Create a Borg repository. temporary_directory = tempfile.mkdtemp() repository_path = os.path.join(temporary_directory, 'test.borg') borgmatic_source_directory = os.path.join(temporary_directory, '.borgmatic') # Write out a special file to ensure that it gets properly excluded and Borg doesn't hang on it. os.mkfifo(os.path.join(temporary_directory, 'special_file')) original_working_directory = os.getcwd() try: config_path = os.path.join(temporary_directory, 'test.yaml') config = write_configuration( temporary_directory, config_path, repository_path, borgmatic_source_directory ) create_test_tables(config) select_test_tables(config) subprocess.check_call( ['borgmatic', '-v', '2', '--config', config_path, 'rcreate', '--encryption', 'repokey'] ) # Run borgmatic to generate a backup archive including database dumps. subprocess.check_call(['borgmatic', 'create', '--config', config_path, '-v', '2']) # Get the created archive name. output = subprocess.check_output( ['borgmatic', '--config', config_path, 'list', '--json'] ).decode(sys.stdout.encoding) parsed_output = json.loads(output) assert len(parsed_output) == 1 assert len(parsed_output[0]['archives']) == 1 archive_name = parsed_output[0]['archives'][0]['archive'] # Restore the databases from the archive. drop_test_tables(config) subprocess.check_call( ['borgmatic', '-v', '2', '--config', config_path, 'restore', '--archive', archive_name] ) # Ensure the test tables have actually been restored. select_test_tables(config) finally: os.chdir(original_working_directory) shutil.rmtree(temporary_directory) drop_test_tables(config) def test_database_dump_and_restore_with_restore_cli_flags(): # Create a Borg repository. temporary_directory = tempfile.mkdtemp() repository_path = os.path.join(temporary_directory, 'test.borg') borgmatic_source_directory = os.path.join(temporary_directory, '.borgmatic') original_working_directory = os.getcwd() try: config_path = os.path.join(temporary_directory, 'test.yaml') config = write_simple_custom_restore_configuration( temporary_directory, config_path, repository_path, borgmatic_source_directory ) create_test_tables(config) select_test_tables(config) subprocess.check_call( ['borgmatic', '-v', '2', '--config', config_path, 'rcreate', '--encryption', 'repokey'] ) # Run borgmatic to generate a backup archive including a database dump. subprocess.check_call(['borgmatic', 'create', '--config', config_path, '-v', '2']) # Get the created archive name. output = subprocess.check_output( ['borgmatic', '--config', config_path, 'list', '--json'] ).decode(sys.stdout.encoding) parsed_output = json.loads(output) assert len(parsed_output) == 1 assert len(parsed_output[0]['archives']) == 1 archive_name = parsed_output[0]['archives'][0]['archive'] # Restore the database from the archive. drop_test_tables(config) subprocess.check_call( [ 'borgmatic', '-v', '2', '--config', config_path, 'restore', '--archive', archive_name, '--hostname', 'postgresql2', '--port', '5433', '--password', 'test2', ] ) # Ensure the test tables have actually been restored. But first modify the config to contain # the altered restore values from the borgmatic command above. This ensures that the test # tables are selected from the correct database. database = config['postgresql_databases'][0] database['restore_hostname'] = 'postgresql2' database['restore_port'] = '5433' database['restore_password'] = 'test2' select_test_tables(config, use_restore_options=True) finally: os.chdir(original_working_directory) shutil.rmtree(temporary_directory) drop_test_tables(config) drop_test_tables(config, use_restore_options=True) def test_database_dump_and_restore_with_restore_configuration_options(): # Create a Borg repository. temporary_directory = tempfile.mkdtemp() repository_path = os.path.join(temporary_directory, 'test.borg') borgmatic_source_directory = os.path.join(temporary_directory, '.borgmatic') original_working_directory = os.getcwd() try: config_path = os.path.join(temporary_directory, 'test.yaml') config = write_custom_restore_configuration( temporary_directory, config_path, repository_path, borgmatic_source_directory ) create_test_tables(config) select_test_tables(config) subprocess.check_call( ['borgmatic', '-v', '2', '--config', config_path, 'rcreate', '--encryption', 'repokey'] ) # Run borgmatic to generate a backup archive including a database dump. subprocess.check_call(['borgmatic', 'create', '--config', config_path, '-v', '2']) # Get the created archive name. output = subprocess.check_output( ['borgmatic', '--config', config_path, 'list', '--json'] ).decode(sys.stdout.encoding) parsed_output = json.loads(output) assert len(parsed_output) == 1 assert len(parsed_output[0]['archives']) == 1 archive_name = parsed_output[0]['archives'][0]['archive'] # Restore the database from the archive. drop_test_tables(config) subprocess.check_call( ['borgmatic', '-v', '2', '--config', config_path, 'restore', '--archive', archive_name] ) # Ensure the test tables have actually been restored. select_test_tables(config, use_restore_options=True) finally: os.chdir(original_working_directory) shutil.rmtree(temporary_directory) drop_test_tables(config) drop_test_tables(config, use_restore_options=True) def test_database_dump_and_restore_with_directory_format(): # Create a Borg repository. temporary_directory = tempfile.mkdtemp() repository_path = os.path.join(temporary_directory, 'test.borg') borgmatic_source_directory = os.path.join(temporary_directory, '.borgmatic') original_working_directory = os.getcwd() try: config_path = os.path.join(temporary_directory, 'test.yaml') config = write_configuration( temporary_directory, config_path, repository_path, borgmatic_source_directory, postgresql_dump_format='directory', mongodb_dump_format='directory', ) create_test_tables(config) select_test_tables(config) subprocess.check_call( ['borgmatic', '-v', '2', '--config', config_path, 'rcreate', '--encryption', 'repokey'] ) # Run borgmatic to generate a backup archive including a database dump. subprocess.check_call(['borgmatic', 'create', '--config', config_path, '-v', '2']) # Restore the database from the archive. drop_test_tables(config) subprocess.check_call( ['borgmatic', '--config', config_path, 'restore', '--archive', 'latest'] ) # Ensure the test tables have actually been restored. select_test_tables(config) finally: os.chdir(original_working_directory) shutil.rmtree(temporary_directory) drop_test_tables(config) def test_database_dump_with_error_causes_borgmatic_to_exit(): # Create a Borg repository. temporary_directory = tempfile.mkdtemp() repository_path = os.path.join(temporary_directory, 'test.borg') borgmatic_source_directory = os.path.join(temporary_directory, '.borgmatic') original_working_directory = os.getcwd() try: config_path = os.path.join(temporary_directory, 'test.yaml') write_configuration( temporary_directory, config_path, repository_path, borgmatic_source_directory ) subprocess.check_call( ['borgmatic', '-v', '2', '--config', config_path, 'rcreate', '--encryption', 'repokey'] ) # Run borgmatic with a config override such that the database dump fails. with pytest.raises(subprocess.CalledProcessError): subprocess.check_call( [ 'borgmatic', 'create', '--config', config_path, '-v', '2', '--override', "hooks.postgresql_databases=[{'name': 'nope'}]", # noqa: FS003 ] ) finally: os.chdir(original_working_directory) shutil.rmtree(temporary_directory)