修复：保障菜单补丁原子执行

2026-07-27 19:18:23 +08:00
parent b6f3b1e1f0
commit 9a32f1a897
2 changed files with 289 additions and 9 deletions
@@ -1241,6 +1241,154 @@ class SqlContractTest(unittest.TestCase):
            with self.subTest(mutation=name):
                self.assertFalse(_mysql_menu_seed_has_safe_sql_mode_order(mutation))

+    def test_patch_preflights_new_menu_ids_before_atomic_permanent_dml(self) -> None:
+        raw = PATCH_PATH.read_text(encoding='utf-8')
+        statements = _sql_statements(raw)
+        normalized = [_normalized_sql_statement(statement) for statement in statements]
+
+        expected_table = 'pallet_workflow_menu_seed_expected'
+        conflict_guard = 'pallet_workflow_menu_seed_guard'
+
+        def statement_indexes(pattern: str) -> list[int]:
+            return [index for index, statement in enumerate(normalized) if re.fullmatch(pattern, statement, flags=re.S)]
+
+        expected_initial_drops = statement_indexes(rf'drop temporary table if exists {expected_table}')
+        expected_creates = statement_indexes(rf'create temporary table {expected_table}\s*\(.*\)')
+        expected_inserts = statement_indexes(rf'insert into {expected_table}\s*\(.*')
+        guard_initial_drops = statement_indexes(rf'drop temporary table if exists {conflict_guard}')
+        guard_creates = statement_indexes(rf'create temporary table {conflict_guard}\s*\(.*\)')
+        guard_inserts = statement_indexes(rf'insert into {conflict_guard}\s*\(.*')
+        guard_final_drops = statement_indexes(rf'drop temporary table {conflict_guard}')
+        expected_final_drops = statement_indexes(rf'drop temporary table {expected_table}')
+
+        markers = {
+            'expected table initial cleanup': expected_initial_drops,
+            'expected table creation': expected_creates,
+            'expected seed insertion': expected_inserts,
+            'conflict guard initial cleanup': guard_initial_drops,
+            'conflict guard creation': guard_creates,
+            'conflict guard check': guard_inserts,
+            'conflict guard success cleanup': guard_final_drops,
+            'expected table success cleanup': expected_final_drops,
+        }
+        for name, indexes in markers.items():
+            with self.subTest(marker=name):
+                self.assertEqual(1, len(indexes), f'missing or duplicate {name}')
+
+        legacy_guard_cleanup = normalized.index('drop temporary table pallet_workflow_legacy_seed_guard')
+        permanent_deletes = [
+            index
+            for index, statement in enumerate(normalized)
+            if re.match(r'delete from sys_(?:role_menu|menu|dict_data|dict_type)\b', statement)
+        ]
+        self.assertTrue(permanent_deletes)
+        first_permanent_delete = permanent_deletes[0]
+
+        ordered_markers = (
+            legacy_guard_cleanup,
+            expected_initial_drops[0],
+            expected_creates[0],
+            expected_inserts[0],
+            guard_initial_drops[0],
+            guard_creates[0],
+            guard_inserts[0],
+            guard_final_drops[0],
+            expected_final_drops[0],
+            first_permanent_delete,
+        )
+        self.assertEqual(tuple(sorted(ordered_markers)), ordered_markers)
+
+        expected_insert = statements[expected_inserts[0]].replace('`', '').lower()
+        expected_insert_match = re.fullmatch(
+            rf'\s*insert\s+into\s+{expected_table}\s*\((?P<columns>[^)]+)\)\s+values\s+.*',
+            expected_insert,
+            flags=re.S,
+        )
+        self.assertIsNotNone(expected_insert_match)
+        expected_columns = tuple(column.strip() for column in expected_insert_match.group('columns').split(','))
+        self.assertEqual(('menu_id', *MYSQL_MENU_STABLE_COLUMNS), expected_columns)
+
+        expected_rows = _insert_rows(statements[expected_inserts[0]], expected_table)
+        self.assertEqual(len(BUSINESS_MENU_IDS), len(expected_rows))
+        expected_rows_by_id = {int(row[0]): tuple(row) for row in expected_rows}
+        self.assertEqual(BUSINESS_MENU_IDS, set(expected_rows_by_id))
+        seed_rows_by_id = {
+            menu_id: tuple(row[: len(expected_columns)]) for menu_id, row in _business_menu_rows(raw).items()
+        }
+        self.assertEqual(seed_rows_by_id, expected_rows_by_id)
+
+        guard_create = normalized[guard_creates[0]]
+        self.assertRegex(guard_create, r'primary key\s*\(\s*guard_token\s*\)')
+        guard_insert = normalized[guard_inserts[0]]
+        self.assertEqual(2, guard_insert.count("select 'business-menu-seed-conflict'"))
+        self.assertIn(
+            "union all select 'business-menu-seed-conflict' as guard_token where exists (",
+            guard_insert,
+        )
+        self.assertNotIn(' on duplicate key update ', guard_insert)
+        expected_comparisons = ' and '.join(
+            f'actual.{column} <=> expected.{column}' for column in MYSQL_MENU_STABLE_COLUMNS
+        )
+        guard_insert = re.sub(r'\(\s+', '(', guard_insert)
+        guard_insert = re.sub(r'\s+\)', ')', guard_insert)
+        self.assertIn(
+            f'from sys_menu as actual join {expected_table} as expected '
+            f'on actual.menu_id = expected.menu_id where not ({expected_comparisons})',
+            guard_insert,
+        )
+
+        preflight_sql = '; '.join(normalized[expected_creates[0] : first_permanent_delete])
+        self.assertNotRegex(
+            preflight_sql,
+            r'\b(?:insert\s+into|replace\s+into|update|delete\s+from)\s+sys_menu\b',
+        )
+
+    def test_patch_wraps_permanent_seed_dml_in_explicit_transaction(self) -> None:
+        statements = _sql_statements(PATCH_PATH.read_text(encoding='utf-8'))
+        normalized = [_normalized_sql_statement(statement) for statement in statements]
+        start_transactions = [index for index, statement in enumerate(normalized) if statement == 'start transaction']
+        commits = [index for index, statement in enumerate(normalized) if statement == 'commit']
+        self.assertEqual(1, len(start_transactions), 'permanent seed DML must start one explicit transaction')
+        self.assertEqual(1, len(commits), 'permanent seed DML must commit one explicit transaction')
+        start_transaction = start_transactions[0]
+        commit = commits[0]
+
+        permanent_deletes = [
+            index
+            for index, statement in enumerate(normalized)
+            if re.match(r'delete from sys_(?:role_menu|menu|dict_data|dict_type)\b', statement)
+        ]
+        self.assertTrue(permanent_deletes)
+        menu_seed_inserts = [
+            index for index, statement in enumerate(normalized) if statement.startswith('insert into sys_menu values ')
+        ]
+        self.assertEqual(1, len(menu_seed_inserts))
+        menu_seed_insert = menu_seed_inserts[0]
+
+        ordered_markers = (
+            normalized.index(MYSQL_MENU_SQL_MODE_SAVE),
+            normalized.index(MYSQL_MENU_SQL_MODE_ENFORCE),
+            start_transaction,
+            permanent_deletes[0],
+            permanent_deletes[-1],
+            menu_seed_insert,
+            commit,
+            normalized.index(MYSQL_MENU_SQL_MODE_RESTORE),
+        )
+        self.assertEqual(tuple(sorted(ordered_markers)), ordered_markers)
+
+        permanent_mutations = [*permanent_deletes, menu_seed_insert]
+        self.assertTrue(all(start_transaction < index < commit for index in permanent_mutations))
+        self.assertTrue(
+            all(
+                index < start_transaction
+                for index, statement in enumerate(normalized)
+                if re.match(r'(?:create|drop) temporary table\b', statement)
+            )
+        )
+        transaction_sql = '; '.join(normalized[start_transaction + 1 : commit])
+        self.assertNotRegex(transaction_sql, r'\b(?:create|drop|alter|truncate)\b')
+
    def test_patch_cleanup_is_explicit_and_ordered(self) -> None:
        self.assertTrue(PATCH_PATH.is_file(), f'missing patch: {PATCH_PATH}')
        sql = _strip_sql_comments(PATCH_PATH.read_text(encoding='utf-8')).replace('`', '').lower()