扩展SQL跨引擎迁移验收并保留校验失败反例

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)

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

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