"""SQL 迁移验收器权限、摘要和边界测试，不连接真实数据库。"""

import json
from pathlib import Path
import sys
import tempfile
import unittest
from unittest.mock import Mock, patch

sys.path.insert(0, str(Path(__file__).resolve().parents[1] / 'tools'))
import runtime_sql_migration as migration


class RuntimeSQLMigrationTest(unittest.TestCase):
    def test_target_creation_does_not_precreate_database(self):
        for kind in ('mysql', 'postgres'):
            target = migration.MigrationTarget(kind)
            with patch.object(target, 'admin', return_value='0') as admin:
                target.create()
            statements = '\n'.join(call.args[0] for call in admin.call_args_list)
            self.assertNotIn('CREATE DATABASE ', statements)
            self.assertNotIn('DROP ', statements)
            self.assertNotIn('*.*', statements)
            self.assertNotIn('SUPERUSER', statements)
            if kind == 'mysql':
                self.assertIn(target.settings['name'].replace('_', r'\_'), statements)
            else:
                self.assertIn(' LOGIN CREATEDB CONNECTION LIMIT 24', statements)

    def test_existing_target_is_rejected(self):
        target = migration.MigrationTarget('mysql')
        with patch.object(target, 'admin', return_value='1'):
            with self.assertRaises(AssertionError):
                target.create()

    def test_cleanup_revokes_createdb_and_locks_login(self):
        target = migration.MigrationTarget('postgres')
        target.account_created = True
        with patch.object(target, 'admin', side_effect=['', 'f', '', 'f']) as admin:
            target.lock()
        statements = '\n'.join(call.args[0] for call in admin.call_args_list)
        self.assertIn(' NOCREATEDB;', statements)
        self.assertIn(' NOLOGIN;', statements)

    def test_revoke_failure_still_attempts_login_lock(self):
        target = migration.MigrationTarget('postgres')
        target.account_created = True
        with patch.object(target, 'admin', side_effect=[RuntimeError('denied'), '', 'f']) as admin:
            with self.assertRaises(RuntimeError):
                target.lock()
        self.assertIn(' NOLOGIN;', admin.call_args_list[1].args[0])

    def test_unlock_requires_confirmation(self):
        for kind in ('mysql', 'postgres'):
            target = migration.MigrationTarget(kind)
            target.account_created = True
            with patch.object(target, 'admin', return_value=''):
                with self.assertRaises(AssertionError):
                    target.unlock()

    def test_uncreated_account_is_not_changed(self):
        target = migration.MigrationTarget('postgres')
        with patch.object(target, 'admin') as admin:
            target.lock()
            admin.assert_not_called()
            with self.assertRaises(AssertionError):
                target.unlock()

    def test_target_uses_structured_secret(self):
        target = migration.MigrationTarget('postgres')
        payload = migration.target_payload(target)
        self.assertEqual(payload['type'], 'postgresql')
        self.assertEqual(payload['password'], {'plaintext': target.settings['password']})
        self.assertNotIn('dsn', payload)

    def test_untrusted_identifiers_rejected_before_query(self):
        target = migration.MigrationTarget('mysql')
        with patch.object(target, 'query') as query:
            for value in ('user; DROP TABLE email', 'email`', 'a.b', 'x y'):
                with self.assertRaises(AssertionError):
                    migration.table_rows(target, value)
            query.assert_not_called()

    def test_mysql_digest_encoding_keeps_binary_and_null(self):
        target = migration.MigrationTarget('mysql')
        with patch.object(target, 'query', side_effect=['payload\ndigest', '[null,"01"]\n["00FF","02"]']) as query:
            rows = migration.table_rows(target, 'config_records')
        self.assertEqual(len(rows), 2)
        self.assertIn('JSON_ARRAY(HEX(`payload`),HEX(`digest`))', query.call_args.args[0])
        self.assertIn('[null, "01"]', rows)

    def test_fingerprint_stable_when_row_order_changes(self):
        target = migration.MigrationTarget('postgres')
        def read(sql):
            if 'information_schema.tables' in sql:
                return 'config_records\nemail'
            return '{"a":2}\n{"a":1}'
        with patch.object(target, 'query', side_effect=read):
            first = migration.fingerprint(target)
        with patch.object(target, 'query', side_effect=lambda sql: read(sql) if 'information_schema' in sql else '{"a":1}\n{"a":2}'):
            self.assertEqual(first, migration.fingerprint(target))
        with patch.object(target, 'query', side_effect=lambda sql: read(sql) if 'information_schema' in sql else '{"a":3}'):
            self.assertNotEqual(first, migration.fingerprint(target))

    def test_connections_require_source_zero_and_target_positive(self):
        source, target = migration.SQLFixture('mysql'), migration.MigrationTarget('mysql')
        with patch.object(source, 'admin', return_value='0'), patch.object(target, 'admin', return_value='2'):
            self.assertEqual(migration.assert_connections(source, target)['source_connections'], 0)
        with patch.object(source, 'admin', return_value='1'), patch.object(target, 'admin', return_value='2'):
            with self.assertRaises(AssertionError):
                migration.assert_connections(source, target)
        with patch.object(source, 'admin', return_value='0'), patch.object(target, 'admin', return_value='0'):
            with self.assertRaises(AssertionError):
                migration.assert_connections(source, target)

    def test_active_configuration_and_no_hidden_sqlite(self):
        target = migration.MigrationTarget('mysql')
        with tempfile.TemporaryDirectory() as directory:
            root = Path(directory)
            (root / 'config').mkdir()
            (root / 'data').mkdir()
            (root / 'config/config.json').write_text(json.dumps({'database': migration.target_payload(target)}))
            migration.assert_active(root, target)
            (root / 'data/state.sqlite3').touch()
            with self.assertRaises(AssertionError):
                migration.assert_active(root, target)

    def test_formal_directory_rejected_before_any_service_operation(self):
        with patch.object(migration, 'service_state') as state:
            with self.assertRaises(AssertionError):
                migration.verify_migration(Path('/opt/smail'), 'smail.service', 'not-a-secret', {}, Mock(), Mock())
            state.assert_not_called()


if __name__ == '__main__':
    unittest.main()