"""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()