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 71a02bf..4ed6888 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,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] - 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) + self.assertEqual((pre_guard, post_guard), tuple(normalized[index] for index in guard_indexes)) 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()