137 lines
6.6 KiB
Python
137 lines
6.6 KiB
Python
"""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()
|