From 9ef214f040df17b47583abfbcf9a5dc1bf3a8b2c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E7=84=A6=E9=BE=99=E8=A8=80?= Date: Mon, 27 Jul 2026 21:41:12 +0800 Subject: [PATCH] =?UTF-8?q?=E6=B5=8B=E8=AF=95=EF=BC=9A=E5=B0=81=E9=97=AD?= =?UTF-8?q?=E8=8F=9C=E5=8D=95=E8=A1=A5=E4=B8=81=E5=8E=9F=E5=AD=90=E6=80=A7?= =?UTF-8?q?=E5=9B=9E=E5=BD=92?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../module_business/test_sql_contract.py | 144 +++++++++++++----- 1 file changed, 107 insertions(+), 37 deletions(-) diff --git a/shierfeng-fastapi-backend/tests/module_business/test_sql_contract.py b/shierfeng-fastapi-backend/tests/module_business/test_sql_contract.py index 4ed6888..300e8d5 100644 --- a/shierfeng-fastapi-backend/tests/module_business/test_sql_contract.py +++ b/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()