cesnimda
90 lines
4.9 KiB
Python
90 lines
4.9 KiB
Python
"""Exercise the real backup function with a synthetic dump command; 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)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|