"""Exercise deployment backup functions with synthetic commands; never deploy.""" import os from pathlib import Path import shlex import shutil import subprocess import sys import tempfile import unittest ROOT = Path(__file__).resolve().parents[1] BASH = 'C:/Program Files/Git/bin/bash.exe' if os.name == 'nt' else shutil.which('bash') class BackupConnectionTests(unittest.TestCase): def backup(self, connection): source = (ROOT / 'deploy/deploy.sh').read_text(encoding='utf-8') start = source.index('backup_database() {') function = source[start:source.index('\n}\n', start) + 3] with tempfile.TemporaryDirectory() as temporary: folder = Path(temporary) script = '\n'.join([ 'set -euo pipefail', 'BACKUP_DIR=' + shlex.quote(folder.as_posix()), 'BACKUP_STAMP=synthetic', 'DATABASE_PROVIDER=mariadb', 'JOBTRACKER_CONNECTION_STRING=' + shlex.quote(connection), 'python3() { ' + shlex.quote(Path(sys.executable).as_posix()) + ' "$@"; }', 'docker() { echo "Unexpected Docker call" >&2; return 99; }', 'mariadb-dump() { printf "%s\\0" "$@" > "$BACKUP_DIR/args"; ' 'printf "%s" "$MYSQL_PWD" > "$BACKUP_DIR/password"; ' 'printf "CREATE TABLE synthetic (id int);\\n-- Dump completed\\n"; }', 'verify_sql_backup() { gzip -t "$1"; }', function, 'backup_database', ]) result = subprocess.run([BASH, '--noprofile', '--norc'], input=script, text=True, encoding='utf-8', capture_output=True, cwd=ROOT, env={**os.environ, 'DEPLOY_SKIP_DB_BACKUP': 'false'}) arguments = (folder / 'args').read_bytes().split(b'\0')[:-1] if (folder / 'args').exists() else None password = (folder / 'password').read_text(encoding='utf-8') if (folder / 'password').exists() else None return result, arguments, password def test_supported_host_alias_and_case_reach_dump_unchanged(self): for host in ['Server', 'SERVER', 'Host', 'Data Source', 'Address', 'Addr', 'Network Address']: with self.subTest(host=host): result, args, password = self.backup(f'{host}=127.0.0.1;DATABASE=synthetic;UID=reader;PWD=dummy-secret') self.assertEqual(0, result.returncode, result.stderr + result.stdout) self.assertEqual([b'--host=127.0.0.1', b'--port=3306', b'--user=reader', b'--single-transaction', b'--routines', b'--events', b'--quick', b'synthetic'], args) self.assertEqual('dummy-secret', password) self.assertNotIn('dummy-secret', result.stdout + result.stderr) def test_quoted_separators_and_shell_characters_remain_data(self): result, args, password = self.backup('Server=127.0.0.1;Port=3307;Database=synthetic;User ID=reader;Password="semi;eq=quote""$(false)"') self.assertEqual(0, result.returncode, result.stderr + result.stdout) self.assertIn(b'--port=3307', args) self.assertEqual('semi;eq=quote"$(false)', password) def test_empty_password_is_preserved(self): result, args, password = self.backup('Server=127.0.0.1;Database=synthetic;User ID=reader;Password=') self.assertEqual(0, result.returncode, result.stderr + result.stdout) self.assertIsNotNone(args) self.assertEqual('', password) def test_single_quoted_unicode_password_and_database_alias_are_preserved(self): result, args, password = self.backup("Host=127.0.0.1;Initial Catalog=synthetic;Username=reader;Password='søknad你好;it''s';SslMode=VerifyFull") self.assertEqual(0, result.returncode, result.stderr + result.stdout) self.assertEqual(b'synthetic', args[-1]) self.assertEqual("søknad你好;it's", password) def test_invalid_or_ambiguous_input_never_invokes_dump_or_exposes_input(self): for connection in [ 'Host=127.0.0.1;Database=synthetic;Uid=reader;Port=0;Pwd=dummy-secret', 'Host=127.0.0.1;Database=synthetic;Uid=reader;Port=65536;Pwd=dummy-secret', 'Host=127.0.0.1;Server=other;Database=synthetic;Uid=reader;Pwd=dummy-secret', 'Host=127.0.0.1;Database=synthetic;Uid=reader;Pwd="dummy-secret', 'Host=127.0.0.1;Database=synthetic;Uid=reader;Pwd=dummy-secret\nline', 'Host=127.0.0.1;Uid=reader;Pwd=dummy-secret', ]: with self.subTest(connection=connection.replace('dummy-secret', 'REDACTED')): result, args, _ = self.backup(connection) self.assertNotEqual(0, result.returncode) self.assertIsNone(args) self.assertNotIn('dummy-secret', result.stdout + result.stderr) class BackupUploadTests(unittest.TestCase): def upload(self, fail_operation=''): source = (ROOT / 'deploy/deploy.sh').read_text(encoding='utf-8') start = source.index('upload_current_backup_set() {') function = source[start:source.index('\n}\n', start) + 3] gate_start = source.index('if ! apply_backup_retention || ! upload_current_backup_set; then') gate = source[gate_start:source.index('\nfi\n', gate_start) + 4] with tempfile.TemporaryDirectory() as temporary: folder = Path(temporary) for filename in ['jobtracker-db-synthetic.sql.gz', 'jobtracker-files-synthetic.tar.gz', 'jobtracker-db-older.sql.gz']: (folder / filename).touch() (folder / 'monthly' / '2099-01').mkdir(parents=True) script = '\n'.join([ 'set -euo pipefail', 'BACKUP_DIR=' + shlex.quote(folder.as_posix()), 'BACKUP_STAMP=synthetic', 'BACKUP_RCLONE_REMOTE=fixture:encrypted/', 'FAIL_OPERATION=' + shlex.quote(fail_operation), 'date() { echo 2099-01; }', 'apply_backup_retention() { return 0; }', 'rclone() {', ' printf "UPLOAD %s %s %s\\n" "$1" "$3" "$4";', ' if [ "$1" = "$FAIL_OPERATION" ]; then return 23; fi;', '}', function, gate, 'echo "DEPLOY WOULD CONTINUE"', ]) result = subprocess.run([BASH, '--noprofile', '--norc'], input=script, text=True, encoding='utf-8', capture_output=True, cwd=ROOT, timeout=10, env={key: value for key, value in os.environ.items() if key.upper() in {'PATH', 'SYSTEMROOT', 'WINDIR', 'TEMP', 'TMP'}}) calls = [line for line in result.stdout.splitlines() if line.startswith('UPLOAD ')] return result, calls def test_uploads_only_current_daily_files_then_monthly_snapshot(self): result, calls = self.upload() self.assertEqual(0, result.returncode, result.stdout + result.stderr) self.assertEqual([ 'UPLOAD copyto fixture:encrypted/daily/jobtracker-db-synthetic.sql.gz --immutable', 'UPLOAD copyto fixture:encrypted/daily/jobtracker-files-synthetic.tar.gz --immutable', 'UPLOAD copy fixture:encrypted/monthly/2099-01 --immutable', ], calls) self.assertIn('DEPLOY WOULD CONTINUE', result.stdout) self.assertIn('Current recovery set copied to configured off-host remote.', result.stdout) def test_daily_copy_failure_stops_before_remaining_files_and_monthly_copy(self): result, calls = self.upload('copyto') self.assertEqual(1, result.returncode, result.stdout + result.stderr) self.assertEqual([ 'UPLOAD copyto fixture:encrypted/daily/jobtracker-db-synthetic.sql.gz --immutable', ], calls) self.assertIn('Daily off-host backup upload failed', result.stdout) self.assertNotIn('DEPLOY WOULD CONTINUE', result.stdout) self.assertNotIn('Current recovery set copied', result.stdout) def test_monthly_copy_failure_blocks_deploy(self): result, calls = self.upload('copy') self.assertEqual(1, result.returncode, result.stdout + result.stderr) self.assertEqual(3, len(calls)) self.assertEqual('UPLOAD copy fixture:encrypted/monthly/2099-01 --immutable', calls[-1]) self.assertIn('Monthly off-host backup upload failed', result.stdout) self.assertNotIn('DEPLOY WOULD CONTINUE', result.stdout) self.assertNotIn('Current recovery set copied', result.stdout) class BackupRetentionTests(unittest.TestCase): def retain(self, fail_operation='', *, previous_snapshot=False): source = (ROOT / 'deploy/deploy.sh').read_text(encoding='utf-8') start = source.index('apply_backup_retention() {') function = source[start:source.index('\n}\n', start) + 3] gate_start = source.index('if ! apply_backup_retention || ! upload_current_backup_set; then') gate = source[gate_start:source.index('\nfi\n', gate_start) + 4] fixtures = {'jobtracker-db-synthetic.sql.gz': b'database fixture', 'jobtracker-files-synthetic.tar.gz': b'files fixture'} with tempfile.TemporaryDirectory() as temporary: folder = Path(temporary) for name, contents in fixtures.items(): (folder / name).write_bytes(contents) monthly = folder / 'monthly' / '2099-01' if previous_snapshot: monthly.mkdir(parents=True) (monthly / 'jobtracker-db-earlier.sql.gz').write_bytes(b'earlier fixture') (monthly / '.complete').write_bytes(b'earlier\n') script = '\n'.join([ 'set -euo pipefail', 'BACKUP_DIR=' + shlex.quote(folder.as_posix()), 'BACKUP_STAMP=synthetic', 'FAIL_OPERATION=' + shlex.quote(fail_operation), 'date() { echo 2099-01; }', 'upload_current_backup_set() { echo "UPLOAD WOULD START"; }', 'find() { if [[ "$*" == *-delete* ]]; then echo "PRUNING WOULD START"; fi; }', 'rm() { echo "CLEANUP REQUEST $1 $2"; }', 'cp() {', ' if [ "$FAIL_OPERATION" = cp ]; then', ' printf partial > "${@: -1}/partial"; return 23;', ' fi;', ' command cp "$@";', '}', 'mv() { if [ "$FAIL_OPERATION" = mv ]; then return 24; fi; command mv "$@"; }', function, gate, 'echo "DEPLOY WOULD CONTINUE"', ]) result = subprocess.run([BASH, '--noprofile', '--norc'], input=script, text=True, encoding='utf-8', capture_output=True, cwd=ROOT, timeout=10, env={key: value for key, value in os.environ.items() if key.upper() in {'PATH', 'SYSTEMROOT', 'WINDIR', 'TEMP', 'TMP'}}) artifacts = {file.name: file.read_bytes() for file in monthly.iterdir()} if monthly.exists() else {} return result, artifacts def test_completed_monthly_snapshot_contains_both_artifacts_and_stamp(self): result, artifacts = self.retain() self.assertEqual(0, result.returncode, result.stdout + result.stderr) self.assertEqual({'jobtracker-db-synthetic.sql.gz': b'database fixture', 'jobtracker-files-synthetic.tar.gz': b'files fixture', '.complete': b'synthetic\n'}, artifacts) self.assertIn('PRUNING WOULD START', result.stdout) self.assertIn('UPLOAD WOULD START', result.stdout) self.assertIn('DEPLOY WOULD CONTINUE', result.stdout) def test_partial_monthly_copy_cannot_mark_snapshot_complete_or_proceed(self): result, artifacts = self.retain('cp') self.assertEqual(1, result.returncode, result.stdout + result.stderr) self.assertEqual({'partial': b'partial'}, artifacts) self.assertIn('Could not retain monthly recovery files', result.stdout) self.assertNotIn('PRUNING WOULD START', result.stdout) self.assertNotIn('UPLOAD WOULD START', result.stdout) self.assertNotIn('DEPLOY WOULD CONTINUE', result.stdout) def test_marker_publish_failure_does_not_expose_complete_marker_or_proceed(self): result, artifacts = self.retain('mv') self.assertEqual(1, result.returncode, result.stdout + result.stderr) self.assertNotIn('.complete', artifacts) self.assertIn('CLEANUP REQUEST -f --', result.stdout) self.assertIn('Could not publish monthly recovery marker', result.stdout) self.assertNotIn('PRUNING WOULD START', result.stdout) self.assertNotIn('UPLOAD WOULD START', result.stdout) self.assertNotIn('DEPLOY WOULD CONTINUE', result.stdout) def test_preserves_the_existing_completed_monthly_snapshot(self): result, artifacts = self.retain('cp', previous_snapshot=True) self.assertEqual(0, result.returncode, result.stdout + result.stderr) self.assertEqual({'jobtracker-db-earlier.sql.gz': b'earlier fixture', '.complete': b'earlier\n'}, artifacts) self.assertIn('DEPLOY WOULD CONTINUE', result.stdout) if __name__ == '__main__': unittest.main()