niuniu/ShiErFeng
4
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

迁移：新增业务菜单层级补丁

Browse Source
This commit is contained in:
焦龙言
2026-07-27 21:02:48 +08:00
parent db5755586d
commit a1613f8780
2 changed files with 408 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
shierfeng-fastapi-backend/tests/module_business/test_sql_contract.py
+252
View File
@@ -1,6 +1,7 @@
import re
import unittest
from collections import Counter
from collections.abc import Callable
from pathlib import Path
SQL_ROOT = Path(__file__).resolve().parents[2] / 'sql'
@@ -1152,6 +1153,257 @@ class SqlContractTest(unittest.TestCase):
self.assertEqual(sorted(drop_positions), drop_positions)
self.assertLess(drop_positions[-1], sql.index('drop table if exists sys_user'))
def test_menu_promotion_patch_is_fail_closed_atomic_and_exact(self) -> None: # noqa: PLR0915
self.assertTrue(MENU_PROMOTION_PATCH_PATH.is_file(), f'missing patch: {MENU_PROMOTION_PATCH_PATH}')
raw = MENU_PROMOTION_PATCH_PATH.read_text(encoding='utf-8')
executable = _strip_sql_comments(raw).replace('`', '')
normalized = [_normalized_sql_statement(statement) for statement in _sql_statements(raw)]
for required_note in ('未自动执行', '执行前备份', '可幂等重跑', '禁止使用 mysql --force', '独立连接'):
self.assertIn(required_note, raw)
self.assertNotIn('--force', executable.lower())
self.assertNotIn('continue-on-error', raw.lower())
expected_rows = [
('1', '系统管理', 'system', None, '', '', '1', '0', 'M', '0', '0', '', 'system', '0', '1', '0', '4'),
('2', '系统监控', 'monitor', None, '', '', '1', '0', 'M', '0', '0', '', 'monitor', '0', '2', '0', '5'),
('3', '系统工具', 'tool', None, '', '', '1', '0', 'M', '0', '0', '', 'tool', '0', '3', '0', '6'),
(
'121',
'基础资料维护',
'businessMaster',
None,
'',
'',
'1',
'0',
'M',
'0',
'0',
'',
'#',
'1',
'10',
'0',
'1',
),
(
'122',
'业务管理',
'businessOperation',
None,
'',
'',
'1',
'0',
'M',
'0',
'0',
'',
'#',
'1',
'11',
'0',
'2',
),
(
'123',
'报表查询',
'businessReport',
None,
'',
'',
'1',
'0',
'M',
'0',
'0',
'',
'#',
'1',
'12',
'0',
'3',
),
]
self.assertEqual(expected_rows, [tuple(row) for row in _insert_rows(raw, 'promote_business_menus_expected')])
def unique_index(predicate: Callable[[str], bool], label: str) -> int:
matches = [index for index, statement in enumerate(normalized) if predicate(statement)]
self.assertEqual(1, len(matches), label)
return matches[0]
expected_columns = (
'menu_id',
'menu_name',
'path',
'component',
'query',
'route_name',
'is_frame',
'is_cache',
'menu_type',
'visible',
'status',
'perms',
'icon',
'old_parent_id',
'old_order_num',
'new_parent_id',
'new_order_num',
)
expected_create_index = unique_index(
lambda statement: statement.startswith('create temporary table promote_business_menus_expected '),
'expected table creation',
)
expected_insert_index = unique_index(
lambda statement: statement.startswith('insert into promote_business_menus_expected '),
'expected row insertion',
)
expected_create = normalized[expected_create_index]
for column in expected_columns:
self.assertRegex(expected_create, rf'\b{column}\b')
guard_create_index = unique_index(
lambda statement: statement.startswith('create temporary table promote_business_menus_guard '),
'guard table creation',
)
self.assertIn('primary key (guard_token)', normalized[guard_create_index])
start_indexes = [index for index, statement in enumerate(normalized) if statement == 'start transaction']
commit_indexes = [index for index, statement in enumerate(normalized) if statement == 'commit']
self.assertEqual(1, len(start_indexes))
self.assertEqual(1, len(commit_indexes))
start_index = start_indexes[0]
commit_index = commit_indexes[0]
initial_cleanup_indexes = (
unique_index(
lambda statement: statement == 'drop temporary table if exists promote_business_menus_expected',
'expected table initial cleanup',
),
unique_index(
lambda statement: statement == 'drop temporary table if exists promote_business_menus_guard',
'guard table initial cleanup',
),
)
self.assertTrue(
all(
index < start_index
for index in (
*initial_cleanup_indexes,
expected_create_index,
expected_insert_index,
guard_create_index,
)
)
)
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 == lock_statement, 'six-row locking read')
guard_indexes = [
index
for index, statement in enumerate(normalized)
if statement.startswith('insert into promote_business_menus_guard ')
]
self.assertEqual(2, len(guard_indexes))
pre_guard_index, post_guard_index = guard_indexes
pre_guard = normalized[pre_guard_index]
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(
"""
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)
"""
)
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):
continue
if statement.startswith('insert into promote_business_menus_expected '):
continue
if statement.startswith('insert into promote_business_menus_guard '):
continue
permanent_dml.append(statement)
self.assertEqual([expected_update], permanent_dml)
self.assertNotRegex(executable.lower(), r'\bsys_role_menu\b')
self.assertLess(start_index, lock_index)
self.assertLess(lock_index, pre_guard_index)
self.assertLess(pre_guard_index, update_index)
self.assertLess(update_index, post_guard_index)
self.assertLess(post_guard_index, commit_index)
guard_drop_index = unique_index(
lambda statement: statement == 'drop temporary table promote_business_menus_guard',
'guard table final cleanup',
)
expected_drop_index = unique_index(
lambda statement: statement == 'drop temporary table promote_business_menus_expected',
'expected table final cleanup',
)
self.assertLess(commit_index, guard_drop_index)
self.assertLess(commit_index, expected_drop_index)
expected_verification = (
'select menu_id, menu_name, parent_id, order_num, path, component, menu_type, visible, status, perms '
'from sys_menu where menu_id in (1, 2, 3, 121, 122, 123) order by order_num, menu_id'
)
self.assertEqual(expected_verification, normalized[-1])
self.assertLess(guard_drop_index, len(normalized) - 1)
self.assertLess(expected_drop_index, len(normalized) - 1)
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()