测试：封闭菜单补丁原子性回归

shierfeng-fastapi-backend/tests/module_business/test_sql_contract.py

@@ -924,6 +924,30 @@ def _assert_business_permission_inventory(test_case: unittest.TestCase, sql: str
     test_case.assertEqual(Counter(EXPECTED_PERMISSIONS), actual_permissions)
 
 
+MENU_PROMOTION_LOCK_STATEMENT = (
+    'select menu_id from sys_menu where menu_id in (1, 2, 3, 121, 122, 123) order by menu_id for update'
+)
+MENU_PROMOTION_EXPECTED_UPDATE = _normalized_sql_statement(
+    """
+    update sys_menu
+    set parent_id = case
+          when menu_id in (121, 122, 123) then 0
+          else parent_id
+        end,
+        order_num = case menu_id
+          when 121 then 1
+          when 122 then 2
+          when 123 then 3
+          when 1 then 4
+          when 2 then 5
+          when 3 then 6
+          else order_num
+        end
+    where menu_id in (1, 2, 3, 121, 122, 123)
+    """
+)
+
+
 def _assert_menu_promotion_guard_structure(test_case: unittest.TestCase, sql: str) -> tuple[str, str]:
     normalized = [_normalized_sql_statement(statement) for statement in _sql_statements(sql)]
     guards = [
@@ -931,6 +955,29 @@ def _assert_menu_promotion_guard_structure(test_case: unittest.TestCase, sql: st
     ]
     test_case.assertEqual(2, len(guards))
     pre_guard, post_guard = guards
+    identity_fields = (
+        ('menu_name', True),
+        ('path', True),
+        ('component', True),
+        ('query', True),
+        ('route_name', True),
+        ('is_frame', False),
+        ('is_cache', False),
+        ('menu_type', True),
+        ('visible', True),
+        ('status', True),
+        ('perms', True),
+        ('icon', True),
+    )
+    identity_comparisons = [
+        (
+            rf'\(binary\s+actual\.{field}\s*<=>\s*binary\s+expected\.{field}\)'
+            if binary
+            else rf'\(actual\.{field}\s*<=>\s*expected\.{field}\)'
+        )
+        for field, binary in identity_fields
+    ]
+    identity_aggregate = r'sum\s*\(\s*' + r'\s+and\s+'.join(identity_comparisons) + r'\s*\)\s*=\s*6'
     for guard, phase in ((pre_guard, 'precondition'), (post_guard, 'postcondition')):
         conflict_token = f"'promote-business-menus-{phase}-conflict'"
         test_case.assertEqual(2, guard.count(conflict_token))
@@ -945,21 +992,7 @@ def _assert_menu_promotion_guard_structure(test_case: unittest.TestCase, sql: st
         expected_join = 'left join sys_menu as actual on actual.menu_id = expected.menu_id'
         test_case.assertEqual(1, guard.count(expected_join))
         test_case.assertIn('count(actual.menu_id) = 6', guard)
-        for field in (
-            'menu_name',
-            'path',
-            'component',
-            'query',
-            'route_name',
-            'menu_type',
-            'visible',
-            'status',
-            'perms',
-            'icon',
-        ):
-            test_case.assertIn(f'binary actual.{field} <=> binary expected.{field}', guard)
-        for field in ('is_frame', 'is_cache'):
-            test_case.assertIn(f'actual.{field} <=> expected.{field}', guard)
+        test_case.assertRegex(guard, identity_aggregate)
 
     old_state = (
         r'sum\(\s*\(actual\.parent_id <=> expected\.old_parent_id\)\s*'
@@ -986,6 +1019,31 @@ def _assert_menu_promotion_guard_structure(test_case: unittest.TestCase, sql: st
     return pre_guard, post_guard
 
 
+def _assert_menu_promotion_transaction_body(test_case: unittest.TestCase, sql: str) -> None:
+    normalized = [_normalized_sql_statement(statement) for statement in _sql_statements(sql)]
+    pre_guard, post_guard = _assert_menu_promotion_guard_structure(test_case, sql)
+
+    def unique_index(statement: str) -> int:
+        matches = [index for index, actual in enumerate(normalized) if actual == statement]
+        test_case.assertEqual(1, len(matches), statement)
+        return matches[0]
+
+    start_index = unique_index('start transaction')
+    lock_index = unique_index(MENU_PROMOTION_LOCK_STATEMENT)
+    pre_guard_index = unique_index(pre_guard)
+    update_index = unique_index(MENU_PROMOTION_EXPECTED_UPDATE)
+    post_guard_index = unique_index(post_guard)
+    commit_index = unique_index('commit')
+    test_case.assertEqual(
+        sorted((start_index, lock_index, pre_guard_index, update_index, post_guard_index, commit_index)),
+        [start_index, lock_index, pre_guard_index, update_index, post_guard_index, commit_index],
+    )
+    test_case.assertEqual(
+        [MENU_PROMOTION_LOCK_STATEMENT, pre_guard, MENU_PROMOTION_EXPECTED_UPDATE, post_guard],
+        normalized[start_index + 1 : commit_index],
+    )
+
+
 class SqlContractTest(unittest.TestCase):
     def test_initializers_define_exact_business_table_set(self) -> None:
         for filename in INITIALIZERS:
@@ -1360,10 +1418,10 @@ class SqlContractTest(unittest.TestCase):
             )
         )
 
-        lock_statement = (
-            'select menu_id from sys_menu where menu_id in (1, 2, 3, 121, 122, 123) order by menu_id for update'
+        lock_index = unique_index(
+            lambda statement: statement == MENU_PROMOTION_LOCK_STATEMENT,
+            'six-row locking read',
         )
-        lock_index = unique_index(lambda statement: statement == lock_statement, 'six-row locking read')
 
         pre_guard, post_guard = _assert_menu_promotion_guard_structure(self, raw)
         guard_indexes = [
@@ -1375,26 +1433,12 @@ class SqlContractTest(unittest.TestCase):
         pre_guard_index, post_guard_index = guard_indexes
         self.assertEqual((pre_guard, post_guard), tuple(normalized[index] for index in guard_indexes))
 
-        expected_update = _normalized_sql_statement(
-            """
-            update sys_menu
-            set parent_id = case
-                  when menu_id in (121, 122, 123) then 0
-                  else parent_id
-                end,
-                order_num = case menu_id
-                  when 121 then 1
-                  when 122 then 2
-                  when 123 then 3
-                  when 1 then 4
-                  when 2 then 5
-                  when 3 then 6
-                  else order_num
-                end
-            where menu_id in (1, 2, 3, 121, 122, 123)
-            """
+        expected_update = MENU_PROMOTION_EXPECTED_UPDATE
+        _assert_menu_promotion_transaction_body(self, raw)
+        update_index = unique_index(
+            lambda statement: statement == expected_update,
+            'exact permanent update',
         )
-        update_index = unique_index(lambda statement: statement == expected_update, 'exact permanent update')
         permanent_dml = []
         for statement in normalized:
             if not re.match(r'^(?:insert|update|delete|replace|truncate)\b', statement):
@@ -1456,6 +1500,32 @@ class SqlContractTest(unittest.TestCase):
                 with self.assertRaises(AssertionError):
                     _assert_menu_promotion_guard_structure(self, mutated)
 
+    def test_menu_promotion_contract_rejects_identity_threshold_and_transaction_body_mutations(self) -> None:
+        raw = MENU_PROMOTION_PATCH_PATH.read_text(encoding='utf-8')
+        identity_threshold_mutation, replacements = re.subn(
+            r'(\(binary\s+actual\.icon\s+<=>\s+binary\s+expected\.icon\)\s*\)\s*)=\s*6',
+            r'\1>= 0',
+            raw,
+            count=1,
+            flags=re.I,
+        )
+        with self.subTest(mutation='identity_threshold'):
+            self.assertEqual(1, replacements)
+            with self.assertRaises(AssertionError):
+                _assert_menu_promotion_guard_structure(self, identity_threshold_mutation)
+
+        transaction_ddl_mutation, replacements = re.subn(
+            r'(start\s+transaction\s*;)',
+            r'\1\nalter table sys_menu engine=innodb;',
+            raw,
+            count=1,
+            flags=re.I,
+        )
+        with self.subTest(mutation='transaction_ddl'):
+            self.assertEqual(1, replacements)
+            with self.assertRaises(AssertionError):
+                _assert_menu_promotion_transaction_body(self, transaction_ddl_mutation)
+
     def test_patch_exists_and_preserves_ai_business_tables(self) -> None:
         self.assertTrue(PATCH_PATH.is_file(), f'missing patch: {PATCH_PATH}')
         raw = PATCH_PATH.read_text(encoding='utf-8').lower()