From abdf840a99d1ba383b08b487601e3355560c2620 Mon Sep 17 00:00:00 2001 From: smail Date: Mon, 21 Sep 2026 06:41:43 +0800 Subject: [PATCH] =?UTF-8?q?=E6=89=A9=E5=B1=95SQL=E8=B7=A8=E5=BC=95?= =?UTF-8?q?=E6=93=8E=E8=BF=81=E7=A7=BB=E9=AA=8C=E6=94=B6=E5=B9=B6=E4=BF=9D?= =?UTF-8?q?=E7=95=99=E6=A0=A1=E9=AA=8C=E5=A4=B1=E8=B4=A5=E5=8F=8D=E4=BE=8B?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- test/deployment/test_runtime_sql_migration.py | 54 +++++++++++++++++++ test/tools/runtime_review.py | 46 ++++++++++++---- test/tools/runtime_sql_migration.py | 28 ++++++++-- 3 files changed, 116 insertions(+), 12 deletions(-) diff --git a/test/deployment/test_runtime_sql_migration.py b/test/deployment/test_runtime_sql_migration.py index 1d2d989..5358ccb 100644 --- a/test/deployment/test_runtime_sql_migration.py +++ b/test/deployment/test_runtime_sql_migration.py @@ -1,5 +1,6 @@ """SQL 迁移验收器权限、摘要和边界测试，不连接真实数据库。""" +import hashlib import json from pathlib import Path import sys @@ -9,9 +10,62 @@ from unittest.mock import Mock, patch sys.path.insert(0, str(Path(__file__).resolve().parents[1] / 'tools')) import runtime_sql_migration as migration +import runtime_review as review class RuntimeSQLMigrationTest(unittest.TestCase): + def test_cross_engine_requires_explicit_sql_migration(self): + for source, target in (('mysql', 'postgres'), ('postgres', 'mysql')): + self.assertEqual(review.migration_kind(source, target, True, False), target) + self.assertEqual(review.migration_kind('sqlite', None, True, False), 'sqlite') + for arguments in (('sqlite', 'mysql', True, False), ('mysql', 'postgres', False, False), + ('mysql', 'postgres', True, True), ('mysql', 'sqlite', True, False)): + with self.assertRaises(ValueError): + review.migration_kind(*arguments) + + def test_all_database_bridges_closed_even_when_one_fails(self): + first, second = Mock(), Mock() + second.close.side_effect = RuntimeError('bridge failure') + result = {} + review.stop_database_bridges({3306: first, 5432: second}, result) + first.close.assert_called_once_with() + second.close.assert_called_once_with() + self.assertFalse(result['database_bridge_stopped']) + self.assertEqual(result['database_bridge_ports_stopped'], [3306]) + self.assertEqual(result['cleanup_errors'], [{'action': 'database_bridge', 'port': 5432, 'error': 'RuntimeError'}]) + + def test_both_database_bridges_reported_stopped(self): + result = {} + review.stop_database_bridges({3306: Mock(), 5432: Mock()}, result) + self.assertTrue(result['database_bridge_stopped']) + self.assertEqual(result['database_bridge_ports_stopped'], [3306, 5432]) + self.assertFalse(result['cleanup_errors']) + + def test_history_compares_complete_raw_bytes_across_engines(self): + payload = b'{"text":"\\u4e2d\\u6587","zero":"\\u0000"}' + digest = hashlib.sha256(payload).hexdigest().encode() + row = [value.hex() for value in (b'backup', b'identifier', payload, digest)] + records = [] + for kind in ('mysql', 'postgres'): + database = migration.SQLFixture(kind) + with patch.object(database, 'query', return_value=json.dumps(row)) as query: + records.append(migration.history_rows(database)) + for column in ('kind', 'id', 'payload', 'digest'): + self.assertIn(column, query.call_args.args[0]) + self.assertEqual(records[0], records[1]) + + def test_invalid_history_fails_instead_of_skipping(self): + database = migration.SQLFixture('mysql') + valid = [value.hex() for value in (b'backup', b'id', b'{}', hashlib.sha256(b'{}').hexdigest().encode())] + invalid = [[], [None] * 4, ['xyz'] * 4, valid[:2] + [b'corrupt'.hex(), valid[3]]] + for row in invalid: + with patch.object(database, 'query', return_value=json.dumps(row)): + with self.assertRaises(AssertionError): + migration.history_rows(database) + with patch.object(database, 'query', return_value=json.dumps(valid) + '\n' + json.dumps(valid)): + with self.assertRaises(AssertionError): + migration.history_rows(database) + def test_target_creation_does_not_precreate_database(self): for kind in ('mysql', 'postgres'): target = migration.MigrationTarget(kind) diff --git a/test/tools/runtime_review.py b/test/tools/runtime_review.py index f490560..037cf6d 100644 --- a/test/tools/runtime_review.py +++ b/test/tools/runtime_review.py @@ -72,6 +72,27 @@ def collect_and_stop(result, unit, probe_unit, proxy_unit=None): result['cleanup_errors'] = stop_test_units(names) +def migration_kind(source, target, enabled, interrupted): + if target is None: + return source + if not enabled or interrupted or source == 'sqlite' or target not in ('mysql', 'postgres'): + raise ValueError('跨引擎目标必须用于独立 SQL 迁移验收') + return target + + +def stop_database_bridges(bridges, result): + errors = result.setdefault('cleanup_errors', []) + closed = [] + for port, bridge in reversed(list(bridges.items())): + try: + bridge.close() + closed.append(port) + except Exception as error: + errors.append({'action': 'database_bridge', 'port': port, 'error': type(error).__name__}) + result['database_bridge_ports_stopped'] = sorted(closed) + result['database_bridge_stopped'] = bool(bridges) and len(closed) == len(bridges) + + def main(): parser = argparse.ArgumentParser() parser.add_argument('--main-sha', required=True) @@ -82,7 +103,12 @@ def main(): parser.add_argument('--migration', action='store_true') parser.add_argument('--interruption', action='store_true') parser.add_argument('--database-type', choices=('sqlite', 'mysql', 'postgres'), default='sqlite') + parser.add_argument('--migration-target-type', choices=('mysql', 'postgres')) args = parser.parse_args() + try: + target_kind = migration_kind(args.database_type, args.migration_target_type, args.migration, args.interruption) + except ValueError as error: + parser.error(str(error)) if args.interruption and (not args.migration or args.database_type != 'sqlite'): parser.error('中断验收必须选择 SQLite 迁移独立场景') if args.migration and (args.nginx_template or args.restore): @@ -165,7 +191,7 @@ def main(): sql_fixture = None migration_target = None network_unit = None - bridge = None + bridges = {} result['database_type'] = args.database_type if args.database_type != 'sqlite': from runtime_sql import SQLFixture @@ -173,8 +199,9 @@ def main(): result['sql_database'] = sql_fixture.settings['name'] if args.migration: from runtime_sql_migration import MigrationTarget - migration_target = MigrationTarget(args.database_type) + migration_target = MigrationTarget(target_kind) result['sql_target_database'] = migration_target.settings['name'] + result['migration_target_type'] = target_kind if args.nginx_template: from runtime_proxy import prepare_nginx prepare_nginx(target, root, args.nginx_template) @@ -230,7 +257,12 @@ def main(): assert (root / 'smail').stat().st_ino != (live / 'smail').stat().st_ino if sql_fixture: from runtime_sql_bridge import DatabaseBridge - bridge = DatabaseBridge(int(pid), sql_fixture.port) + ports = {sql_fixture.port} + if migration_target: + ports.add(migration_target.port) + for port in sorted(ports): + bridges[port] = DatabaseBridge(int(pid), port) + result['database_bridge_ports'] = sorted(bridges) if args.nginx_template: subprocess.run([ 'systemd-run', '--unit=' + proxy_unit, '--property=Type=exec', @@ -304,12 +336,8 @@ def main(): result['integrity'] = {'error': type(error).__name__} finally: result.setdefault('cleanup_errors', []) - if bridge: - try: - bridge.close() - result['database_bridge_stopped'] = True - except Exception as error: - result['cleanup_errors'].append({'action': 'database_bridge', 'error': type(error).__name__}) + if bridges: + stop_database_bridges(bridges, result) if sql_fixture: try: sql_fixture.lock() diff --git a/test/tools/runtime_sql_migration.py b/test/tools/runtime_sql_migration.py index cf1c1c5..60f1c7d 100644 --- a/test/tools/runtime_sql_migration.py +++ b/test/tools/runtime_sql_migration.py @@ -99,6 +99,27 @@ def fingerprint(database): return hashlib.sha256(json.dumps(content, sort_keys=True).encode()).hexdigest() +def history_rows(database): + # 两个引擎统一比较四列原始字节，不能直接比较各自的 BLOB JSON 表示。 + if database.kind == 'mysql': + values = ','.join('LOWER(HEX(`' + column + '`))' for column in ('kind', 'id', 'payload', 'digest')) + query = 'SELECT JSON_ARRAY(' + values + ') FROM config_records' + else: + values = ["encode(convert_to(" + column + ",'UTF8'),'hex')" for column in ('kind', 'id')] + values += ["encode(payload,'hex')", "encode(convert_to(digest,'UTF8'),'hex')"] + query = 'SELECT json_build_array(' + ','.join(values) + ') FROM config_records' + rows = [] + for line in database.query(query).splitlines(): + values = json.loads(line) + assert isinstance(values, list) and len(values) == 4 + assert all(isinstance(value, str) and re.fullmatch(r'(?:[0-9a-f]{2})+', value) for value in values) + payload, digest = bytes.fromhex(values[2]), bytes.fromhex(values[3]).decode('ascii') + assert hashlib.sha256(payload).hexdigest() == digest, '配置历史载荷与摘要不一致' + rows.append(json.dumps(values, separators=(',', ':'))) + assert len(rows) == len(set(rows)), '配置历史出现重复完整行' + return sorted(rows) + + def assert_connections(source, target): counts = [] for database in (source, target): @@ -166,6 +187,7 @@ 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, + 'target_type': target.kind, 'cross_engine': source.kind != target.kind, 'source_database': source.settings['name'], 'target_database': target.settings['name']} settings = {'source': source.settings, 'target': target.settings} before = service_state(unit) @@ -177,7 +199,7 @@ def verify_migration(root, unit, password, result, source, target): details['internal_restart_same_process'] = True details['connections_after_internal'] = assert_connections(source, target) source_digest = fingerprint(source) - history = table_rows(source, 'config_records') + history = history_rows(source) assert history details['internal'] = run_probe(root, unit, password, 'internal', settings) subprocess.run(['systemctl', 'restart', unit], check=True, timeout=55) @@ -209,7 +231,7 @@ def verify_migration(root, unit, password, result, source, target): 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') + current = history_rows(target) 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) @@ -227,7 +249,7 @@ def main(): 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'] + assert 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