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-28 08:43:31 +08:00
parent 9ef214f040
commit b4822fef12
5 changed files with 368 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files
shierfeng-fastapi-backend/tests/module_business/test_sql_contract.py
+208 -5
View File
@@ -8,8 +8,10 @@ SQL_ROOT = Path(__file__).resolve().parents[2] / 'sql'
INITIALIZERS = ('shierfeng-fastapi.sql', 'shierfeng-fastapi-pg.sql')
PATCH_PATH = SQL_ROOT / 'patches/20260724_pallet_workflow_mysql.sql'
MENU_PROMOTION_PATCH_PATH = SQL_ROOT / 'patches/20260727_promote_business_menus_mysql.sql'
BUSINESS_MENU_ICON_PATCH_PATH = SQL_ROOT / 'patches/20260728_add_business_menu_icons_mysql.sql'
SYS_MENU_VALUE_COUNT = 20
SYS_MENU_PERMS_INDEX = 13
SYS_MENU_ICON_INDEX = 14
REMOVED_AI_DICT_CODES = set(range(33, 70))
EXPECTED_TABLES = {
@@ -591,10 +593,15 @@ EXPECTED_TOP_LEVEL_MENU_POSITIONS = {
2: ('系统监控', 0, 5),
3: ('系统工具', 0, 6),
}
EXPECTED_BUSINESS_ROOT_ICONS = {
121: 'dict',
122: 'form',
123: 'chart',
}
EXPECTED_TOP_LEVEL_MENU_STABLE_FIELDS = {
121: ('基础资料维护', 'businessMaster', None, '', '', '1', '0', 'M', '0', '0', '', '#'),
122: ('业务管理', 'businessOperation', None, '', '', '1', '0', 'M', '0', '0', '', '#'),
123: ('报表查询', 'businessReport', None, '', '', '1', '0', 'M', '0', '0', '', '#'),
121: ('基础资料维护', 'businessMaster', None, '', '', '1', '0', 'M', '0', '0', '', 'dict'),
122: ('业务管理', 'businessOperation', None, '', '', '1', '0', 'M', '0', '0', '', 'form'),
123: ('报表查询', 'businessReport', None, '', '', '1', '0', 'M', '0', '0', '', 'chart'),
1: ('系统管理', 'system', None, '', '', '1', '0', 'M', '0', '0', '', 'system'),
2: ('系统监控', 'monitor', None, '', '', '1', '0', 'M', '0', '0', '', 'monitor'),
3: ('系统工具', 'tool', None, '', '', '1', '0', 'M', '0', '0', '', 'tool'),
@@ -1044,6 +1051,82 @@ def _assert_menu_promotion_transaction_body(test_case: unittest.TestCase, sql: s
)
def _assert_business_menu_icon_guard_primary_key(test_case: unittest.TestCase, sql: str) -> None:
normalized = [_normalized_sql_statement(statement) for statement in _sql_statements(sql)]
guard_creates = [
statement
for statement in normalized
if statement.startswith('create temporary table business_menu_icon_guard ')
]
test_case.assertEqual(1, len(guard_creates))
test_case.assertIn('primary key (guard_token)', guard_creates[0])
def _assert_business_menu_icon_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 business_menu_icon_guard ')
]
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),
('parent_id', False),
('order_num', False),
)
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*3'
for guard, phase in ((pre_guard, 'precondition'), (post_guard, 'postcondition')):
conflict_token = f"'business-menu-icon-{phase}-conflict'"
test_case.assertEqual(2, guard.count(conflict_token))
test_case.assertIn(f"'business-menu-icon-{phase}-verified'", guard)
duplicate_key_structure = (
r'^insert into business_menu_icon_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'\bbusiness_menu_icon_expected\b', guard)))
test_case.assertIn('count(actual.menu_id) = 3', guard)
test_case.assertIn(
'left join sys_menu as actual on actual.menu_id = expected.menu_id',
guard,
)
test_case.assertRegex(guard, identity_aggregate)
for field in (
'menu_name',
'path',
'component',
'query',
'route_name',
'menu_type',
'visible',
'status',
'perms',
):
test_case.assertIn(f'binary actual.{field} <=> binary expected.{field}', guard)
for field in ('is_frame', 'is_cache', 'parent_id', 'order_num'):
test_case.assertIn(f'actual.{field} <=> expected.{field}', guard)
return pre_guard, post_guard
class SqlContractTest(unittest.TestCase):
def test_initializers_define_exact_business_table_set(self) -> None:
for filename in INITIALIZERS:
@@ -1142,10 +1225,121 @@ class SqlContractTest(unittest.TestCase):
sql = (SQL_ROOT / filename).read_text(encoding='utf-8')
_assert_top_level_menu_fields(self, sql)
def test_initializers_define_renderable_business_root_icons(self) -> None:
icon_root = SQL_ROOT.parent.parent / 'shierfeng-fastapi-frontend' / 'src' / 'assets' / 'icons' / 'svg'
for icon in EXPECTED_BUSINESS_ROOT_ICONS.values():
with self.subTest(icon=icon):
self.assertTrue((icon_root / f'{icon}.svg').is_file())
for filename in INITIALIZERS:
with self.subTest(filename=filename):
rows = _menu_rows((SQL_ROOT / filename).read_text(encoding='utf-8'), set(EXPECTED_BUSINESS_ROOT_ICONS))
actual = {menu_id: row[SYS_MENU_ICON_INDEX] for menu_id, row in rows.items()}
self.assertEqual(EXPECTED_BUSINESS_ROOT_ICONS, actual)
def test_business_menu_icon_patch_is_fail_closed_atomic_and_exact(self) -> None:
self.assertTrue(BUSINESS_MENU_ICON_PATCH_PATH.is_file(), f'missing patch: {BUSINESS_MENU_ICON_PATCH_PATH}')
raw = BUSINESS_MENU_ICON_PATCH_PATH.read_text(encoding='utf-8')
executable = _strip_sql_comments(raw).replace('`', '')
normalized = [_normalized_sql_statement(statement) for statement in _sql_statements(raw)]
_assert_business_menu_icon_guard_primary_key(self, raw)
pre_guard, post_guard = _assert_business_menu_icon_guard_structure(self, raw)
for required_note in (
'未自动执行',
'执行前备份',
'可幂等重跑',
'禁止使用 mysql --force',
'独立连接',
'20260727_promote_business_menus_mysql.sql',
'禁止逆序重跑旧补丁',
):
self.assertIn(required_note, raw)
self.assertNotIn('--force', executable.lower())
self.assertNotRegex(executable.lower(), r'\bsys_role_menu\b')
expected_rows = [
('121', '基础资料维护', 'businessMaster', None, '', '', '1', '0', 'M', '0', '0', '', '0', '1', '#', 'dict'),
('122', '业务管理', 'businessOperation', None, '', '', '1', '0', 'M', '0', '0', '', '0', '2', '#', 'form'),
('123', '报表查询', 'businessReport', None, '', '', '1', '0', 'M', '0', '0', '', '0', '3', '#', 'chart'),
]
self.assertEqual(
expected_rows,
[tuple(row) for row in _insert_rows(raw, 'business_menu_icon_expected')],
)
start_index = normalized.index('start transaction')
commit_index = normalized.index('commit')
self.assertEqual(1, normalized.count('start transaction'))
self.assertEqual(1, normalized.count('commit'))
transaction_body = normalized[start_index + 1 : commit_index]
lock_statement = (
'select menu_id from sys_menu where menu_id in (121, 122, 123) order by menu_id for update'
)
expected_update = (
"update sys_menu set icon = case menu_id when 121 then 'dict' when 122 then 'form' "
"when 123 then 'chart' else icon end where menu_id in (121, 122, 123)"
)
self.assertEqual(lock_statement, transaction_body[0])
self.assertEqual(expected_update, transaction_body[2])
self.assertEqual(4, len(transaction_body))
self.assertRegex(
pre_guard,
r'sum\(\(binary actual\.icon <=> binary expected\.old_icon\)\) = 3 or '
r'sum\(\(binary actual\.icon <=> binary expected\.new_icon\)\) = 3',
)
self.assertNotIn('old_icon', post_guard)
self.assertIn('sum((binary actual.icon <=> binary expected.new_icon)) = 3', post_guard)
permanent_dml = [
statement
for statement in normalized
if re.match(r'^(?:insert|update|delete|replace|truncate)\b', statement)
and not statement.startswith('insert into business_menu_icon_expected ')
and not statement.startswith('insert into business_menu_icon_guard ')
]
self.assertEqual([expected_update], permanent_dml)
self.assertEqual(
'select menu_id, menu_name, parent_id, order_num, path, icon from sys_menu '
'where menu_id in (121, 122, 123) order by order_num, menu_id',
normalized[-1],
)
def test_business_menu_icon_patch_rejects_guard_primary_key_removal(self) -> None:
raw = BUSINESS_MENU_ICON_PATCH_PATH.read_text(encoding='utf-8')
mutated, replacements = re.subn(
r',\s*primary\s+key\s*\(guard_token\)',
'',
raw,
count=1,
flags=re.I,
)
self.assertEqual(1, replacements)
with self.assertRaises(AssertionError):
_assert_business_menu_icon_guard_primary_key(self, mutated)
def test_business_menu_icon_guard_contract_rejects_structural_mutations(self) -> None:
raw = BUSINESS_MENU_ICON_PATCH_PATH.read_text(encoding='utf-8')
mutation_specs = (
('union_distinct', r'\bunion\s+all\b', 'union distinct'),
(
'identity_threshold',
r'(\(actual\.order_num\s*<=>\s*expected\.order_num\)\s*\)\s*)=\s*3',
r'\1>= 0',
),
)
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_business_menu_icon_guard_structure(self, mutated)
def test_top_level_menu_contract_rejects_same_stable_field_drift_in_both_initializers(self) -> None:
mutations = (
(1, '系统管理', 'system', 'driftSystemIcon'),
(121, '基础资料维护', '#', 'driftBusinessIcon'),
(121, '基础资料维护', 'dict', 'driftBusinessIcon'),
)
for filename in INITIALIZERS:
sql = (SQL_ROOT / filename).read_text(encoding='utf-8')
@@ -1580,8 +1774,17 @@ class SqlContractTest(unittest.TestCase):
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, *range(4, 15))
expected_legacy_rows = {
menu_id: tuple(
'#'
if index == SYS_MENU_ICON_INDEX and menu_id in EXPECTED_BUSINESS_ROOT_ICONS
else row[index]
for index in compared_indexes
)
for menu_id, row in mysql_rows.items()
}
self.assertEqual(
{menu_id: tuple(row[index] for index in compared_indexes) for menu_id, row in mysql_rows.items()},
expected_legacy_rows,
{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')