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

修复：校验 MySQL 菜单固定ID冲突

Browse Source
This commit is contained in:
焦龙言
2026-07-27 14:24:30 +08:00
parent 3d1b64971f
commit 9b5f479f0d
2 changed files with 191 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files
shierfeng-fastapi-backend/tests/module_business/test_sql_contract.py
+143 -3
View File
@@ -409,6 +409,29 @@ EXPECTED_BUTTONS = {
)
}
BUSINESS_MENU_IDS = set(range(121, 124)) | set(range(502, 519)) | set(range(1065, 1109))
MYSQL_MENU_STABLE_COLUMNS = (
'menu_name',
'parent_id',
'order_num',
'path',
'component',
'query',
'route_name',
'is_frame',
'is_cache',
'menu_type',
'visible',
'status',
'perms',
'icon',
)
MYSQL_MENU_SQL_MODE_SAVE = 'set @pallet_workflow_original_sql_mode = @@session.sql_mode'
MYSQL_MENU_SQL_MODE_ENFORCE = (
"set session sql_mode = concat_ws(',', nullif(@pallet_workflow_original_sql_mode, ''), "
"if(find_in_set('strict_all_tables', @pallet_workflow_original_sql_mode) = 0, "
"'strict_all_tables', null))"
)
MYSQL_MENU_SQL_MODE_RESTORE = 'set session sql_mode = @pallet_workflow_original_sql_mode'
def _strip_sql_comments(sql: str) -> str:
@@ -481,6 +504,36 @@ def _sql_statements(sql: str) -> list[str]:
return [statement.strip() for statement in _strip_sql_comments(sql).split(';') if statement.strip()]
def _normalized_sql_statement(statement: str) -> str:
return re.sub(r'\s+', ' ', statement.replace('`', '')).strip().lower()
def _mysql_menu_seed_has_safe_sql_mode_order(statements: list[str]) -> bool:
normalized = [_normalized_sql_statement(statement) for statement in statements]
predicates = (
lambda statement: statement == MYSQL_MENU_SQL_MODE_SAVE,
lambda statement: statement == MYSQL_MENU_SQL_MODE_ENFORCE,
lambda statement: (
statement.startswith('insert into sys_menu values ') and ' on duplicate key update ' in statement
),
lambda statement: statement == MYSQL_MENU_SQL_MODE_RESTORE,
lambda statement: (
re.fullmatch(
r'select\s+.+\s+from\s+sys_menu\s+where\s+menu_id\s+in\s*\([^)]+\)',
statement,
)
is not None
),
)
positions = []
for predicate in predicates:
matches = [index for index, statement in enumerate(normalized) if predicate(statement)]
if len(matches) != 1:
return False
positions.append(matches[0])
return positions == sorted(positions) and positions[-1] == len(statements) - 1
def _insert_rows(sql: str, table: str) -> list[list[str | None]]:
rows = []
pattern = re.compile(
@@ -730,7 +783,7 @@ class SqlContractTest(unittest.TestCase):
def test_mysql_and_postgresql_business_menu_fields_are_identical(self) -> None:
mysql_rows = _business_menu_rows((SQL_ROOT / INITIALIZERS[0]).read_text(encoding='utf-8'))
postgres_rows = _business_menu_rows((SQL_ROOT / INITIALIZERS[1]).read_text(encoding='utf-8'))
compared_indexes = (0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13)
compared_indexes = tuple(range(15))
mysql_contract = {
menu_id: tuple(row[index] for index in compared_indexes) for menu_id, row in mysql_rows.items()
}
@@ -863,12 +916,13 @@ class SqlContractTest(unittest.TestCase):
self.assertEqual(EXPECTED_CHECKS[table], checks)
patch_rows = _business_menu_rows(raw)
mysql_rows = _business_menu_rows((SQL_ROOT / 'shierfeng-fastapi.sql').read_text(encoding='utf-8'))
compared_indexes = (0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13)
compared_indexes = tuple(range(15))
self.assertEqual(
{menu_id: tuple(row[index] for index in compared_indexes) for menu_id, row in mysql_rows.items()},
{menu_id: tuple(row[index] for index in compared_indexes) for menu_id, row in patch_rows.items()},
)
self.assertRegex(lower, r'\bon\s+duplicate\s+key\s+update\b|\binsert\s+ignore\s+into\s+sys_menu\b')
self.assertRegex(lower, r'\bon\s+duplicate\s+key\s+update\b')
self.assertNotRegex(lower, r'\binsert\s+ignore\b')
self.assertNotRegex(_strip_sql_comments(lower), r'\binsert(?:\s+ignore)?\s+into\s+`?sys_role_menu`?\b')
self.assertIn('/getRouters', raw)
self.assertIn('用户信息', raw)
@@ -878,6 +932,92 @@ class SqlContractTest(unittest.TestCase):
self.assertNotIn('sys_dict:ai_provider_type', executable)
self.assertNotRegex(lower, r'\bbigserial\b|\balter\s+sequence\b|\bcomment\s+on\b')
def test_patch_menu_seed_fails_closed_on_fixed_id_conflicts(self) -> None:
raw = PATCH_PATH.read_text(encoding='utf-8')
executable = _strip_sql_comments(raw).replace('`', '').lower()
self.assertNotRegex(executable, r'\binsert\s+ignore\b')
statements = _sql_statements(raw)
self.assertTrue(_mysql_menu_seed_has_safe_sql_mode_order(statements))
menu_insert_statements = [
statement.replace('`', '').lower()
for statement in statements
if re.match(r'\s*insert(?:\s+ignore)?\s+into\s+`?sys_menu`?\b', statement, flags=re.I)
]
self.assertEqual(1, len(menu_insert_statements))
menu_insert_statement = menu_insert_statements[0]
menu_insert = re.fullmatch(
r'\s*insert\s+into\s+sys_menu\s+values\s*.*?'
r'\bon\s+duplicate\s+key\s+update\s+'
r'menu_name\s*=\s*if\s*\(\s*(?P<guard>.*?)\s*,\s*menu_name\s*,\s*null\s*\)\s*',
menu_insert_statement,
flags=re.S,
)
self.assertIsNotNone(menu_insert, 'sys_menu seed must use a fail-closed upsert')
self.assertEqual(len(BUSINESS_MENU_IDS), len(_insert_rows(menu_insert_statement, 'sys_menu')))
expected_guard = ' and '.join(f'{column} <=> values({column})' for column in MYSQL_MENU_STABLE_COLUMNS)
actual_guard = re.sub(r'\s+', ' ', menu_insert.group('guard')).strip()
self.assertEqual(expected_guard, actual_guard)
self.assertIn('任一稳定字段冲突', raw)
self.assertIn('menu_name NOT NULL', raw)
self.assertIn('中止整条语句', raw)
self.assertIn('人工解决冲突', raw)
self.assertIn('不会静默覆盖未知 ID', raw)
verification_statement = statements[-1].replace('`', '').lower()
verification = re.fullmatch(
r'\s*select\s+(?P<columns>.*?)\s+from\s+sys_menu\s+'
r'where\s+menu_id\s+in\s*\((?P<ids>[^)]+)\)\s*',
verification_statement,
flags=re.S,
)
self.assertIsNotNone(verification, 'patch must end with a manual business menu verification query')
verification_columns = tuple(column.strip() for column in verification.group('columns').split(','))
self.assertEqual(('menu_id', *MYSQL_MENU_STABLE_COLUMNS), verification_columns)
verification_ids = [int(value.strip()) for value in verification.group('ids').split(',')]
self.assertEqual(len(BUSINESS_MENU_IDS), len(verification_ids))
self.assertEqual(BUSINESS_MENU_IDS, set(verification_ids))
self.assertRegex(raw, r'查询结果必须为\s*64')
normalized = [_normalized_sql_statement(statement) for statement in statements]
marker_indexes = {
'save': normalized.index(MYSQL_MENU_SQL_MODE_SAVE),
'enforce': normalized.index(MYSQL_MENU_SQL_MODE_ENFORCE),
'insert': next(
index
for index, statement in enumerate(normalized)
if statement.startswith('insert into sys_menu values ')
),
'restore': normalized.index(MYSQL_MENU_SQL_MODE_RESTORE),
'verification': len(statements) - 1,
}
mutations = {
f'missing_{name}': [
statement for index, statement in enumerate(statements) if index != marker_indexes[name]
]
for name in ('save', 'enforce', 'restore')
}
strict_after_insert = statements.copy()
strict_after_insert[marker_indexes['enforce']], strict_after_insert[marker_indexes['insert']] = (
strict_after_insert[marker_indexes['insert']],
strict_after_insert[marker_indexes['enforce']],
)
mutations['strict_after_insert'] = strict_after_insert
restore_after_verification = statements.copy()
(
restore_after_verification[marker_indexes['restore']],
restore_after_verification[marker_indexes['verification']],
) = (
restore_after_verification[marker_indexes['verification']],
restore_after_verification[marker_indexes['restore']],
)
mutations['restore_after_verification'] = restore_after_verification
for name, mutation in mutations.items():
with self.subTest(mutation=name):
self.assertFalse(_mysql_menu_seed_has_safe_sql_mode_order(mutation))
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()