测试：加固菜单补丁门禁契约

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

@@ -924,6 +924,68 @@ def _assert_business_permission_inventory(test_case: unittest.TestCase, sql: str
     test_case.assertEqual(Counter(EXPECTED_PERMISSIONS), actual_permissions)
 
 
+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 = [
+        statement for statement in normalized if statement.startswith('insert into promote_business_menus_guard ')
+    ]
+    test_case.assertEqual(2, len(guards))
+    pre_guard, post_guard = guards
+    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))
+        test_case.assertIn(f"'promote-business-menus-{phase}-verified'", guard)
+        duplicate_key_structure = (
+            r'^insert into promote_business_menus_guard \(guard_token\) '
+            r'select guard_token from \( select '
+            rf'{re.escape(conflict_token)} as guard_token union all select if\('
+        )
+        test_case.assertRegex(guard, duplicate_key_structure)
+        test_case.assertEqual(1, len(re.findall(r'\bpromote_business_menus_expected\b', guard)))
+        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)
+
+    old_state = (
+        r'sum\(\s*\(actual\.parent_id <=> expected\.old_parent_id\)\s*'
+        r'and\s*\(actual\.order_num <=> expected\.old_order_num\)\s*\)\s*=\s*6'
+    )
+    new_state = (
+        r'sum\(\s*\(actual\.parent_id <=> expected\.new_parent_id\)\s*'
+        r'and\s*\(actual\.order_num <=> expected\.new_order_num\)\s*\)\s*=\s*6'
+    )
+    precondition_state_group = (
+        rf'and\s*\(\s*{old_state}\s+or\s+{new_state}\s*\)\s*,\s*'
+        r"'promote-business-menus-precondition-verified'\s*,\s*"
+        r"'promote-business-menus-precondition-conflict'\s*\)"
+    )
+    test_case.assertRegex(pre_guard, precondition_state_group)
+    test_case.assertNotIn('old_parent_id', post_guard)
+    test_case.assertNotIn('old_order_num', post_guard)
+    postcondition_state = (
+        rf'and\s*{new_state}\s*,\s*'
+        r"'promote-business-menus-postcondition-verified'\s*,\s*"
+        r"'promote-business-menus-postcondition-conflict'\s*\)"
+    )
+    test_case.assertRegex(post_guard, postcondition_state)
+    return pre_guard, post_guard
+
+
 class SqlContractTest(unittest.TestCase):
     def test_initializers_define_exact_business_table_set(self) -> None:
         for filename in INITIALIZERS:
@@ -1303,6 +1365,7 @@ class SqlContractTest(unittest.TestCase):
         )
         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 = [
             index
             for index, statement in enumerate(normalized)
@@ -1310,42 +1373,7 @@ class SqlContractTest(unittest.TestCase):
         ]
         self.assertEqual(2, len(guard_indexes))
         pre_guard_index, post_guard_index = guard_indexes
-        pre_guard = normalized[pre_guard_index]
+        self.assertEqual((pre_guard, post_guard), tuple(normalized[index] for index in guard_indexes))
-        post_guard = normalized[post_guard_index]
-        for guard, phase in ((pre_guard, 'precondition'), (post_guard, 'postcondition')):
-            conflict_token = f"'promote-business-menus-{phase}-conflict'"
-            self.assertEqual(2, guard.count(conflict_token))
-            self.assertIn(f"'promote-business-menus-{phase}-verified'", guard)
-            self.assertIn('count(actual.menu_id) = 6', guard)
-            for field in (
-                'menu_name',
-                'path',
-                'component',
-                'query',
-                'route_name',
-                'menu_type',
-                'visible',
-                'status',
-                'perms',
-                'icon',
-            ):
-                self.assertIn(f'binary actual.{field} <=> binary expected.{field}', guard)
-            for field in ('is_frame', 'is_cache'):
-                self.assertIn(f'actual.{field} <=> expected.{field}', guard)
-
-        old_state = (
-            r'sum\(\s*\(actual\.parent_id <=> expected\.old_parent_id\)\s*'
-            r'and\s*\(actual\.order_num <=> expected\.old_order_num\)\s*\)\s*=\s*6'
-        )
-        new_state = (
-            r'sum\(\s*\(actual\.parent_id <=> expected\.new_parent_id\)\s*'
-            r'and\s*\(actual\.order_num <=> expected\.new_order_num\)\s*\)\s*=\s*6'
-        )
-        self.assertRegex(pre_guard, old_state)
-        self.assertRegex(pre_guard, new_state)
-        self.assertNotIn('old_parent_id', post_guard)
-        self.assertNotIn('old_order_num', post_guard)
-        self.assertRegex(post_guard, new_state)
 
         expected_update = _normalized_sql_statement(
             """
@@ -1404,6 +1432,30 @@ class SqlContractTest(unittest.TestCase):
         self.assertLess(guard_drop_index, len(normalized) - 1)
         self.assertLess(expected_drop_index, len(normalized) - 1)
 
+    def test_menu_promotion_guard_contract_rejects_structural_mutations(self) -> None:
+        raw = MENU_PROMOTION_PATCH_PATH.read_text(encoding='utf-8')
+        mutation_specs = (
+            ('union_distinct', r'\bunion\s+all\b', 'union distinct'),
+            (
+                'old_and_new_state',
+                r'(old_order_num\)\s*\)\s*=\s*6\s*)or(\s+sum\s*\()',
+                r'\1and\2',
+            ),
+            ('inner_join_actual', r'\bleft\s+join\s+sys_menu\s+as\s+actual\b', 'inner join sys_menu as actual'),
+            (
+                'duplicate_expected_reference',
+                r'\bfrom\s+promote_business_menus_expected\s+as\s+expected\b',
+                'from promote_business_menus_expected as expected '
+                'cross join promote_business_menus_expected as duplicate_expected',
+            ),
+        )
+        for name, pattern, replacement in mutation_specs:
+            with self.subTest(mutation=name):
+                mutated, replacements = re.subn(pattern, replacement, raw, count=1, flags=re.I)
+                self.assertEqual(1, replacements)
+                with self.assertRaises(AssertionError):
+                    _assert_menu_promotion_guard_structure(self, mutated)
+
     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()