import re import unittest from pathlib import Path 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' SYS_MENU_VALUE_COUNT = 20 REMOVED_AI_DICT_CODES = set(range(33, 70)) EXPECTED_TABLES = { 'biz_strain', 'biz_warehouse', 'biz_customer', 'biz_customer_site', 'biz_production_line', 'biz_pallet', 'biz_user_warehouse', 'biz_user_site', 'biz_inbound_order', 'biz_inbound_item', 'biz_filling_record', 'biz_shipment_order', 'biz_shipment_item', 'biz_usage_record', 'biz_return_order', 'biz_return_item', 'biz_pallet_inventory', 'biz_pallet_event', } AUDIT_COLUMNS = {'create_by', 'create_time', 'update_by', 'update_time', 'remark'} CREATE_AUDIT_COLUMNS = {'create_by', 'create_time'} EXPECTED_COLUMNS = { 'biz_strain': {'strain_id', 'strain_code', 'strain_name', 'status'} | AUDIT_COLUMNS, 'biz_warehouse': {'warehouse_id', 'warehouse_code', 'warehouse_name', 'address', 'status'} | AUDIT_COLUMNS, 'biz_customer': {'customer_id', 'customer_code', 'customer_name', 'status'} | AUDIT_COLUMNS, 'biz_customer_site': {'site_id', 'customer_id', 'site_code', 'site_name', 'address', 'status'} | AUDIT_COLUMNS, 'biz_production_line': {'line_id', 'site_id', 'line_code', 'line_name', 'status'} | AUDIT_COLUMNS, 'biz_pallet': {'pallet_id', 'pallet_code', 'qr_token', 'capacity_packages', 'asset_status'} | AUDIT_COLUMNS, 'biz_user_warehouse': {'user_id', 'warehouse_id'} | CREATE_AUDIT_COLUMNS, 'biz_user_site': {'user_id', 'site_id'} | CREATE_AUDIT_COLUMNS, 'biz_inbound_order': { 'inbound_id', 'inbound_no', 'warehouse_id', 'status', 'confirm_idempotency_key', 'confirm_request_hash', 'confirmed_by', 'confirmed_at', } | AUDIT_COLUMNS, 'biz_inbound_item': {'inbound_item_id', 'inbound_id', 'pallet_id'} | CREATE_AUDIT_COLUMNS, 'biz_filling_record': { 'filling_id', 'filling_no', 'pallet_id', 'warehouse_id', 'strain_id', 'strain_code_snapshot', 'strain_name_snapshot', 'batch_no', 'package_spec', 'initial_packages', 'active_slot', 'closed_at', 'idempotency_key', 'request_hash', 'operator_id', 'operated_at', } | AUDIT_COLUMNS, 'biz_shipment_order': { 'shipment_id', 'shipment_no', 'source_warehouse_id', 'target_site_id', 'status', 'confirm_idempotency_key', 'confirm_request_hash', 'confirmed_by', 'confirmed_at', } | AUDIT_COLUMNS, 'biz_shipment_item': { 'shipment_item_id', 'shipment_id', 'pallet_id', 'active_filling_id', 'package_qty', } | CREATE_AUDIT_COLUMNS, 'biz_usage_record': { 'usage_id', 'usage_no', 'pallet_id', 'active_filling_id', 'site_id', 'line_id', 'used_packages', 'remaining_packages_after', 'idempotency_key', 'request_hash', 'operator_id', 'occurred_at', }, 'biz_return_order': { 'return_id', 'return_no', 'source_site_id', 'target_warehouse_id', 'status', 'request_idempotency_key', 'request_hash', 'confirm_idempotency_key', 'confirm_request_hash', 'requested_by', 'requested_at', 'confirmed_by', 'confirmed_at', } | AUDIT_COLUMNS, 'biz_return_item': { 'return_item_id', 'return_id', 'pallet_id', 'active_filling_id_snapshot', 'remaining_packages_snapshot', } | CREATE_AUDIT_COLUMNS, 'biz_pallet_inventory': { 'pallet_id', 'active_filling_id', 'warehouse_id', 'site_id', 'current_line_id', 'flow_state', 'remaining_packages', 'last_event_id', 'version', 'updated_at', }, 'biz_pallet_event': { 'event_id', 'pallet_id', 'event_seq', 'event_type', 'source_type', 'source_id', 'before_state', 'after_state', 'before_warehouse_id', 'before_site_id', 'before_line_id', 'after_warehouse_id', 'after_site_id', 'after_line_id', 'before_remaining_packages', 'after_remaining_packages', 'operator_id', 'occurred_at', 'idempotency_key', 'request_hash', }, } EXPECTED_PRIMARY_KEYS = { 'biz_strain': ('strain_id',), 'biz_warehouse': ('warehouse_id',), 'biz_customer': ('customer_id',), 'biz_customer_site': ('site_id',), 'biz_production_line': ('line_id',), 'biz_pallet': ('pallet_id',), 'biz_user_warehouse': ('user_id', 'warehouse_id'), 'biz_user_site': ('user_id', 'site_id'), 'biz_inbound_order': ('inbound_id',), 'biz_inbound_item': ('inbound_item_id',), 'biz_filling_record': ('filling_id',), 'biz_shipment_order': ('shipment_id',), 'biz_shipment_item': ('shipment_item_id',), 'biz_usage_record': ('usage_id',), 'biz_return_order': ('return_id',), 'biz_return_item': ('return_item_id',), 'biz_pallet_inventory': ('pallet_id',), 'biz_pallet_event': ('event_id',), } EXPECTED_UNIQUE_KEYS = { 'biz_strain': {('strain_code',)}, 'biz_warehouse': {('warehouse_code',)}, 'biz_customer': {('customer_code',)}, 'biz_customer_site': {('customer_id', 'site_code')}, 'biz_production_line': {('site_id', 'line_code')}, 'biz_pallet': {('pallet_code',), ('qr_token',)}, 'biz_user_warehouse': set(), 'biz_user_site': set(), 'biz_inbound_order': {('inbound_no',), ('confirm_idempotency_key',)}, 'biz_inbound_item': {('inbound_id', 'pallet_id')}, 'biz_filling_record': {('filling_no',), ('idempotency_key',), ('pallet_id', 'active_slot')}, 'biz_shipment_order': {('shipment_no',), ('confirm_idempotency_key',)}, 'biz_shipment_item': {('shipment_id', 'pallet_id')}, 'biz_usage_record': {('usage_no',), ('idempotency_key',)}, 'biz_return_order': {('return_no',), ('request_idempotency_key',), ('confirm_idempotency_key',)}, 'biz_return_item': {('return_id', 'pallet_id')}, 'biz_pallet_inventory': set(), 'biz_pallet_event': {('pallet_id', 'event_seq'), ('pallet_id', 'idempotency_key')}, } EXPECTED_FOREIGN_KEYS = { 'biz_strain': set(), 'biz_warehouse': set(), 'biz_customer': set(), 'biz_customer_site': {('customer_id', 'biz_customer.customer_id')}, 'biz_production_line': {('site_id', 'biz_customer_site.site_id')}, 'biz_pallet': set(), 'biz_user_warehouse': {('user_id', 'sys_user.user_id'), ('warehouse_id', 'biz_warehouse.warehouse_id')}, 'biz_user_site': {('user_id', 'sys_user.user_id'), ('site_id', 'biz_customer_site.site_id')}, 'biz_inbound_order': { ('warehouse_id', 'biz_warehouse.warehouse_id'), ('confirmed_by', 'sys_user.user_id'), }, 'biz_inbound_item': { ('inbound_id', 'biz_inbound_order.inbound_id'), ('pallet_id', 'biz_pallet.pallet_id'), }, 'biz_filling_record': { ('pallet_id', 'biz_pallet.pallet_id'), ('warehouse_id', 'biz_warehouse.warehouse_id'), ('strain_id', 'biz_strain.strain_id'), ('operator_id', 'sys_user.user_id'), }, 'biz_shipment_order': { ('source_warehouse_id', 'biz_warehouse.warehouse_id'), ('target_site_id', 'biz_customer_site.site_id'), ('confirmed_by', 'sys_user.user_id'), }, 'biz_shipment_item': { ('shipment_id', 'biz_shipment_order.shipment_id'), ('pallet_id', 'biz_pallet.pallet_id'), ('active_filling_id', 'biz_filling_record.filling_id'), }, 'biz_usage_record': { ('pallet_id', 'biz_pallet.pallet_id'), ('active_filling_id', 'biz_filling_record.filling_id'), ('site_id', 'biz_customer_site.site_id'), ('line_id', 'biz_production_line.line_id'), ('operator_id', 'sys_user.user_id'), }, 'biz_return_order': { ('source_site_id', 'biz_customer_site.site_id'), ('target_warehouse_id', 'biz_warehouse.warehouse_id'), ('requested_by', 'sys_user.user_id'), ('confirmed_by', 'sys_user.user_id'), }, 'biz_return_item': { ('return_id', 'biz_return_order.return_id'), ('pallet_id', 'biz_pallet.pallet_id'), ('active_filling_id_snapshot', 'biz_filling_record.filling_id'), }, 'biz_pallet_inventory': { ('pallet_id', 'biz_pallet.pallet_id'), ('active_filling_id', 'biz_filling_record.filling_id'), ('warehouse_id', 'biz_warehouse.warehouse_id'), ('site_id', 'biz_customer_site.site_id'), ('current_line_id', 'biz_production_line.line_id'), ('last_event_id', 'biz_pallet_event.event_id'), }, 'biz_pallet_event': { ('pallet_id', 'biz_pallet.pallet_id'), ('before_warehouse_id', 'biz_warehouse.warehouse_id'), ('before_site_id', 'biz_customer_site.site_id'), ('before_line_id', 'biz_production_line.line_id'), ('after_warehouse_id', 'biz_warehouse.warehouse_id'), ('after_site_id', 'biz_customer_site.site_id'), ('after_line_id', 'biz_production_line.line_id'), ('operator_id', 'sys_user.user_id'), }, } EXPECTED_CHECKS = { 'biz_strain': {"statusin('0','1')"}, 'biz_warehouse': {"statusin('0','1')"}, 'biz_customer': {"statusin('0','1')"}, 'biz_customer_site': {"statusin('0','1')"}, 'biz_production_line': {"statusin('0','1')"}, 'biz_pallet': {'capacity_packages>0', "asset_statusin('0','1')"}, 'biz_user_warehouse': set(), 'biz_user_site': set(), 'biz_inbound_order': {"statusin('draft','confirmed')"}, 'biz_inbound_item': set(), 'biz_filling_record': {'initial_packages>0', "active_slotisnulloractive_slot='active'"}, 'biz_shipment_order': {"statusin('draft','confirmed')"}, 'biz_shipment_item': {'package_qtyisnullorpackage_qty>0'}, 'biz_usage_record': {'used_packages>0', 'remaining_packages_after>=0'}, 'biz_return_order': {"statusin('requested','confirmed')"}, 'biz_return_item': {'remaining_packages_snapshot>=0'}, 'biz_pallet_inventory': { "flow_statein('unstocked','empty_in_warehouse','filled_in_warehouse','at_site_unused','at_site_partial'," "'at_site_empty','returning')", 'remaining_packages>=0', 'version>=0', 'not(warehouse_idisnotnullandsite_idisnotnull)', }, 'biz_pallet_event': { "event_typein('inbound_confirmed','filling_confirmed','shipment_confirmed','usage_recorded'," "'return_requested','return_confirmed')" }, } EXPECTED_PAGES = { 502: ('菌种资料', 121, 1, 'strain', 'master/strain/index', 'master:strain:list'), 503: ('客户资料', 121, 2, 'customer', 'master/customer/index', 'master:customer:list'), 504: ('客户现场', 121, 3, 'site', 'master/site/index', 'master:site:list'), 505: ('产线资料', 121, 4, 'line', 'master/line/index', 'master:line:list'), 506: ('仓库资料', 121, 5, 'warehouse', 'master/warehouse/index', 'master:warehouse:list'), 507: ('托盘资料', 121, 6, 'pallet', 'master/pallet/index', 'master:pallet:list'), 508: ('仓库入库', 122, 1, 'inbound', 'business/inbound/index', 'business:inbound:list'), 509: ('工厂装填', 122, 2, 'filling', 'business/filling/index', 'business:filling:list'), 510: ('发货出库', 122, 3, 'shipment', 'business/shipment/index', 'business:shipment:list'), 511: ('客户现场使用', 122, 4, 'usage', 'business/usage/index', 'business:usage:list'), 512: ('退回仓库', 122, 5, 'return', 'business/return/index', 'business:return:list'), 513: ( '托盘库存', 123, 1, 'palletInventory', 'report/palletInventory/index', 'report:palletinventory:list', ), 514: ('装填记录', 123, 2, 'reportFilling', 'report/filling/index', 'report:filling:list'), 515: ('发货出库', 123, 3, 'reportShipment', 'report/shipment/index', 'report:shipment:list'), 516: ('产线消耗', 123, 4, 'consumption', 'report/consumption/index', 'report:consumption:list'), 517: ('退回仓库', 123, 5, 'reportReturn', 'report/return/index', 'report:return:list'), 518: ('托盘追溯', 123, 6, 'palletTrace', 'report/palletTrace/index', 'report:pallettrace:list'), } BUTTON_SPECS = """ 1065 502 1 菌种查询 master:strain:query 1066 502 2 菌种新增 master:strain:add 1067 502 3 菌种修改 master:strain:edit 1068 502 4 菌种删除 master:strain:remove 1069 503 1 客户查询 master:customer:query 1070 503 2 客户新增 master:customer:add 1071 503 3 客户修改 master:customer:edit 1072 503 4 客户删除 master:customer:remove 1073 504 1 现场查询 master:site:query 1074 504 2 现场新增 master:site:add 1075 504 3 现场修改 master:site:edit 1076 504 4 现场删除 master:site:remove 1077 504 5 分配用户 master:site:assign 1078 505 1 产线查询 master:line:query 1079 505 2 产线新增 master:line:add 1080 505 3 产线修改 master:line:edit 1081 505 4 产线删除 master:line:remove 1082 506 1 仓库查询 master:warehouse:query 1083 506 2 仓库新增 master:warehouse:add 1084 506 3 仓库修改 master:warehouse:edit 1085 506 4 仓库删除 master:warehouse:remove 1086 506 5 分配用户 master:warehouse:assign 1087 507 1 托盘查询 master:pallet:query 1088 507 2 托盘新增 master:pallet:add 1089 507 3 托盘修改 master:pallet:edit 1090 507 4 托盘删除 master:pallet:remove 1091 507 5 托盘打印 master:pallet:print 1092 508 1 入库查询 business:inbound:query 1093 508 2 入库新增 business:inbound:add 1094 508 3 入库修改 business:inbound:edit 1095 508 4 入库删除 business:inbound:remove 1096 508 5 入库确认 business:inbound:confirm 1097 509 1 装填查询 business:filling:query 1098 509 2 装填新增 business:filling:add 1099 510 1 发货查询 business:shipment:query 1100 510 2 发货新增 business:shipment:add 1101 510 3 发货修改 business:shipment:edit 1102 510 4 发货删除 business:shipment:remove 1103 510 5 发货确认 business:shipment:confirm 1104 511 1 使用查询 business:usage:query 1105 511 2 使用新增 business:usage:add 1106 512 1 退回查询 business:return:query 1107 512 2 退回申请 business:return:request 1108 512 3 退回确认 business:return:confirm """ EXPECTED_BUTTONS = { int(menu_id): (name, int(parent_id), int(order_num), permission) for menu_id, parent_id, order_num, name, permission in ( line.split() for line in BUTTON_SPECS.splitlines() if line.strip() ) } BUSINESS_MENU_IDS = set(range(121, 124)) | set(range(502, 519)) | set(range(1065, 1109)) def _strip_sql_comments(sql: str) -> str: sql = re.sub(r'/\*.*?\*/', ' ', sql, flags=re.S) return re.sub(r'(?m)^\s*(?:--|#).*$\n?', ' ', sql) def _matching_parenthesis(text: str, opening: int) -> int: depth = 0 quote = None index = opening while index < len(text): char = text[index] if quote: if char == quote and index + 1 < len(text) and text[index + 1] == quote: index += 2 continue if char == quote: quote = None elif char in '\'"`': quote = char elif char == '(': depth += 1 elif char == ')': depth -= 1 if depth == 0: return index index += 1 raise AssertionError('unbalanced SQL parentheses') def _split_values(values: str) -> list[str]: parts = [] start = 0 depth = 0 quote = None index = 0 while index < len(values): char = values[index] if quote: if char == quote and index + 1 < len(values) and values[index + 1] == quote: index += 2 continue if char == quote: quote = None elif char in '\'"`': quote = char elif char == '(': depth += 1 elif char == ')': depth -= 1 elif char == ',' and depth == 0: parts.append(values[start:index].strip()) start = index + 1 index += 1 parts.append(values[start:].strip()) return parts def _decode_value(value: str) -> str | None: value = value.strip() if value.lower() == 'null': return None if value.startswith("'") and value.endswith("'"): return value[1:-1].replace("''", "'") return value def _sql_statements(sql: str) -> list[str]: return [statement.strip() for statement in _strip_sql_comments(sql).split(';') if statement.strip()] def _insert_rows(sql: str, table: str) -> list[list[str | None]]: rows = [] pattern = re.compile( rf'\binsert(?:\s+ignore)?\s+into\s+(?:[`"]?[a-z0-9_]+[`"]?\.)?[`"]?{table}[`"]?' r'(?:\s*\([^;]*?\))?\s+values\s*', re.I | re.S, ) for statement in _sql_statements(sql): match = pattern.search(statement) if not match: continue index = match.end() while index < len(statement): while index < len(statement) and (statement[index].isspace() or statement[index] == ','): index += 1 if index >= len(statement) or statement[index] != '(': break closing = _matching_parenthesis(statement, index) rows.append([_decode_value(value) for value in _split_values(statement[index + 1 : closing])]) index = closing + 1 return rows def _create_table_bodies(sql: str) -> dict[str, str]: clean_sql = _strip_sql_comments(sql) pattern = re.compile( r'\bcreate\s+table\s+(?:if\s+not\s+exists\s+)?' r'(?:[`"]?[a-z0-9_]+[`"]?\.)?[`"]?([a-z_][a-z0-9_]*)[`"]?\s*\(', re.I, ) tables = {} for match in pattern.finditer(clean_sql): opening = match.end() - 1 closing = _matching_parenthesis(clean_sql, opening) tables[match.group(1).lower()] = clean_sql[opening + 1 : closing] return tables def _identifiers(value: str) -> tuple[str, ...]: return tuple(part.strip().strip('`"').lower() for part in value.split(',')) def _normalized_check(value: str) -> str: normalized = re.sub(r'[\s`"]+', '', value).lower() while normalized.startswith('(') and normalized.endswith(')'): try: if _matching_parenthesis(normalized, 0) != len(normalized) - 1: break except AssertionError: break normalized = normalized[1:-1] return normalized def _table_contract( body: str, ) -> tuple[set[str], tuple[str, ...], set[tuple[str, ...]], set[tuple[str, str]], set[str]]: columns = set() primary_key = () unique_keys = set() foreign_keys = set() checks = set() reserved = {'constraint', 'primary', 'unique', 'foreign', 'check', 'key', 'index'} for clause in _split_values(body): clean = clause.strip() identifier_match = re.match(r'[`"]?([a-z_][a-z0-9_]*)[`"]?', clean, flags=re.I) first = identifier_match.group(1).lower() if identifier_match else '' is_column = first and first not in reserved if is_column: columns.add(first) if re.search(r'\bprimary\s+key\b', clean, flags=re.I): primary_key = (first,) if re.search(r'\bunique\b', clean, flags=re.I): unique_keys.add((first,)) reference = re.search( r'\breferences\s+[`"]?([a-z_][a-z0-9_]*)[`"]?\s*\(\s*[`"]?([a-z_][a-z0-9_]*)', clean, flags=re.I, ) if reference: foreign_keys.add((first, f'{reference.group(1).lower()}.{reference.group(2).lower()}')) primary_match = re.search(r'\bprimary\s+key\s*\(', clean, flags=re.I) if primary_match: opening = primary_match.end() - 1 primary_key = _identifiers(clean[opening + 1 : _matching_parenthesis(clean, opening)]) unique_match = re.search( r'\bunique(?:\s+(?:key|index))?(?:\s+[`"]?[a-z_][a-z0-9_]*[`"]?)?\s*\(', clean, flags=re.I, ) if unique_match: opening = unique_match.end() - 1 unique_keys.add(_identifiers(clean[opening + 1 : _matching_parenthesis(clean, opening)])) foreign_match = re.search(r'\bforeign\s+key\s*\(', clean, flags=re.I) if foreign_match: opening = foreign_match.end() - 1 local_columns = _identifiers(clean[opening + 1 : _matching_parenthesis(clean, opening)]) reference = re.search( r'\breferences\s+[`"]?([a-z_][a-z0-9_]*)[`"]?\s*\((.*?)\)', clean, flags=re.I | re.S, ) if reference: target_columns = _identifiers(reference.group(2)) for local_column, target_column in zip(local_columns, target_columns, strict=False): foreign_keys.add((local_column, f'{reference.group(1).lower()}.{target_column}')) check_match = re.search(r'\bcheck\s*\(', clean, flags=re.I) if check_match: opening = check_match.end() - 1 checks.add(_normalized_check(clean[opening + 1 : _matching_parenthesis(clean, opening)])) return columns, primary_key, unique_keys, foreign_keys, checks def _business_menu_rows(sql: str) -> dict[int, list]: rows = _insert_rows(sql, 'sys_menu') selected = [row for row in rows if row and str(row[0]).isdigit() and int(row[0]) in BUSINESS_MENU_IDS] counts = {menu_id: sum(int(row[0]) == menu_id for row in selected) for menu_id in BUSINESS_MENU_IDS} if set(counts) != BUSINESS_MENU_IDS or any(count != 1 for count in counts.values()): raise AssertionError(f'business menu row counts differ: {counts}') return {int(row[0]): row for row in selected} class SqlContractTest(unittest.TestCase): def test_initializers_define_exact_business_table_set(self) -> None: for filename in INITIALIZERS: with self.subTest(filename=filename): sql = (SQL_ROOT / filename).read_text(encoding='utf-8') actual = {table for table in _create_table_bodies(sql) if table.startswith('biz_')} self.assertEqual(EXPECTED_TABLES, actual) def test_initializers_match_orm_table_contract(self) -> None: for filename in INITIALIZERS: tables = _create_table_bodies((SQL_ROOT / filename).read_text(encoding='utf-8')) for table in sorted(EXPECTED_TABLES): with self.subTest(filename=filename, table=table): self.assertIn(table, tables) columns, primary_key, unique_keys, foreign_keys, checks = _table_contract(tables[table]) self.assertEqual(EXPECTED_COLUMNS[table], columns) self.assertEqual(EXPECTED_PRIMARY_KEYS[table], primary_key) self.assertEqual(EXPECTED_UNIQUE_KEYS[table], unique_keys) self.assertEqual(EXPECTED_FOREIGN_KEYS[table], foreign_keys) self.assertEqual(EXPECTED_CHECKS[table], checks) def test_initializers_define_fixed_business_menus_and_permissions(self) -> None: expected_permissions = {page[-1] for page in EXPECTED_PAGES.values()} | { button[-1] for button in EXPECTED_BUTTONS.values() } self.assertEqual(61, len(expected_permissions)) for filename in INITIALIZERS: with self.subTest(filename=filename): rows = _business_menu_rows((SQL_ROOT / filename).read_text(encoding='utf-8')) self.assertTrue(all(len(row) == SYS_MENU_VALUE_COUNT for row in rows.values())) self.assertEqual( ('基础资料维护', '1', '10', 'businessMaster', 'M', ''), tuple(rows[121][index] for index in (1, 2, 3, 4, 10, 13)), ) self.assertEqual( ('业务管理', '1', '11', 'businessOperation', 'M', ''), tuple(rows[122][index] for index in (1, 2, 3, 4, 10, 13)), ) self.assertEqual( ('报表查询', '1', '12', 'businessReport', 'M', ''), tuple(rows[123][index] for index in (1, 2, 3, 4, 10, 13)), ) for menu_id in (121, 122, 123): self.assertEqual( (None, '', '', '1', '0', '0', '0'), tuple(rows[menu_id][i] for i in (5, 6, 7, 8, 9, 11, 12)) ) for menu_id, expected in EXPECTED_PAGES.items(): row = rows[menu_id] actual = (row[1], int(row[2]), int(row[3]), row[4], row[5], row[13]) self.assertEqual(expected, actual, menu_id) self.assertEqual( ('', '', '1', '0', 'C', '0', '0'), tuple(row[index] for index in (6, 7, 8, 9, 10, 11, 12)) ) for menu_id, expected in EXPECTED_BUTTONS.items(): row = rows[menu_id] actual = (row[1], int(row[2]), int(row[3]), row[13]) self.assertEqual(expected, actual, menu_id) self.assertEqual( ('', '', '', '', '1', '0', 'F', '0', '0'), tuple(row[index] for index in (4, 5, 6, 7, 8, 9, 10, 11, 12)), ) all_menu_rows = _insert_rows((SQL_ROOT / filename).read_text(encoding='utf-8'), 'sys_menu') actual_permissions = { row[13] for row in all_menu_rows if len(row) == SYS_MENU_VALUE_COUNT and row[13] and re.match(r'^(?:master|business|report):', row[13]) } self.assertEqual(expected_permissions, actual_permissions) 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) mysql_contract = { menu_id: tuple(row[index] for index in compared_indexes) for menu_id, row in mysql_rows.items() } postgres_contract = { menu_id: tuple(row[index] for index in compared_indexes) for menu_id, row in postgres_rows.items() } self.assertEqual(mysql_contract, postgres_contract) def test_initializers_do_not_grant_new_business_menus_to_roles(self) -> None: for filename in INITIALIZERS: with self.subTest(filename=filename): rows = _insert_rows((SQL_ROOT / filename).read_text(encoding='utf-8'), 'sys_role_menu') granted = {int(row[1]) for row in rows if len(row) > 1 and str(row[1]).isdigit()} self.assertFalse(BUSINESS_MENU_IDS & granted) self.assertFalse({4, 99, 118, 119, 1061, 1062, 1063, 1064} & granted) def test_initializers_exclude_removed_ai_and_external_link_seeds(self) -> None: removed_menu_ids = {4, 99, 118, 119, 1061, 1062, 1063, 1064} for filename in INITIALIZERS: with self.subTest(filename=filename): sql = (SQL_ROOT / filename).read_text(encoding='utf-8') menu_ids = { int(row[0]) for row in _insert_rows(sql, 'sys_menu') if row and row[0] is not None and str(row[0]).isdigit() } self.assertFalse(removed_menu_ids & menu_ids) dict_types = _insert_rows(sql, 'sys_dict_type') self.assertFalse(any(row[0] == '12' or row[2] == 'ai_provider_type' for row in dict_types)) dict_data = _insert_rows(sql, 'sys_dict_data') self.assertFalse( any(int(row[0]) in REMOVED_AI_DICT_CODES or row[4] == 'ai_provider_type' for row in dict_data) ) def test_initializers_keep_menu_allocator_and_use_native_dialects(self) -> None: mysql = (SQL_ROOT / 'shierfeng-fastapi.sql').read_text(encoding='utf-8').lower() postgres = (SQL_ROOT / 'shierfeng-fastapi-pg.sql').read_text(encoding='utf-8').lower() self.assertRegex(mysql, r'auto_increment\s*=\s*2000') self.assertNotRegex(mysql, r'\bbigserial\b|\balter\s+sequence\b|\bcomment\s+on\b') self.assertRegex(postgres, r'alter\s+sequence\s+sys_menu_menu_id_seq\s+restart\s+2000') self.assertNotIn('`', postgres) self.assertNotRegex(postgres, r'\bauto_increment\b|\bengine\s*=') def test_initializers_create_event_before_inventory_foreign_key(self) -> None: for filename in INITIALIZERS: with self.subTest(filename=filename): sql = _strip_sql_comments((SQL_ROOT / filename).read_text(encoding='utf-8')).lower() self.assertIn('create table biz_pallet_event', sql) self.assertIn('create table biz_pallet_inventory', sql) self.assertLess( sql.index('create table biz_pallet_event'), sql.index('create table biz_pallet_inventory') ) def test_initializers_drop_business_tables_before_recreating_referenced_system_tables(self) -> None: reverse_dependency_order = ( 'biz_pallet_inventory', 'biz_pallet_event', 'biz_return_item', 'biz_return_order', 'biz_usage_record', 'biz_shipment_item', 'biz_shipment_order', 'biz_filling_record', 'biz_inbound_item', 'biz_inbound_order', 'biz_user_site', 'biz_user_warehouse', 'biz_pallet', 'biz_production_line', 'biz_customer_site', 'biz_customer', 'biz_warehouse', 'biz_strain', ) for filename in INITIALIZERS: with self.subTest(filename=filename): sql = _strip_sql_comments((SQL_ROOT / filename).read_text(encoding='utf-8')).lower() drop_positions = [sql.index(f'drop table if exists {table};') for table in reverse_dependency_order] self.assertEqual(sorted(drop_positions), drop_positions) self.assertLess(drop_positions[-1], sql.index('drop table if exists sys_user')) 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() sql = re.sub(r'/\*.*?\*/', '', raw, flags=re.S) sql = re.sub(r'(?m)^\s*(?:--|#).*$\n?', '', sql).replace('`', '') for table in ( 'ai_models', 'ai_chat_config', 'ai_sessions', 'ai_memories', 'ai_metrics', 'ai_eval_runs', 'ai_knowledge', 'ai_culture', 'ai_traces', 'ai_spans', 'ai_schema_versions', ): qualified = rf'(?:[a-z0-9_]+\.)?{table}' patterns = ( rf'\bdrop\s+(?:temporary\s+)?table\s+(?:if\s+exists\s+)?{qualified}\b', rf'\btruncate\s+(?:table\s+)?{qualified}\b', rf'\bdelete\s+from\s+{qualified}\b', rf'\balter\s+table\s+(?:if\s+exists\s+)?{qualified}\b', ) for pattern in patterns: self.assertIsNone(re.search(pattern, sql), f'dangerous AI table mutation: {table}') self.assertNotRegex(sql, r'\b(?:drop\s+(?:temporary\s+)?table|truncate\s+(?:table\s+)?|delete\s+from)\s+biz_') def test_patch_is_documented_idempotent_and_contains_full_contract(self) -> None: self.assertTrue(PATCH_PATH.is_file(), f'missing patch: {PATCH_PATH}') raw = PATCH_PATH.read_text(encoding='utf-8') lower = raw.lower() self.assertIn('隐式提交', raw) self.assertIn('备份', raw) self.assertIn('人工审查', raw) self.assertIn('未自动执行', raw) self.assertLess(lower.index('ddl'), lower.index('dml')) self.assertEqual(18, len(re.findall(r'create\s+table\s+if\s+not\s+exists\s+`?biz_', lower))) tables = _create_table_bodies(raw) self.assertEqual(EXPECTED_TABLES, {table for table in tables if table.startswith('biz_')}) for table in EXPECTED_TABLES: with self.subTest(table=table): columns, primary_key, unique_keys, foreign_keys, checks = _table_contract(tables[table]) self.assertEqual(EXPECTED_COLUMNS[table], columns) self.assertEqual(EXPECTED_PRIMARY_KEYS[table], primary_key) self.assertEqual(EXPECTED_UNIQUE_KEYS[table], unique_keys) self.assertEqual(EXPECTED_FOREIGN_KEYS[table], foreign_keys) 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) 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.assertNotRegex(_strip_sql_comments(lower), r'\binsert(?:\s+ignore)?\s+into\s+`?sys_role_menu`?\b') self.assertIn('/getRouters', raw) self.assertIn('用户信息', raw) self.assertIn('sys_dict:ai_provider_type', raw) executable = _strip_sql_comments(raw) self.assertNotIn('/getRouters', executable) 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_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() role_cleanup = re.search(r'delete\s+from\s+sys_role_menu\s+where\s+menu_id\s+in\s*\([^;]+\)', sql) ai_menu_cleanup = re.search(r'delete\s+from\s+sys_menu\s+where\s+menu_id\s+in\s*\([^;]+\)', sql) dict_data_cleanup = re.search(r'delete\s+from\s+sys_dict_data\s+where\s+[^;]+', sql) dict_type_cleanup = re.search(r'delete\s+from\s+sys_dict_type\s+where\s+[^;]+', sql) external_link_cleanup = re.search(r'delete\s+from\s+sys_menu\s+where\s+menu_id\s*=\s*99\b', sql) for cleanup in ( role_cleanup, ai_menu_cleanup, dict_data_cleanup, dict_type_cleanup, external_link_cleanup, ): self.assertIsNotNone(cleanup) self.assertRegex(role_cleanup.group(), r'\b4\b.*\b99\b.*\b118\b.*\b119\b.*\b1061\b.*\b1064\b') self.assertRegex(ai_menu_cleanup.group(), r'\b4\b.*\b118\b.*\b119\b.*\b1061\b.*\b1064\b') self.assertRegex(dict_data_cleanup.group(), r'dict_code\s+between\s+33\s+and\s+69') self.assertIn('ai_provider_type', dict_data_cleanup.group()) self.assertRegex(dict_type_cleanup.group(), r'dict_id\s*=\s*12') self.assertIn('ai_provider_type', dict_type_cleanup.group()) self.assertLess(role_cleanup.start(), ai_menu_cleanup.start()) self.assertLess(ai_menu_cleanup.start(), dict_data_cleanup.start()) self.assertLess(dict_data_cleanup.start(), dict_type_cleanup.start()) self.assertLess(dict_type_cleanup.start(), external_link_cleanup.start()) self.assertNotRegex(sql, r'\bdelete\s+from\s+(?:[a-z0-9_]+\.)?(?:biz|ai)_[a-z0-9_]+\b') if __name__ == '__main__': unittest.main()