From 48447620647d19167fd3466393501e9e9efa6ab8 Mon Sep 17 00:00:00 2001 From: smail Date: Mon, 21 Sep 2026 05:55:28 +0800 Subject: [PATCH] =?UTF-8?q?=E8=A1=A5=E9=BD=90=E5=8F=8CSQL=E5=8D=95?= =?UTF-8?q?=E6=B4=BB=E5=8A=A8=E5=BA=93=E5=88=87=E6=8D=A2=E5=8F=8A=E7=9C=9F?= =?UTF-8?q?=E5=AE=9E=E9=87=8D=E5=90=AF=E6=81=A2=E5=A4=8D=E9=AA=8C=E6=94=B6?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- test/deployment/test_runtime_sql_migration.py | 136 ++++++++++ test/tools/runtime_migration.py | 8 +- test/tools/runtime_review.py | 64 ++++- test/tools/runtime_sql_migration.py | 240 ++++++++++++++++++ 4 files changed, 437 insertions(+), 11 deletions(-) create mode 100644 test/deployment/test_runtime_sql_migration.py create mode 100644 test/tools/runtime_sql_migration.py diff --git a/test/deployment/test_runtime_sql_migration.py b/test/deployment/test_runtime_sql_migration.py new file mode 100644 index 0000000..1d2d989 --- /dev/null +++ b/test/deployment/test_runtime_sql_migration.py @@ -0,0 +1,136 @@ +"""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() diff --git a/test/tools/runtime_migration.py b/test/tools/runtime_migration.py index e0dcdea..ccde247 100644 --- a/test/tools/runtime_migration.py +++ b/test/tools/runtime_migration.py @@ -194,12 +194,13 @@ def process_uses_target(root, state): assert '/opt/smail/data/smail.db' not in paths, '进程仍然打开旧业务库' -def run_probe(root, unit, password, phase): +def run_probe(root, unit, password, phase, sql_settings=None): state = service_state(unit) pid = state['MainPID'] assert pid.isdigit() and int(pid) > 0 assert Path('/proc/' + pid + '/ns/net').readlink() != Path('/proc/self/ns/net').readlink() probe = 'smail-migration-probe-' + unit.removeprefix('smail-runtime-review-') + '-' + phase + script = 'runtime_sql_migration.py' if sql_settings else 'runtime_migration.py' command = ['systemd-run', '--wait', '--pipe', '--unit=' + probe, '--property=User=smail', '--property=Group=smail', '--property=NetworkNamespacePath=/proc/' + pid + '/ns/net', @@ -208,10 +209,11 @@ def run_probe(root, unit, password, phase): '--property=ReadWritePaths=' + str(root.parent), '--property=MemoryMax=48M', '--property=MemorySwapMax=0', '--property=CPUQuota=25%', '--property=TasksMax=32', '--property=RuntimeMaxSec=150s', '--property=KillMode=control-group', - '--property=TimeoutStopSec=10s', 'python3', str(root.parent / 'runtime_migration.py'), + '--property=TimeoutStopSec=10s', 'python3', str(root.parent / script), str(root), phase] try: - completed = subprocess.run(command, input=password + '\n', text=True, + data = password + '\n' + (json.dumps(sql_settings) + '\n' if sql_settings else '') + completed = subprocess.run(command, input=data, text=True, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, timeout=175) (root.parent / ('migration-' + phase + '.log')).write_text(completed.stdout, encoding='utf-8') assert completed.returncode == 0, '迁移探针失败，请检查独立阶段日志' diff --git a/test/tools/runtime_review.py b/test/tools/runtime_review.py index 61123e6..58055cf 100644 --- a/test/tools/runtime_review.py +++ b/test/tools/runtime_review.py @@ -37,7 +37,7 @@ def database_integrity(root): def stop_test_units(names): - assert all(re.fullmatch(r'smail-runtime-(review|probe|nginx)-[0-9]{8}-[0-9]{6}', name) + assert all(re.fullmatch(r'smail-runtime-(review|probe|nginx|netns)-[0-9]{8}-[0-9]{6}', name) for name in names), '只能停止本轮隔离测试单元' errors = [] for name in names: @@ -82,8 +82,8 @@ def main(): parser.add_argument('--migration', action='store_true') parser.add_argument('--database-type', choices=('sqlite', 'mysql', 'postgres'), default='sqlite') args = parser.parse_args() - if args.migration and (args.database_type != 'sqlite' or args.nginx_template or args.restore): - parser.error('迁移进程验收当前只覆盖直接 TLS 的 SQLite 独立场景') + if args.migration and (args.nginx_template or args.restore): + parser.error('迁移进程验收仅使用直接 TLS 的独立场景') if args.restore and args.database_type != 'sqlite': parser.error('SQL 数据库备份恢复需要独立验收，不能套用 SQLite 冷备流程') assert os.geteuid() == 0 and run('hostname') == 'ikun2' @@ -127,6 +127,10 @@ def main(): shutil.copy2(Path(__file__).with_name('runtime_sql.py'), target / 'runtime_sql.py') (target / 'runtime_sql.py').chmod(0o640) shutil.chown(target / 'runtime_sql.py', user='root', group='smail') + if args.migration: + shutil.copy2(Path(__file__).with_name('runtime_sql_migration.py'), target / 'runtime_sql_migration.py') + (target / 'runtime_sql_migration.py').chmod(0o640) + shutil.chown(target / 'runtime_sql_migration.py', user='root', group='smail') if args.nginx_template: shutil.copy2(Path(__file__).with_name('runtime_proxy.py'), target / 'runtime_proxy.py') (target / 'runtime_proxy.py').chmod(0o640) @@ -152,12 +156,18 @@ def main(): service_env = [] probe_args = [] sql_fixture = None + migration_target = None + network_unit = None bridge = None result['database_type'] = args.database_type if args.database_type != 'sqlite': from runtime_sql import SQLFixture sql_fixture = SQLFixture(args.database_type) result['sql_database'] = sql_fixture.settings['name'] + if args.migration: + from runtime_sql_migration import MigrationTarget + migration_target = MigrationTarget(args.database_type) + result['sql_target_database'] = migration_target.settings['name'] if args.nginx_template: from runtime_proxy import prepare_nginx prepare_nginx(target, root, args.nginx_template) @@ -171,16 +181,34 @@ def main(): if sql_fixture: sql_fixture.create() probe_args.append('--sql-stdin') + network_args = ['--property=PrivateNetwork=yes'] + if migration_target: + migration_target.create() + # 使用独立限时单元固定隔离网络，应用换进程后仍能访问本轮数据库桥接。 + network_unit = 'smail-runtime-netns-' + stamp + subprocess.run([ + 'systemd-run', '--unit=' + network_unit, '--property=Type=exec', + '--property=User=smail', '--property=Group=smail', '--property=PrivateNetwork=yes', + '--property=ProtectSystem=strict', '--property=ProtectHome=yes', + '--property=NoNewPrivileges=yes', '--property=InaccessiblePaths=/opt/smail /run/smail', + '--property=MemoryMax=8M', '--property=MemorySwapMax=0', '--property=TasksMax=8', + '--property=RuntimeMaxSec=900s', '/usr/bin/sleep', '900', + ], check=True) + network_pid = run('systemctl', 'show', network_unit, '--property=MainPID', '--value') + assert network_pid.isdigit() and int(network_pid) > 0 + assert Path('/proc/' + network_pid + '/ns/net').readlink() != Path('/proc/self/ns/net').readlink() + network_args = ['--property=NetworkNamespacePath=/proc/' + network_pid + '/ns/net'] subprocess.run([ 'systemd-run', '--unit=' + unit, '--property=Type=exec', '--property=User=smail', '--property=Group=smail', - '--property=PrivateNetwork=yes', '--property=PrivateTmp=yes', + *network_args, '--property=PrivateTmp=yes', '--property=ProtectSystem=strict', '--property=ProtectHome=yes', '--property=NoNewPrivileges=yes', '--property=CapabilityBoundingSet=CAP_NET_BIND_SERVICE', '--property=AmbientCapabilities=CAP_NET_BIND_SERVICE', '--property=MemoryMax=' + str(args.memory_max_mib) + 'M', '--property=MemorySwapMax=0', '--property=CPUQuota=25%', - '--property=TasksMax=128', '--property=RuntimeMaxSec=300s', '--property=OOMPolicy=stop', + '--property=TasksMax=128', '--property=RuntimeMaxSec=' + ('600s' if args.migration else '300s'), + '--property=OOMPolicy=stop', '--property=KillMode=control-group', '--property=TimeoutStopSec=10s', '--property=BindPaths=' + str(root) + ':/opt/smail ' + str(control) + ':/run/smail', '--property=ReadWritePaths=/opt/smail /run/smail', @@ -239,8 +267,12 @@ def main(): assert database.execute('SELECT COUNT(*) FROM email WHERE msg_id=?', (message_id,)).fetchone()[0] == 0 result['live_message_absent'] = True if args.migration: - from runtime_migration import verify_migration - verify_migration(root, unit, password, result) + if migration_target: + from runtime_sql_migration import verify_migration + verify_migration(root, unit, password, result, sql_fixture, migration_target) + else: + from runtime_migration import verify_migration + verify_migration(root, unit, password, result) finally: try: collect_and_stop(result, unit, probe_unit, proxy_unit if args.nginx_template else None) @@ -249,7 +281,11 @@ def main(): result['live_unchanged'] = (result['after'] == before and hashlib.file_digest(stream, 'sha256').hexdigest() == live_sha) try: - if args.migration: + if migration_target: + from runtime_sql_migration import assert_active + assert_active(root, migration_target) + result['integrity'] = migration_target.integrity(root) + elif args.migration: from runtime_migration import migration_integrity result['integrity'] = migration_integrity(root) else: @@ -270,6 +306,16 @@ def main(): result['test_database_account_locked'] = True except Exception as error: result['cleanup_errors'].append({'action': 'database_account', 'error': type(error).__name__}) + if migration_target: + try: + migration_target.lock() + result['migration_target_account_locked'] = True + except Exception as error: + result['cleanup_errors'].append({'action': 'migration_target_account', 'error': type(error).__name__}) + if network_unit: + errors = stop_test_units([network_unit]) + result['cleanup_errors'].extend(errors) + result['network_anchor_stopped'] = not errors (target / 'service.log').write_text(run('journalctl', '-u', unit, '--no-pager'), encoding='utf-8') kernel = run('journalctl', '-k', '--since', started, '--no-pager') result['oom_events'] = sum('oom-kill:constraint=' in line for line in kernel.splitlines()) @@ -280,6 +326,8 @@ def main(): assert not result['resource_errors'], '资源证据不完整，需要核查保留的结果' if sql_fixture: assert result['integrity'].get('sqlite_files') == 0 and result['integrity']['config_records'] > 0 + if args.migration: + assert result['migration']['passed'] and result['network_anchor_stopped'] elif args.migration: assert result['migration']['passed'] and result['integrity']['active_database'] == 'migrated.db' else: diff --git a/test/tools/runtime_sql_migration.py b/test/tools/runtime_sql_migration.py new file mode 100644 index 0000000..cf1c1c5 --- /dev/null +++ b/test/tools/runtime_sql_migration.py @@ -0,0 +1,240 @@ +"""SQL 切库的真实重启与账号不可用恢复验收，只操作随机隔离库。""" + +import argparse +from email.message import EmailMessage +import hashlib +import json +import os +from pathlib import Path +import re +import subprocess +import sys +import time + +from runtime_migration import clear_pending_restart, login, run_probe, service_state, wait_ready +from runtime_probe import ACCOUNT, ADDRESS, protocol_operation, smtp_connect, verify_protocol_read +from runtime_sql import SQLDatabase, SQLFixture + + +def identifier(value): + assert re.fullmatch(r'[A-Za-z_][A-Za-z0-9_]*', value), '验收数据表标识不符合安全范围' + return value + + +class MigrationTarget(SQLFixture): + def create(self): + name, user, password = (self.settings[key] for key in ('name', 'user', 'password')) + if self.kind == 'mysql': + self.admin("CREATE USER '" + user + "'@'localhost' IDENTIFIED BY '" + password + "';") + self.account_created = True + self.admin('GRANT ALL ON `' + name.replace('_', r'\_') + "`.* TO '" + user + "'@'localhost';") + count = self.admin("SELECT COUNT(*) FROM information_schema.schemata WHERE schema_name='" + name + "';") + else: + # PostgreSQL 创建目标库需要临时建库权限；复制结束立即撤销，绝不授予超级用户。 + self.admin('CREATE ROLE ' + user + " LOGIN CREATEDB CONNECTION LIMIT 24 PASSWORD '" + password + "';") + self.account_created = True + count = self.admin("SELECT COUNT(*) FROM pg_database WHERE datname='" + name + "';") + assert count == '0', '迁移目标必须不存在，不能复用已有库' + + def revoke_creation(self): + if self.account_created and self.kind == 'postgres': + user = self.settings['user'] + self.admin('ALTER ROLE ' + user + ' NOCREATEDB;') + assert self.admin("SELECT rolcreatedb FROM pg_roles WHERE rolname='" + user + "';") == 'f' + + def lock(self): + try: + self.revoke_creation() + finally: + super().lock() + + def unlock(self): + assert self.account_created + user = self.settings['user'] + if self.kind == 'mysql': + self.admin("ALTER USER '" + user + "'@'localhost' ACCOUNT UNLOCK;") + assert self.admin("SELECT account_locked FROM mysql.user WHERE User='" + user + + "' AND Host='localhost';") == 'N' + else: + self.admin('ALTER ROLE ' + user + ' LOGIN;') + assert self.admin("SELECT rolcanlogin FROM pg_roles WHERE rolname='" + user + "';") == 't' + + +def target_payload(database): + settings = database.settings + return {'type': 'mysql' if database.kind == 'mysql' else 'postgresql', + 'host': '127.0.0.1', 'port': database.port, 'username': settings['user'], + 'name': settings['name'], 'password': {'plaintext': settings['password']}} + + +def assert_active(root, target): + database = json.loads((root / 'config/config.json').read_text())['database'] + expected = target_payload(target) + for field in ('type', 'host', 'port', 'username', 'name'): + assert database[field] == expected[field], '活动配置没有指向本轮唯一 SQL 目标' + assert not list((root / 'data').rglob('*.db')) + assert not list((root / 'data').rglob('*.sqlite*')) + + +def table_rows(database, table): + table = identifier(table) + if database.kind == 'mysql': + columns = database.query("SELECT COLUMN_NAME FROM information_schema.columns WHERE " + "table_schema=DATABASE() AND table_name='" + table + "' ORDER BY ORDINAL_POSITION") + # 对二进制和文本统一取十六进制，保留 NULL，避免 BLOB 不能进入 JSON 的问题。 + values = ','.join('HEX(`' + identifier(column) + '`)' for column in columns.splitlines()) + assert values + output = database.query('SELECT JSON_ARRAY(' + values + ') FROM `' + table + '`') + else: + output = database.query('SELECT row_to_json(t) FROM "' + table + '" t') + return sorted(json.dumps(json.loads(line), sort_keys=True, ensure_ascii=True) for line in output.splitlines()) + + +def fingerprint(database): + schema = 'DATABASE()' if database.kind == 'mysql' else "'public'" + tables = database.query('SELECT table_name FROM information_schema.tables WHERE table_schema=' + + schema + " AND table_type='BASE TABLE' ORDER BY table_name").splitlines() + assert 'config_records' in tables and 'email' in tables + content = {table: table_rows(database, table) for table in tables} + return hashlib.sha256(json.dumps(content, sort_keys=True).encode()).hexdigest() + + +def assert_connections(source, target): + counts = [] + for database in (source, target): + name, user = database.settings['name'], database.settings['user'] + if database.kind == 'mysql': + query = "SELECT COUNT(*) FROM information_schema.processlist WHERE USER='" + user + "' AND DB='" + name + "';" + else: + query = "SELECT COUNT(*) FROM pg_stat_activity WHERE usename='" + user + "' AND datname='" + name + "';" + counts.append(int(database.admin(query))) + assert counts[0] == 0 and counts[1] > 0, '源库仍有活动连接或目标库没有应用连接' + return {'source_connections': counts[0], 'target_connections': counts[1]} + + +def prepare(root, password, source, target): + client, _ = login(root, password) + generation = clear_pending_restart(client) + endpoint = '/api/v1/database/migrations' + preview = client.success(endpoint + '/preview', { + 'expectedGeneration': generation, 'target': target_payload(target), + }) + record = client.success(endpoint, {'id': preview['migration']['id'], 'token': preview['token']}) + assert record['status'] == 'succeeded' + assert_active(root, target) + pending = client.success('/api/v1/config/status', method='GET') + assert pending['activeGeneration'] > generation and pending['pendingRestart'] + status, denied, _ = client.request('/api/v1/security/credentials/protocols/generate', {'protocol': 'imap'}) + assert status == 503 and denied is not None and denied['data'] == 'DATABASE_RESTART_REQUIRED' + assert client.success('/api/v1/system/restart')['status'] == 'accepted' + wait_ready(client, pending['activeGeneration']) + assert client.success('/api/v1/user/info')['account'] == ACCOUNT + return {'switch_succeeded': True, 'pending_write_denied': True, 'internal_restart_completed': True, + 'existing_session_preserved': True, 'generation': pending['activeGeneration']} + + +def verify_mail(root, password, phase, source, target): + client, context = login(root, password) + assert_active(root, target) + assert not client.success('/api/v1/config/status', method='GET')['pendingRestart'] + codes = {} + for protocol in ('smtp', 'imap', 'pop3'): + codes[protocol] = client.success('/api/v1/security/credentials/protocols/generate', {'protocol': protocol})['code'] + assert protocol_operation(protocol, context, codes[protocol])() == 'success' + original = json.loads((root.parent / 'probe-result.json').read_text())['smtp']['message_id'] + body = '内部25端口真实收信，检查中文正文和附件。' + attachment = '隔离验收附件内容'.encode('utf-8') + historical = verify_protocol_read(context, codes, original, body, attachment) + message_id = original + '.' + phase + assert re.fullmatch(r'[a-zA-Z0-9@._-]+', message_id) + message = EmailMessage() + message['From'], message['To'] = 'migration@sender.invalid', ADDRESS + message['Subject'] = 'SQL 切库重启后的中文邮件验收' + message['Message-ID'] = '<' + message_id + '>' + message.set_content(body) + message.add_attachment(attachment, maintype='text', subtype='plain', filename='验收附件.txt') + with smtp_connect(context) as smtp: + assert not smtp.send_message(message) + received = verify_protocol_read(context, codes, message_id, body, attachment) + target.verify_message(message_id, ACCOUNT, body, attachment) + assert source.query("SELECT COUNT(*) FROM email WHERE msg_id='" + message_id + "'") == '0' + return {'login': True, 'historical_mail': historical, 'new_mail': received, + 'new_mail_only_in_target': True, 'message_id': message_id} + + +def verify_migration(root, unit, password, result, source, target): + assert root.parent.parent == Path('/var/tmp') and re.fullmatch(r'smail-review-runtime-\d{8}-\d{6}', root.parent.name) + assert root.name == 'instance' and not root.is_symlink() and root.resolve(strict=True) == root + details = result['migration'] = {'passed': False, 'database_type': source.kind, + 'source_database': source.settings['name'], 'target_database': target.settings['name']} + settings = {'source': source.settings, 'target': target.settings} + before = service_state(unit) + details['prepare'] = run_probe(root, unit, password, 'prepare', settings) + target.revoke_creation() + details['temporary_creation_privilege_revoked'] = True + internal = service_state(unit) + assert before['MainPID'] == internal['MainPID'] and before['InvocationID'] == internal['InvocationID'] + details['internal_restart_same_process'] = True + details['connections_after_internal'] = assert_connections(source, target) + source_digest = fingerprint(source) + history = table_rows(source, 'config_records') + assert history + details['internal'] = run_probe(root, unit, password, 'internal', settings) + subprocess.run(['systemctl', 'restart', unit], check=True, timeout=55) + restarted = service_state(unit) + assert restarted['MainPID'] != internal['MainPID'] and restarted['InvocationID'] != internal['InvocationID'] + details['process'] = run_probe(root, unit, password, 'process', settings) + details['connections_after_process'] = assert_connections(source, target) + details['process_states'] = {'before': before, 'internal': internal, 'restarted': restarted} + try: + # 只锁定本轮目标账号，不停止共享数据库，不撤销任何正式账号权限。 + target.lock() + subprocess.run(['systemctl', 'restart', unit], check=True, timeout=55) + for _ in range(70): + failed = service_state(unit) + if failed['MainPID'] == '0' and failed['ActiveState'] == 'failed': + break + time.sleep(1) + else: + raise AssertionError('目标账号不可用时应用没有明确停止') + assert_active(root, target) + assert fingerprint(source) == source_digest + details['unavailable_target'] = {'failed_closed': True, 'source_unchanged': True, + 'configuration_preserved': True, 'state': failed} + finally: + target.unlock() + subprocess.run(['systemctl', 'start', unit], check=True, timeout=55) + details['recovered'] = run_probe(root, unit, password, 'recovered', settings) + final = service_state(unit) + assert final['MainPID'] not in (before['MainPID'], restarted['MainPID']) + details['connections_after_recovery'] = assert_connections(source, target) + assert fingerprint(source) == source_digest + current = table_rows(target, 'config_records') + assert set(history).issubset(current) + details['history'] = {'source_records': len(history), 'target_records': len(current), 'all_source_records_equal': True} + details.update(source_unchanged=True, target_account_restored=True, final_state=final, passed=True) + (root.parent / 'migration-result.json').write_text(json.dumps(details, ensure_ascii=False, indent=2), encoding='utf-8') + + +def main(): + parser = argparse.ArgumentParser() + parser.add_argument('root', type=Path) + parser.add_argument('phase', choices=('prepare', 'internal', 'process', 'recovered')) + args = parser.parse_args() + assert os.geteuid() != 0 + password = sys.stdin.readline().strip() + assert len(password) == 12 + settings = json.loads(sys.stdin.readline()) + assert set(settings) == {'source', 'target'} + source, target = (SQLDatabase(settings[key]) for key in ('source', 'target')) + assert source.kind == target.kind and source.settings['name'] != target.settings['name'] + result = (prepare(args.root, password, source, target) if args.phase == 'prepare' else + verify_mail(args.root, password, args.phase, source, target)) + result['passed'] = True + (args.root.parent / ('migration-' + args.phase + '.json')).write_text( + json.dumps(result, ensure_ascii=False, indent=2), encoding='utf-8') + print(json.dumps(result, ensure_ascii=False, indent=2)) + + +if __name__ == '__main__': + main()