2645 lines
116 KiB
Python
2645 lines
116 KiB
Python
import re
|
||
import unittest
|
||
from collections import Counter
|
||
from collections.abc import Callable
|
||
from pathlib import Path
|
||
|
||
SQL_ROOT = Path(__file__).resolve().parents[2] / 'sql'
|
||
INITIALIZERS = ('shierfeng-fastapi.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'
|
||
TEST_BATCH_MYSQL_PATCH_PATH = SQL_ROOT / 'patches/20260922_test_data_batch_mysql.sql'
|
||
SHIPMENT_RECEIPT_PATCH_PATH = SQL_ROOT / 'patches/20261008_shipment_receipt_mysql.sql'
|
||
MASTER_EXCEL_PATCH_PATH = SQL_ROOT / 'patches/20261009_master_excel_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_TEST_BATCH_TABLES = {'biz_test_batch', 'biz_test_batch_user'}
|
||
TEST_BATCH_OWNED_TABLES = {
|
||
'biz_strain',
|
||
'biz_customer',
|
||
'biz_customer_site',
|
||
'biz_production_line',
|
||
'biz_warehouse',
|
||
'biz_pallet',
|
||
'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',
|
||
}
|
||
|
||
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_DATETIME_COLUMNS = {
|
||
'biz_strain': {'create_time', 'update_time'},
|
||
'biz_warehouse': {'create_time', 'update_time'},
|
||
'biz_customer': {'create_time', 'update_time'},
|
||
'biz_customer_site': {'create_time', 'update_time'},
|
||
'biz_production_line': {'create_time', 'update_time'},
|
||
'biz_pallet': {'create_time', 'update_time'},
|
||
'biz_user_warehouse': {'create_time'},
|
||
'biz_user_site': {'create_time'},
|
||
'biz_inbound_order': {'confirmed_at', 'create_time', 'update_time'},
|
||
'biz_inbound_item': {'create_time'},
|
||
'biz_filling_record': {'closed_at', 'operated_at', 'create_time', 'update_time'},
|
||
'biz_shipment_order': {'confirmed_at', 'create_time', 'update_time'},
|
||
'biz_shipment_item': {'create_time'},
|
||
'biz_usage_record': {'occurred_at'},
|
||
'biz_return_order': {'requested_at', 'confirmed_at', 'create_time', 'update_time'},
|
||
'biz_return_item': {'create_time'},
|
||
'biz_pallet_inventory': {'updated_at'},
|
||
'biz_pallet_event': {'occurred_at'},
|
||
}
|
||
|
||
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',), ('warehouse_id', 'confirm_idempotency_key')},
|
||
'biz_inbound_item': {('inbound_id', 'pallet_id')},
|
||
'biz_filling_record': {
|
||
('filling_no',),
|
||
('warehouse_id', 'idempotency_key'),
|
||
('pallet_id', 'active_slot'),
|
||
},
|
||
'biz_shipment_order': {
|
||
('shipment_no',),
|
||
('source_warehouse_id', 'target_site_id', 'confirm_idempotency_key'),
|
||
},
|
||
'biz_shipment_item': {('shipment_id', 'pallet_id')},
|
||
'biz_usage_record': {('usage_no',), ('site_id', 'idempotency_key')},
|
||
'biz_return_order': {
|
||
('return_no',),
|
||
('source_site_id', 'target_warehouse_id', 'request_idempotency_key'),
|
||
('source_site_id', 'target_warehouse_id', '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')"
|
||
},
|
||
}
|
||
|
||
LEGACY_EXPECTED_TABLES = set(EXPECTED_TABLES)
|
||
LEGACY_EXPECTED_COLUMNS = {table: set(columns) for table, columns in EXPECTED_COLUMNS.items()}
|
||
LEGACY_EXPECTED_PRIMARY_KEYS = dict(EXPECTED_PRIMARY_KEYS)
|
||
LEGACY_EXPECTED_UNIQUE_KEYS = {table: set(keys) for table, keys in EXPECTED_UNIQUE_KEYS.items()}
|
||
LEGACY_EXPECTED_FOREIGN_KEYS = {table: set(keys) for table, keys in EXPECTED_FOREIGN_KEYS.items()}
|
||
LEGACY_EXPECTED_CHECKS = {table: set(checks) for table, checks in EXPECTED_CHECKS.items()}
|
||
|
||
EXPECTED_COLUMNS['biz_filling_record'].update({'production_date', 'expiry_date'})
|
||
EXPECTED_CHECKS['biz_filling_record'].add(
|
||
'(production_dateisnullandexpiry_dateisnull)or(production_dateisnotnullandexpiry_dateisnotnullandexpiry_date>=production_date)'
|
||
)
|
||
|
||
CURRENT_TEST_BATCH_OWNED_TABLES = TEST_BATCH_OWNED_TABLES | {'biz_tunnel'}
|
||
EXPECTED_TABLES.add('biz_tunnel')
|
||
EXPECTED_COLUMNS.update(
|
||
{
|
||
'biz_tunnel': {
|
||
'tunnel_id',
|
||
'site_id',
|
||
'tunnel_code',
|
||
'tunnel_name',
|
||
'current_trace_no',
|
||
'current_trace_started_at',
|
||
'status',
|
||
}
|
||
| AUDIT_COLUMNS,
|
||
'biz_usage_record': EXPECTED_COLUMNS['biz_usage_record']
|
||
| {'tunnel_id', 'trace_no', 'batch_no_snapshot'},
|
||
'biz_return_item': EXPECTED_COLUMNS['biz_return_item']
|
||
| {'tunnel_id_snapshot', 'trace_no_snapshot', 'batch_no_snapshot'},
|
||
'biz_pallet_inventory': EXPECTED_COLUMNS['biz_pallet_inventory'] | {'tunnel_id', 'trace_no'},
|
||
'biz_pallet_event': EXPECTED_COLUMNS['biz_pallet_event']
|
||
| {'before_tunnel_id', 'after_tunnel_id', 'trace_no', 'batch_no_snapshot'},
|
||
}
|
||
)
|
||
EXPECTED_DATETIME_COLUMNS['biz_tunnel'] = {'current_trace_started_at', 'create_time', 'update_time'}
|
||
EXPECTED_PRIMARY_KEYS['biz_tunnel'] = ('tunnel_id',)
|
||
EXPECTED_UNIQUE_KEYS['biz_tunnel'] = {('site_id', 'tunnel_code')}
|
||
EXPECTED_FOREIGN_KEYS.update(
|
||
{
|
||
'biz_tunnel': {('site_id', 'biz_customer_site.site_id')},
|
||
'biz_usage_record': EXPECTED_FOREIGN_KEYS['biz_usage_record']
|
||
| {('tunnel_id', 'biz_tunnel.tunnel_id')},
|
||
'biz_return_item': EXPECTED_FOREIGN_KEYS['biz_return_item']
|
||
| {('tunnel_id_snapshot', 'biz_tunnel.tunnel_id')},
|
||
'biz_pallet_inventory': EXPECTED_FOREIGN_KEYS['biz_pallet_inventory']
|
||
| {('tunnel_id', 'biz_tunnel.tunnel_id')},
|
||
'biz_pallet_event': EXPECTED_FOREIGN_KEYS['biz_pallet_event']
|
||
| {
|
||
('before_tunnel_id', 'biz_tunnel.tunnel_id'),
|
||
('after_tunnel_id', 'biz_tunnel.tunnel_id'),
|
||
},
|
||
}
|
||
)
|
||
EXPECTED_CHECKS['biz_tunnel'] = {"statusin('0','1')"}
|
||
EXPECTED_CHECKS['biz_pallet_event'] = {
|
||
"event_typein('inbound_confirmed','filling_confirmed','shipment_confirmed','shipment_received','tunnel_entered','usage_recorded',"
|
||
"'return_requested','return_confirmed')"
|
||
}
|
||
|
||
# 新契约必须位于 LEGACY_EXPECTED_* 的独立副本之后，旧补丁保持原始结构。
|
||
for fifo_table in ('biz_usage_record', 'biz_pallet_event'):
|
||
EXPECTED_COLUMNS[fifo_table].update({'fifo_skip_reason', 'fifo_recommendation'})
|
||
|
||
EXPECTED_COLUMNS['biz_shipment_order'].update({
|
||
'receipt_state', 'received_by', 'received_at', 'receipt_idempotency_key', 'receipt_request_hash',
|
||
})
|
||
EXPECTED_DATETIME_COLUMNS['biz_shipment_order'].add('received_at')
|
||
EXPECTED_UNIQUE_KEYS['biz_shipment_order'].add(('target_site_id', 'receipt_idempotency_key'))
|
||
EXPECTED_FOREIGN_KEYS['biz_shipment_order'].add(('received_by', 'sys_user.user_id'))
|
||
EXPECTED_CHECKS['biz_shipment_order'].add("receipt_statein('pending','received','legacy')")
|
||
EXPECTED_COLUMNS['biz_pallet_inventory'].add('site_arrived_at')
|
||
EXPECTED_DATETIME_COLUMNS['biz_pallet_inventory'].add('site_arrived_at')
|
||
EXPECTED_CHECKS['biz_pallet_inventory'] = {
|
||
check.replace("'filled_in_warehouse','at_site_unused'", "'filled_in_warehouse','in_transit','at_site_unused'")
|
||
for check in EXPECTED_CHECKS['biz_pallet_inventory']
|
||
}
|
||
|
||
EXPECTED_TABLES |= EXPECTED_TEST_BATCH_TABLES
|
||
for table_name in CURRENT_TEST_BATCH_OWNED_TABLES:
|
||
EXPECTED_COLUMNS[table_name].add('test_batch_id')
|
||
EXPECTED_FOREIGN_KEYS[table_name].add(('test_batch_id', 'biz_test_batch.batch_id'))
|
||
|
||
EXPECTED_COLUMNS.update(
|
||
{
|
||
'biz_test_batch': {
|
||
'batch_id',
|
||
'batch_code',
|
||
'batch_name',
|
||
'status',
|
||
'create_by',
|
||
'create_time',
|
||
'update_by',
|
||
'update_time',
|
||
'closed_by',
|
||
'closed_at',
|
||
'purged_by',
|
||
'purged_at',
|
||
'purge_summary',
|
||
'preview_action',
|
||
'preview_token_hash',
|
||
'preview_scope_hash',
|
||
'preview_expires_at',
|
||
'remark',
|
||
},
|
||
'biz_test_batch_user': {'batch_id', 'user_id', 'create_by', 'create_time'},
|
||
}
|
||
)
|
||
EXPECTED_DATETIME_COLUMNS.update(
|
||
{
|
||
'biz_test_batch': {'create_time', 'update_time', 'closed_at', 'purged_at', 'preview_expires_at'},
|
||
'biz_test_batch_user': {'create_time'},
|
||
}
|
||
)
|
||
EXPECTED_PRIMARY_KEYS.update(
|
||
{
|
||
'biz_test_batch': ('batch_id',),
|
||
'biz_test_batch_user': ('batch_id', 'user_id'),
|
||
}
|
||
)
|
||
EXPECTED_UNIQUE_KEYS.update(
|
||
{
|
||
'biz_test_batch': {('batch_code',)},
|
||
'biz_test_batch_user': set(),
|
||
}
|
||
)
|
||
EXPECTED_FOREIGN_KEYS.update(
|
||
{
|
||
'biz_test_batch': {
|
||
('closed_by', 'sys_user.user_id'),
|
||
('purged_by', 'sys_user.user_id'),
|
||
},
|
||
'biz_test_batch_user': {
|
||
('batch_id', 'biz_test_batch.batch_id'),
|
||
('user_id', 'sys_user.user_id'),
|
||
},
|
||
}
|
||
)
|
||
EXPECTED_CHECKS.update(
|
||
{
|
||
'biz_test_batch': {
|
||
"statusin('active','closed','purged')",
|
||
"preview_actionisnullorpreview_actionin('adopt','purge')",
|
||
},
|
||
'biz_test_batch_user': set(),
|
||
}
|
||
)
|
||
|
||
EXPECTED_PAGES = {
|
||
519: ('测试数据批次', 1, 7, 'testBatch', 'system/testBatch/index', 'system:testbatch:list'),
|
||
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'),
|
||
520: ('隧道资料', 121, 7, 'tunnel', 'master/tunnel/index', 'master:tunnel: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
|
||
1069 503 1 客户查询 master:customer:query
|
||
1070 503 2 客户新增 master:customer:add
|
||
1071 503 3 客户修改 master:customer:edit
|
||
1073 504 1 现场查询 master:site:query
|
||
1074 504 2 现场新增 master:site:add
|
||
1075 504 3 现场修改 master:site:edit
|
||
1077 504 5 分配用户 master:site:assign
|
||
1078 505 1 产线查询 master:line:query
|
||
1079 505 2 产线新增 master:line:add
|
||
1080 505 3 产线修改 master:line:edit
|
||
1082 506 1 仓库查询 master:warehouse:query
|
||
1083 506 2 仓库新增 master:warehouse:add
|
||
1084 506 3 仓库修改 master:warehouse:edit
|
||
1086 506 5 分配用户 master:warehouse:assign
|
||
1087 507 1 托盘查询 master:pallet:query
|
||
1088 507 2 托盘新增 master:pallet:add
|
||
1089 507 3 托盘修改 master:pallet:edit
|
||
1091 507 5 托盘打印 master:pallet:print
|
||
1116 520 1 隧道查询 master:tunnel:query
|
||
1117 520 2 隧道新增 master:tunnel:add
|
||
1118 520 3 隧道修改 master:tunnel:edit
|
||
1092 508 1 入库查询 business:inbound:query
|
||
1093 508 2 入库新增 business:inbound:add
|
||
1094 508 3 入库修改 business:inbound:edit
|
||
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
|
||
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
|
||
1109 519 1 批次查询 system:testbatch:query
|
||
1110 519 2 批次新增 system:testbatch:add
|
||
1111 519 3 批次修改 system:testbatch:edit
|
||
1112 519 4 批次授权 system:testbatch:assign
|
||
1113 519 5 批次关闭 system:testbatch:close
|
||
1114 519 6 数据纳入 system:testbatch:adopt
|
||
1115 519 7 批次清理 system:testbatch:purge
|
||
"""
|
||
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()
|
||
)
|
||
}
|
||
EXPECTED_PAGE_PERMISSIONS = frozenset(
|
||
[
|
||
'system:testbatch:list',
|
||
'master:strain:list',
|
||
'master:customer:list',
|
||
'master:site:list',
|
||
'master:line:list',
|
||
'master:warehouse:list',
|
||
'master:pallet:list',
|
||
'master:tunnel:list',
|
||
'business:inbound:list',
|
||
'business:filling:list',
|
||
'business:shipment:list',
|
||
'business:usage:list',
|
||
'business:return:list',
|
||
'report:palletinventory:list',
|
||
'report:filling:list',
|
||
'report:shipment:list',
|
||
'report:consumption:list',
|
||
'report:return:list',
|
||
'report:pallettrace:list',
|
||
]
|
||
)
|
||
EXPECTED_BUTTON_PERMISSIONS = frozenset(
|
||
[
|
||
'master:strain:query',
|
||
'master:strain:add',
|
||
'master:strain:edit',
|
||
'master:customer:query',
|
||
'master:customer:add',
|
||
'master:customer:edit',
|
||
'master:site:query',
|
||
'master:site:add',
|
||
'master:site:edit',
|
||
'master:site:assign',
|
||
'master:line:query',
|
||
'master:line:add',
|
||
'master:line:edit',
|
||
'master:warehouse:query',
|
||
'master:warehouse:add',
|
||
'master:warehouse:edit',
|
||
'master:warehouse:assign',
|
||
'master:pallet:query',
|
||
'master:pallet:add',
|
||
'master:pallet:edit',
|
||
'master:pallet:print',
|
||
'master:tunnel:query',
|
||
'master:tunnel:add',
|
||
'master:tunnel:edit',
|
||
'business:inbound:query',
|
||
'business:inbound:add',
|
||
'business:inbound:edit',
|
||
'business:inbound:confirm',
|
||
'business:filling:query',
|
||
'business:filling:add',
|
||
'business:shipment:query',
|
||
'business:shipment:add',
|
||
'business:shipment:edit',
|
||
'system:testbatch:query',
|
||
'system:testbatch:add',
|
||
'system:testbatch:edit',
|
||
'system:testbatch:assign',
|
||
'system:testbatch:close',
|
||
'system:testbatch:adopt',
|
||
'system:testbatch:purge',
|
||
'business:shipment:confirm',
|
||
'business:usage:query',
|
||
'business:usage:add',
|
||
'business:return:query',
|
||
'business:return:request',
|
||
'business:return:confirm',
|
||
]
|
||
)
|
||
EXPECTED_PERMISSIONS = EXPECTED_PAGE_PERMISSIONS | EXPECTED_BUTTON_PERMISSIONS
|
||
BUSINESS_MENU_SPECS = """
|
||
519|1|测试数据批次|testBatch||7|system:testbatch:list|system/testBatch/index
|
||
121|0|基础资料维护|businessMaster||1||<NULL>
|
||
122|0|业务管理|businessOperation||2||<NULL>
|
||
123|0|报表查询|businessReport||3||<NULL>
|
||
502|121|菌种资料|strain||1|master:strain:list|master/strain/index
|
||
503|121|客户资料|customer||2|master:customer:list|master/customer/index
|
||
504|121|客户现场|site||3|master:site:list|master/site/index
|
||
505|121|产线资料|line||4|master:line:list|master/line/index
|
||
506|121|仓库资料|warehouse||5|master:warehouse:list|master/warehouse/index
|
||
507|121|托盘资料|pallet||6|master:pallet:list|master/pallet/index
|
||
520|121|隧道资料|tunnel||7|master:tunnel:list|master/tunnel/index
|
||
508|122|仓库入库|inbound||1|business:inbound:list|business/inbound/index
|
||
509|122|工厂装填|filling||2|business:filling:list|business/filling/index
|
||
510|122|发货出库|shipment||3|business:shipment:list|business/shipment/index
|
||
511|122|客户现场使用|usage||4|business:usage:list|business/usage/index
|
||
512|122|退回仓库|return||5|business:return:list|business/return/index
|
||
513|123|托盘库存|palletInventory||1|report:palletinventory:list|report/palletInventory/index
|
||
514|123|装填记录|reportFilling||2|report:filling:list|report/filling/index
|
||
515|123|发货出库|reportShipment||3|report:shipment:list|report/shipment/index
|
||
516|123|产线消耗|consumption||4|report:consumption:list|report/consumption/index
|
||
517|123|退回仓库|reportReturn||5|report:return:list|report/return/index
|
||
518|123|托盘追溯|palletTrace||6|report:pallettrace:list|report/palletTrace/index
|
||
1065|502|菌种查询|||1|master:strain:query|
|
||
1066|502|菌种新增|||2|master:strain:add|
|
||
1067|502|菌种修改|||3|master:strain:edit|
|
||
1069|503|客户查询|||1|master:customer:query|
|
||
1070|503|客户新增|||2|master:customer:add|
|
||
1071|503|客户修改|||3|master:customer:edit|
|
||
1073|504|现场查询|||1|master:site:query|
|
||
1074|504|现场新增|||2|master:site:add|
|
||
1075|504|现场修改|||3|master:site:edit|
|
||
1077|504|分配用户|||5|master:site:assign|
|
||
1078|505|产线查询|||1|master:line:query|
|
||
1079|505|产线新增|||2|master:line:add|
|
||
1080|505|产线修改|||3|master:line:edit|
|
||
1082|506|仓库查询|||1|master:warehouse:query|
|
||
1083|506|仓库新增|||2|master:warehouse:add|
|
||
1084|506|仓库修改|||3|master:warehouse:edit|
|
||
1086|506|分配用户|||5|master:warehouse:assign|
|
||
1087|507|托盘查询|||1|master:pallet:query|
|
||
1088|507|托盘新增|||2|master:pallet:add|
|
||
1089|507|托盘修改|||3|master:pallet:edit|
|
||
1091|507|托盘打印|||5|master:pallet:print|
|
||
1116|520|隧道查询|||1|master:tunnel:query|
|
||
1117|520|隧道新增|||2|master:tunnel:add|
|
||
1118|520|隧道修改|||3|master:tunnel:edit|
|
||
1092|508|入库查询|||1|business:inbound:query|
|
||
1093|508|入库新增|||2|business:inbound:add|
|
||
1094|508|入库修改|||3|business:inbound:edit|
|
||
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|
|
||
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|
|
||
1109|519|批次查询|||1|system:testbatch:query|
|
||
1110|519|批次新增|||2|system:testbatch:add|
|
||
1111|519|批次修改|||3|system:testbatch:edit|
|
||
1112|519|批次授权|||4|system:testbatch:assign|
|
||
1113|519|批次关闭|||5|system:testbatch:close|
|
||
1114|519|数据纳入|||6|system:testbatch:adopt|
|
||
1115|519|批次清理|||7|system:testbatch:purge|
|
||
""".strip()
|
||
|
||
|
||
def _expected_business_menu_fields() -> dict[int, tuple]:
|
||
expected = {}
|
||
for spec in BUSINESS_MENU_SPECS.splitlines():
|
||
menu_id, parent_id, menu_name, path, route_name, order_num, perms, component = spec.split('|')
|
||
expected[int(menu_id)] = (
|
||
int(parent_id),
|
||
menu_name,
|
||
path,
|
||
route_name,
|
||
int(order_num),
|
||
perms,
|
||
None if component == '<NULL>' else component,
|
||
)
|
||
return expected
|
||
|
||
|
||
EXPECTED_BUSINESS_MENU_FIELDS = _expected_business_menu_fields()
|
||
|
||
LEGACY_REMOVED_BUTTON_SPECS = {
|
||
1068: ('菌种删除', 502, 4, 'master:strain:remove'),
|
||
1072: ('客户删除', 503, 4, 'master:customer:remove'),
|
||
1076: ('现场删除', 504, 4, 'master:site:remove'),
|
||
1081: ('产线删除', 505, 4, 'master:line:remove'),
|
||
1085: ('仓库删除', 506, 4, 'master:warehouse:remove'),
|
||
1090: ('托盘删除', 507, 4, 'master:pallet:remove'),
|
||
1095: ('入库删除', 508, 4, 'business:inbound:remove'),
|
||
1102: ('发货删除', 510, 4, 'business:shipment:remove'),
|
||
}
|
||
LEGACY_BUSINESS_MENU_IDS = set(range(121, 124)) | set(range(502, 519)) | set(range(1065, 1109))
|
||
LEGACY_EXPECTED_BUSINESS_MENU_FIELDS = {
|
||
menu_id: fields for menu_id, fields in EXPECTED_BUSINESS_MENU_FIELDS.items() if menu_id in LEGACY_BUSINESS_MENU_IDS
|
||
}
|
||
LEGACY_EXPECTED_BUSINESS_MENU_FIELDS.update(
|
||
{
|
||
menu_id: (parent_id, name, '', '', order_num, permission, '')
|
||
for menu_id, (name, parent_id, order_num, permission) in LEGACY_REMOVED_BUTTON_SPECS.items()
|
||
}
|
||
)
|
||
LEGACY_EXPECTED_PAGE_PERMISSIONS = EXPECTED_PAGE_PERMISSIONS - {'system:testbatch:list', 'master:tunnel:list'}
|
||
LEGACY_EXPECTED_BUTTON_PERMISSIONS = EXPECTED_BUTTON_PERMISSIONS - {
|
||
permission
|
||
for permission in EXPECTED_BUTTON_PERMISSIONS
|
||
if permission.startswith(('system:testbatch:', 'master:tunnel:'))
|
||
} | {spec[3] for spec in LEGACY_REMOVED_BUTTON_SPECS.values()}
|
||
LEGACY_EXPECTED_PERMISSIONS = LEGACY_EXPECTED_PAGE_PERMISSIONS | LEGACY_EXPECTED_BUTTON_PERMISSIONS
|
||
LEGACY_PATCH_BUSINESS_MENU_FIELDS = LEGACY_EXPECTED_BUSINESS_MENU_FIELDS | {
|
||
121: (1, '基础资料维护', 'businessMaster', '', 10, '', None),
|
||
122: (1, '业务管理', 'businessOperation', '', 11, '', None),
|
||
123: (1, '报表查询', 'businessReport', '', 12, '', None),
|
||
}
|
||
# 同样在菜单旧快照生成后加入收货权限，避免改变历史迁移的权限清单。
|
||
EXPECTED_BUTTONS[1119] = ('收货核对', 510, 6, 'business:shipment:receive')
|
||
EXPECTED_BUSINESS_MENU_FIELDS[1119] = (510, '收货核对', '', '', 6, 'business:shipment:receive', '')
|
||
MASTER_EXCEL_BUTTONS = {
|
||
1120: ('菌种导入', 502, 4, 'master:strain:import'),
|
||
1121: ('菌种导出', 502, 5, 'master:strain:export'),
|
||
1122: ('客户导入', 503, 4, 'master:customer:import'),
|
||
1123: ('客户导出', 503, 5, 'master:customer:export'),
|
||
1124: ('现场导入', 504, 6, 'master:site:import'),
|
||
1125: ('现场导出', 504, 7, 'master:site:export'),
|
||
1126: ('产线导入', 505, 4, 'master:line:import'),
|
||
1127: ('产线导出', 505, 5, 'master:line:export'),
|
||
1128: ('仓库导入', 506, 6, 'master:warehouse:import'),
|
||
1129: ('仓库导出', 506, 7, 'master:warehouse:export'),
|
||
1130: ('托盘导入', 507, 6, 'master:pallet:import'),
|
||
1131: ('托盘导出', 507, 7, 'master:pallet:export'),
|
||
1132: ('隧道导入', 520, 4, 'master:tunnel:import'),
|
||
1133: ('隧道导出', 520, 5, 'master:tunnel:export'),
|
||
}
|
||
for menu_id, spec in MASTER_EXCEL_BUTTONS.items():
|
||
EXPECTED_BUTTONS[menu_id] = spec
|
||
name, parent_id, order_num, permission = spec
|
||
EXPECTED_BUSINESS_MENU_FIELDS[menu_id] = (parent_id, name, '', '', order_num, permission, '')
|
||
EXPECTED_BUTTON_PERMISSIONS |= {'business:shipment:receive', *(spec[3] for spec in MASTER_EXCEL_BUTTONS.values())}
|
||
EXPECTED_PERMISSIONS = EXPECTED_PAGE_PERMISSIONS | EXPECTED_BUTTON_PERMISSIONS
|
||
EXPECTED_TOP_LEVEL_MENU_POSITIONS = {
|
||
121: ('基础资料维护', 0, 1),
|
||
122: ('业务管理', 0, 2),
|
||
123: ('报表查询', 0, 3),
|
||
1: ('系统管理', 0, 4),
|
||
2: ('系统监控', 0, 5),
|
||
3: ('系统工具', 0, 6),
|
||
}
|
||
EXPECTED_BUSINESS_ROOT_ICONS = {
|
||
121: 'documentation',
|
||
122: 'build',
|
||
123: 'chart',
|
||
}
|
||
EXPECTED_TOP_LEVEL_MENU_STABLE_FIELDS = {
|
||
121: ('基础资料维护', 'businessMaster', None, '', '', '1', '0', 'M', '0', '0', '', 'documentation'),
|
||
122: ('业务管理', 'businessOperation', None, '', '', '1', '0', 'M', '0', '0', '', 'build'),
|
||
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'),
|
||
}
|
||
MENU_PROMOTION_IDS = set(EXPECTED_TOP_LEVEL_MENU_POSITIONS)
|
||
BUSINESS_MENU_IDS = (
|
||
set(range(121, 124))
|
||
| set(range(502, 519))
|
||
| {519, 520, 1119}
|
||
| set(range(1065, 1068))
|
||
| set(range(1069, 1072))
|
||
| set(range(1073, 1076))
|
||
| {1077}
|
||
| set(range(1078, 1081))
|
||
| set(range(1082, 1085))
|
||
| {1086}
|
||
| set(range(1087, 1090))
|
||
| {1091}
|
||
| set(range(1092, 1095))
|
||
| {1096}
|
||
| set(range(1097, 1099))
|
||
| set(range(1099, 1102))
|
||
| {1103}
|
||
| set(range(1104, 1109))
|
||
| set(range(1109, 1119))
|
||
| set(range(1120, 1134))
|
||
)
|
||
TEST_BATCH_MENU_IDS = {519, 1109, 1110, 1111, 1112, 1113, 1114, 1115}
|
||
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:
|
||
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 _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(
|
||
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 _column_definitions(body: str) -> dict[str, str]:
|
||
definitions = {}
|
||
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)
|
||
if identifier_match and identifier_match.group(1).lower() not in reserved:
|
||
definitions[identifier_match.group(1).lower()] = clean
|
||
return definitions
|
||
|
||
|
||
def _assert_test_batch_sql_contract(
|
||
test_case: unittest.TestCase,
|
||
sql: str,
|
||
*,
|
||
dialect: str,
|
||
is_patch: bool,
|
||
) -> None:
|
||
clean_sql = _strip_sql_comments(sql).replace('`', '')
|
||
normalized = re.sub(r'\s+', ' ', clean_sql).strip().lower()
|
||
tables = _create_table_bodies(sql)
|
||
|
||
test_case.assertNotRegex(normalized, r'\bon\s+delete\s+cascade\b')
|
||
if is_patch:
|
||
test_case.assertNotRegex(normalized, r'\bupdate\b')
|
||
|
||
for table_name in EXPECTED_TEST_BATCH_TABLES:
|
||
test_case.assertIn(table_name, tables)
|
||
columns, primary_key, unique_keys, foreign_keys, checks = _table_contract(tables[table_name])
|
||
test_case.assertEqual(EXPECTED_COLUMNS[table_name], columns, table_name)
|
||
test_case.assertEqual(EXPECTED_PRIMARY_KEYS[table_name], primary_key, table_name)
|
||
test_case.assertEqual(EXPECTED_UNIQUE_KEYS[table_name], unique_keys, table_name)
|
||
test_case.assertEqual(EXPECTED_FOREIGN_KEYS[table_name], foreign_keys, table_name)
|
||
test_case.assertEqual(EXPECTED_CHECKS[table_name], checks, table_name)
|
||
|
||
create_batch = normalized.index('create table biz_test_batch (')
|
||
create_batch_user = normalized.index('create table biz_test_batch_user (')
|
||
owned_marker = 'alter table' if is_patch else 'create table'
|
||
owned_tables = TEST_BATCH_OWNED_TABLES if is_patch else CURRENT_TEST_BATCH_OWNED_TABLES
|
||
first_owned = min(normalized.index(f'{owned_marker} {table_name} ') for table_name in owned_tables)
|
||
test_case.assertLess(create_batch, create_batch_user)
|
||
test_case.assertLess(create_batch_user, first_owned)
|
||
|
||
for table_name in owned_tables:
|
||
with test_case.subTest(dialect=dialect, patch=is_patch, table=table_name):
|
||
if is_patch:
|
||
test_case.assertRegex(
|
||
normalized,
|
||
rf'\balter table {table_name}\s+add column test_batch_id bigint null\b',
|
||
)
|
||
else:
|
||
definitions = _column_definitions(tables[table_name])
|
||
test_case.assertIn('test_batch_id', definitions)
|
||
test_case.assertRegex(definitions['test_batch_id'], r'\bbigint\b')
|
||
test_case.assertNotRegex(definitions['test_batch_id'], r'\bnot\s+null\b')
|
||
|
||
index_name = f'idx_{table_name}_test_batch_id'
|
||
index_pattern = rf'\b(?:add\s+)?(?:index|key)\s+{index_name}\s*\(\s*test_batch_id\s*\)'
|
||
test_case.assertRegex(normalized, index_pattern)
|
||
test_case.assertIn("comment '测试数据批次id；null表示正式数据'", normalized)
|
||
|
||
constraint_name = f'fk_{table_name}_test_batch_id'
|
||
test_case.assertRegex(
|
||
normalized,
|
||
rf'\bconstraint\s+{constraint_name}\s+foreign\s+key\s*\(\s*test_batch_id\s*\)\s*'
|
||
r'references\s+biz_test_batch\s*\(\s*batch_id\s*\)',
|
||
)
|
||
|
||
|
||
def _menu_rows(sql: str, menu_ids: set[int]) -> 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 menu_ids]
|
||
counts = {menu_id: sum(int(row[0]) == menu_id for row in selected) for menu_id in menu_ids}
|
||
if set(counts) != menu_ids or any(count != 1 for count in counts.values()):
|
||
raise AssertionError(f'menu row counts differ: {counts}')
|
||
return {int(row[0]): row for row in selected}
|
||
|
||
|
||
def _business_menu_rows(sql: str, menu_ids: set[int] | None = None) -> dict[int, list]:
|
||
return _menu_rows(sql, BUSINESS_MENU_IDS if menu_ids is None else menu_ids)
|
||
|
||
|
||
def _assert_top_level_menu_fields(test_case: unittest.TestCase, sql: str) -> None:
|
||
rows = _menu_rows(sql, MENU_PROMOTION_IDS)
|
||
actual_positions = {menu_id: (row[1], int(row[2]), int(row[3])) for menu_id, row in rows.items()}
|
||
test_case.assertEqual(EXPECTED_TOP_LEVEL_MENU_POSITIONS, actual_positions)
|
||
stable_indexes = (1, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14)
|
||
actual_stable_fields = {menu_id: tuple(row[index] for index in stable_indexes) for menu_id, row in rows.items()}
|
||
test_case.assertEqual(EXPECTED_TOP_LEVEL_MENU_STABLE_FIELDS, actual_stable_fields)
|
||
|
||
|
||
def _assert_business_menu_fields(
|
||
test_case: unittest.TestCase,
|
||
sql: str,
|
||
expected_fields: dict[int, tuple] = EXPECTED_BUSINESS_MENU_FIELDS,
|
||
) -> None:
|
||
rows = _business_menu_rows(sql, set(expected_fields))
|
||
invalid_widths = {menu_id: len(row) for menu_id, row in rows.items() if len(row) != SYS_MENU_VALUE_COUNT}
|
||
test_case.assertEqual({}, invalid_widths)
|
||
actual = {
|
||
menu_id: (
|
||
int(row[2]),
|
||
row[1],
|
||
row[4],
|
||
row[7],
|
||
int(row[3]),
|
||
row[13],
|
||
row[5],
|
||
)
|
||
for menu_id, row in rows.items()
|
||
}
|
||
test_case.assertEqual(expected_fields, actual)
|
||
|
||
|
||
def _assert_business_permission_inventory(
|
||
test_case: unittest.TestCase,
|
||
sql: str,
|
||
expected_permissions: frozenset[str] = EXPECTED_PERMISSIONS,
|
||
) -> None:
|
||
permission_rows = [
|
||
row
|
||
for row in _insert_rows(sql, 'sys_menu')
|
||
if len(row) > SYS_MENU_PERMS_INDEX
|
||
and row[SYS_MENU_PERMS_INDEX]
|
||
and re.match(r'^(?:master|business|report|system:testbatch):', row[SYS_MENU_PERMS_INDEX])
|
||
]
|
||
test_case.assertTrue(all(len(row) == SYS_MENU_VALUE_COUNT for row in permission_rows))
|
||
actual_permissions = Counter(row[SYS_MENU_PERMS_INDEX] for row in permission_rows)
|
||
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 = [
|
||
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
|
||
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))
|
||
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)
|
||
test_case.assertRegex(guard, identity_aggregate)
|
||
|
||
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
|
||
|
||
|
||
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],
|
||
)
|
||
|
||
|
||
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 ShipmentReceiptSqlContractTest(unittest.TestCase):
|
||
def patch_sql(self) -> str:
|
||
self.assertTrue(SHIPMENT_RECEIPT_PATCH_PATH.is_file(), '缺少收货单次迁移脚本')
|
||
return SHIPMENT_RECEIPT_PATCH_PATH.read_text(encoding='utf-8')
|
||
|
||
def test_patch_additions_match_initializer_columns_and_constraints(self) -> None:
|
||
def normalize(value: str) -> str:
|
||
return re.sub(r'\b(bigint)\(20\)', r'\1', value).split(' comment ')[0]
|
||
|
||
raw = _strip_sql_comments(self.patch_sql())
|
||
initial = _create_table_bodies((SQL_ROOT / INITIALIZERS[0]).read_text(encoding='utf-8'))
|
||
alters = re.findall(r'\balter\s+table\s+(\w+)\s+(.*?);', raw, re.I | re.S)
|
||
self.assertEqual(['biz_shipment_order', 'biz_pallet_inventory', 'biz_pallet_event'], [table for table, _ in alters])
|
||
expected_columns = {
|
||
'biz_shipment_order': {'receipt_state', 'received_by', 'received_at', 'receipt_idempotency_key', 'receipt_request_hash'},
|
||
'biz_pallet_inventory': {'site_arrived_at'},
|
||
'biz_pallet_event': set(),
|
||
}
|
||
for table, body in alters:
|
||
with self.subTest(table=table):
|
||
additions = ','.join(
|
||
re.sub(r'^add\s+(?:column\s+)?', '', clause.strip(), flags=re.I)
|
||
for clause in _split_values(body) if clause.strip().lower().startswith('add ')
|
||
)
|
||
columns, _, uniques, foreign_keys, checks = _table_contract(additions)
|
||
self.assertEqual(expected_columns[table], columns)
|
||
self.assertEqual(
|
||
{('target_site_id', 'receipt_idempotency_key')} if table == 'biz_shipment_order' else set(), uniques,
|
||
)
|
||
self.assertEqual({('received_by', 'sys_user.user_id')} if table == 'biz_shipment_order' else set(), foreign_keys)
|
||
expected_checks = (
|
||
{"receipt_statein('pending','received','legacy')"} if table == 'biz_shipment_order'
|
||
else {check for check in EXPECTED_CHECKS[table] if check.startswith(('flow_statein', 'event_typein'))}
|
||
)
|
||
self.assertEqual(expected_checks, checks)
|
||
initializer_columns = _column_definitions(initial[table])
|
||
for column, definition in _column_definitions(additions).items():
|
||
normalized = _normalized_sql_statement(re.sub(r'\s+after\s+\w+\s*$', '', definition, flags=re.I))
|
||
# 类型、可空性和默认值均应与初始化定义一致，注释不参与比较。
|
||
self.assertEqual(normalize(_normalized_sql_statement(initializer_columns[column])), normalize(normalized))
|
||
|
||
def test_patch_only_marks_confirmed_legacy_and_does_not_touch_inventory_or_receipt_time(self) -> None:
|
||
raw = _strip_sql_comments(self.patch_sql())
|
||
updates = re.findall(r'\bupdate\s+([a-z_]+)\s+set\s+(.*?);', raw, re.I | re.S)
|
||
self.assertEqual(1, len(updates))
|
||
self.assertEqual('biz_shipment_order', updates[0][0].lower())
|
||
self.assertEqual("receipt_state = 'legacy' where status = 'confirmed'", _normalized_sql_statement(updates[0][1]))
|
||
inserts = re.findall(r'\binsert\s+into\s+(\w+)', raw, re.I)
|
||
self.assertEqual(['sys_menu'], inserts)
|
||
self.assertNotRegex(raw.lower(), r'\b(?:delete\s+from|replace\s+into|truncate\s+table)\b')
|
||
self.assertNotRegex(raw.lower(), r'\b(?:received_at|site_arrived_at)\s*=')
|
||
|
||
def test_patch_guards_conflicting_menu_ids_and_permissions_before_any_alter(self) -> None:
|
||
raw = _strip_sql_comments(self.patch_sql()).lower()
|
||
guard = raw[:raw.index('alter table')]
|
||
for expression in (
|
||
r'from\s+sys_menu\s+where\s+menu_id\s*=\s*1119\s+or\s+perms\s*=\s*\'business:shipment:receive\'',
|
||
r'from\s+sys_role_menu\s+where\s+menu_id\s*=\s*1119',
|
||
r'menu_id\s*=\s*510\s+and\s+perms\s*=\s*\'business:shipment:list\'',
|
||
):
|
||
self.assertRegex(guard, expression)
|
||
self.assertIn("signal sqlstate '45000'", guard)
|
||
self.assertIn('resignal', guard)
|
||
rows = _insert_rows(raw, 'sys_menu')
|
||
self.assertEqual(1, len(rows))
|
||
self.assertEqual(SYS_MENU_VALUE_COUNT, len(rows[0]))
|
||
self.assertEqual(['1119', '收货核对', '510', '6'], rows[0][:4])
|
||
self.assertEqual('business:shipment:receive', rows[0][SYS_MENU_PERMS_INDEX])
|
||
self.assertNotRegex(raw, r'\b(?:insert\s+into|update|delete\s+from)\s+sys_role_menu\b')
|
||
|
||
def test_patch_requires_prerequisites_and_rejects_repeated_or_partial_execution(self) -> None:
|
||
raw = self.patch_sql()
|
||
for phrase in ('MySQL 8', '停写', '备份', '单次', 'DDL', '隐式提交', '20261005_tunnel_material_flow_mysql.sql', '20261008_material_dates_mysql.sql'):
|
||
self.assertIn(phrase, raw)
|
||
guard = _strip_sql_comments(raw).lower().split('alter table', maxsplit=1)[0]
|
||
self.assertIn('version()', guard)
|
||
self.assertIn('information_schema.columns', guard)
|
||
for field in ('production_date', 'expiry_date', 'tunnel_id', 'trace_no', 'receipt_state', 'received_by',
|
||
'received_at', 'receipt_idempotency_key', 'receipt_request_hash', 'site_arrived_at'):
|
||
self.assertIn(f"'{field}'", guard)
|
||
self.assertIn('information_schema.check_constraints', guard)
|
||
self.assertIn("'ck_biz_filling_record_material_dates'", guard)
|
||
self.assertIn("'ck_biz_pallet_event_event_type'", guard)
|
||
self.assertIn('tunnel_entered', guard)
|
||
self.assertLess(raw.lower().rindex('alter table'), raw.lower().index('start transaction'))
|
||
|
||
def test_current_expectations_leave_legacy_schema_and_menu_snapshots_unchanged(self) -> None:
|
||
self.assertNotIn('receipt_state', LEGACY_EXPECTED_COLUMNS['biz_shipment_order'])
|
||
self.assertNotIn('site_arrived_at', LEGACY_EXPECTED_COLUMNS['biz_pallet_inventory'])
|
||
self.assertNotIn(('target_site_id', 'receipt_idempotency_key'), LEGACY_EXPECTED_UNIQUE_KEYS['biz_shipment_order'])
|
||
self.assertNotIn(('received_by', 'sys_user.user_id'), LEGACY_EXPECTED_FOREIGN_KEYS['biz_shipment_order'])
|
||
self.assertNotIn('in_transit', ' '.join(LEGACY_EXPECTED_CHECKS['biz_pallet_inventory']))
|
||
self.assertNotIn('shipment_received', ' '.join(LEGACY_EXPECTED_CHECKS['biz_pallet_event']))
|
||
self.assertNotIn('business:shipment:receive', LEGACY_EXPECTED_PERMISSIONS)
|
||
self.assertNotIn(1119, LEGACY_EXPECTED_BUSINESS_MENU_FIELDS)
|
||
|
||
|
||
class MasterExcelMenuPatchContractTest(unittest.TestCase):
|
||
def patch_sql(self) -> str:
|
||
self.assertTrue(MASTER_EXCEL_PATCH_PATH.is_file(), '缺少基础资料导入导出菜单迁移脚本')
|
||
return MASTER_EXCEL_PATCH_PATH.read_text(encoding='utf-8')
|
||
|
||
def test_patch_only_inserts_master_excel_menus_without_role_grants(self) -> None:
|
||
raw = self.patch_sql()
|
||
stripped = _strip_sql_comments(raw).lower()
|
||
inserts = re.findall(r'\binsert\s+into\s+(\w+)', stripped)
|
||
self.assertEqual(['sys_menu'], inserts)
|
||
self.assertNotRegex(stripped, r'\b(?:alter\s+table|delete\s+from|replace\s+into|truncate\s+table|update\s+\w+)\b')
|
||
self.assertNotRegex(stripped, r'\b(?:insert\s+into|update|delete\s+from)\s+sys_role_menu\b')
|
||
self.assertIn('不写 sys_role_menu', raw)
|
||
guard = stripped.split('start transaction', maxsplit=1)[0]
|
||
self.assertIn('menu_id between 1120 and 1133', guard)
|
||
self.assertIn('perms in (', guard)
|
||
self.assertIn('from sys_role_menu where menu_id between 1120 and 1133', guard)
|
||
self.assertIn("signal sqlstate '45000'", guard)
|
||
self.assertIn('resignal', guard)
|
||
self.assertLess(stripped.index("signal sqlstate '45000'"), stripped.index('start transaction'))
|
||
|
||
rows = _insert_rows(raw, 'sys_menu')
|
||
self.assertEqual(list(MASTER_EXCEL_BUTTONS), [int(row[0]) for row in rows])
|
||
for row in rows:
|
||
menu_id = int(row[0])
|
||
name, parent_id, order_num, permission = MASTER_EXCEL_BUTTONS[menu_id]
|
||
self.assertEqual(SYS_MENU_VALUE_COUNT, len(row), menu_id)
|
||
self.assertEqual([str(menu_id), name, str(parent_id), str(order_num)], row[:4])
|
||
self.assertEqual(permission, row[SYS_MENU_PERMS_INDEX])
|
||
self.assertEqual(
|
||
('', '', '', '', '1', '0', 'F', '0', '0'),
|
||
tuple(row[index] for index in (4, 5, 6, 7, 8, 9, 10, 11, 12)),
|
||
)
|
||
|
||
def test_patch_does_not_change_legacy_menu_snapshot(self) -> None:
|
||
self.assertTrue(MASTER_EXCEL_BUTTONS.keys().isdisjoint(LEGACY_EXPECTED_BUSINESS_MENU_FIELDS))
|
||
self.assertTrue({spec[3] for spec in MASTER_EXCEL_BUTTONS.values()}.isdisjoint(LEGACY_EXPECTED_PERMISSIONS))
|
||
|
||
|
||
class SqlContractTest(unittest.TestCase):
|
||
def test_test_batch_initializers_and_patches_match_contract(self) -> None:
|
||
scripts = (
|
||
(SQL_ROOT / 'shierfeng-fastapi.sql', 'mysql', False),
|
||
(TEST_BATCH_MYSQL_PATCH_PATH, 'mysql', True),
|
||
)
|
||
for path, dialect, is_patch in scripts:
|
||
with self.subTest(path=path.name):
|
||
self.assertTrue(path.is_file(), f'missing SQL: {path}')
|
||
_assert_test_batch_sql_contract(
|
||
self,
|
||
path.read_text(encoding='utf-8'),
|
||
dialect=dialect,
|
||
is_patch=is_patch,
|
||
)
|
||
|
||
def test_test_batch_patches_migrate_legacy_menu_permissions_without_role_grants(self) -> None:
|
||
legacy_removed_ids = set(LEGACY_REMOVED_BUTTON_SPECS)
|
||
for path in (TEST_BATCH_MYSQL_PATCH_PATH,):
|
||
with self.subTest(path=path.name):
|
||
raw = path.read_text(encoding='utf-8')
|
||
statements = _sql_statements(raw)
|
||
normalized = [_normalized_sql_statement(statement) for statement in statements]
|
||
|
||
role_delete_indexes = [
|
||
index
|
||
for index, statement in enumerate(normalized)
|
||
if re.fullmatch(r'delete from sys_role_menu where menu_id in \([^)]+\)', statement)
|
||
]
|
||
menu_delete_indexes = [
|
||
index
|
||
for index, statement in enumerate(normalized)
|
||
if re.fullmatch(r'delete from sys_menu where menu_id in \([^)]+\)', statement)
|
||
]
|
||
menu_insert_indexes = [
|
||
index
|
||
for index, statement in enumerate(normalized)
|
||
if statement.startswith('insert into sys_menu values ')
|
||
]
|
||
self.assertEqual(1, len(role_delete_indexes))
|
||
self.assertEqual(1, len(menu_delete_indexes))
|
||
self.assertEqual(1, len(menu_insert_indexes))
|
||
role_delete = normalized[role_delete_indexes[0]]
|
||
menu_delete = normalized[menu_delete_indexes[0]]
|
||
role_delete_ids = {
|
||
int(value.strip()) for value in re.search(r'\(([^)]+)\)', role_delete).group(1).split(',')
|
||
}
|
||
menu_delete_ids = {
|
||
int(value.strip()) for value in re.search(r'\(([^)]+)\)', menu_delete).group(1).split(',')
|
||
}
|
||
self.assertEqual(legacy_removed_ids, role_delete_ids)
|
||
self.assertEqual(legacy_removed_ids, menu_delete_ids)
|
||
self.assertLess(role_delete_indexes[0], menu_delete_indexes[0])
|
||
self.assertLess(menu_delete_indexes[0], menu_insert_indexes[0])
|
||
|
||
inserted_rows = _insert_rows(raw, 'sys_menu')
|
||
inserted_by_id = {int(row[0]): row for row in inserted_rows}
|
||
self.assertEqual(TEST_BATCH_MENU_IDS, set(inserted_by_id))
|
||
self.assertTrue(all(len(row) == SYS_MENU_VALUE_COUNT for row in inserted_by_id.values()))
|
||
self.assertEqual('测试数据批次', inserted_by_id[519][1])
|
||
self.assertEqual('1', inserted_by_id[519][2])
|
||
self.assertEqual('7', inserted_by_id[519][3])
|
||
self.assertEqual('system:testbatch:list', inserted_by_id[519][13])
|
||
self.assertEqual(519, int(inserted_by_id[1109][2]))
|
||
self.assertEqual('system:testbatch:purge', inserted_by_id[1115][13])
|
||
self.assertNotRegex('\n'.join(normalized), r'\binsert into sys_role_menu\b')
|
||
|
||
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_mysql_business_datetimes_use_native_datetime_columns(self) -> None:
|
||
self.assertEqual(45, sum(len(columns) for columns in EXPECTED_DATETIME_COLUMNS.values()))
|
||
tables = _create_table_bodies((SQL_ROOT / 'shierfeng-fastapi.sql').read_text(encoding='utf-8'))
|
||
invalid_columns = []
|
||
for table, expected_columns in EXPECTED_DATETIME_COLUMNS.items():
|
||
definitions = _column_definitions(tables[table])
|
||
actual_columns = {
|
||
column
|
||
for column, definition in definitions.items()
|
||
if re.search(r'\bdatetime\b', definition, flags=re.I)
|
||
}
|
||
self.assertEqual(expected_columns, actual_columns, table)
|
||
for column in expected_columns:
|
||
definition = definitions[column]
|
||
if not re.search(r'\bdatetime\b', definition, flags=re.I):
|
||
invalid_columns.append(f'{table}.{column}')
|
||
self.assertEqual([], sorted(invalid_columns))
|
||
|
||
def test_initializers_define_fixed_business_menus_and_permissions(self) -> None:
|
||
self.assertEqual(19, len(EXPECTED_PAGE_PERMISSIONS))
|
||
self.assertEqual(61, len(EXPECTED_BUTTON_PERMISSIONS))
|
||
self.assertFalse(EXPECTED_PAGE_PERMISSIONS & EXPECTED_BUTTON_PERMISSIONS)
|
||
self.assertEqual(80, len(EXPECTED_PERMISSIONS))
|
||
self.assertEqual(83, len(EXPECTED_BUSINESS_MENU_FIELDS))
|
||
for filename in INITIALIZERS:
|
||
with self.subTest(filename=filename):
|
||
sql = (SQL_ROOT / filename).read_text(encoding='utf-8')
|
||
_assert_business_menu_fields(self, sql)
|
||
_assert_business_permission_inventory(self, sql)
|
||
rows = _business_menu_rows(sql)
|
||
self.assertTrue(all(len(row) == SYS_MENU_VALUE_COUNT for row in rows.values()))
|
||
self.assertEqual(
|
||
('基础资料维护', '0', '1', 'businessMaster', 'M', ''),
|
||
tuple(rows[121][index] for index in (1, 2, 3, 4, 10, 13)),
|
||
)
|
||
self.assertEqual(
|
||
('业务管理', '0', '2', 'businessOperation', 'M', ''),
|
||
tuple(rows[122][index] for index in (1, 2, 3, 4, 10, 13)),
|
||
)
|
||
self.assertEqual(
|
||
('报表查询', '0', '3', '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)),
|
||
)
|
||
|
||
def test_initializers_define_expected_top_level_menu_order(self) -> None:
|
||
for filename in INITIALIZERS:
|
||
with self.subTest(filename=filename):
|
||
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',
|
||
'#',
|
||
'documentation',
|
||
),
|
||
('122', '业务管理', 'businessOperation', None, '', '', '1', '0', 'M', '0', '0', '', '0', '2', '#', 'build'),
|
||
('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 'documentation' when 122 then 'build' "
|
||
"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, '基础资料维护', 'documentation', 'driftBusinessIcon'),
|
||
)
|
||
for filename in INITIALIZERS:
|
||
sql = (SQL_ROOT / filename).read_text(encoding='utf-8')
|
||
for menu_id, menu_name, original_icon, drifted_icon in mutations:
|
||
with self.subTest(filename=filename, menu_id=menu_id):
|
||
mutated, replacements = re.subn(
|
||
rf"(\(?\s*'?{menu_id}'?\s*,\s*'{menu_name}'.*?,\s*)'{re.escape(original_icon)}'"
|
||
r"(\s*,\s*'admin')",
|
||
rf"\1'{drifted_icon}'\2",
|
||
sql,
|
||
count=1,
|
||
)
|
||
self.assertEqual(1, replacements)
|
||
with self.assertRaises(AssertionError):
|
||
_assert_top_level_menu_fields(self, mutated)
|
||
|
||
def test_business_permission_inventory_rejects_duplicate_permission(self) -> None:
|
||
sql = (SQL_ROOT / INITIALIZERS[0]).read_text(encoding='utf-8')
|
||
sql += """
|
||
insert into sys_menu values(
|
||
'1999', '重复权限', '1', '1', '', '', '', '', 1, 0,
|
||
'F', '0', '0', 'master:strain:list', '#', 'admin', sysdate(), '', null, ''
|
||
);
|
||
"""
|
||
|
||
with self.assertRaises(AssertionError):
|
||
_assert_business_permission_inventory(self, sql)
|
||
|
||
def test_business_menu_contract_rejects_same_field_drift_in_both_initializers(self) -> None:
|
||
for filename in INITIALIZERS:
|
||
with self.subTest(filename=filename):
|
||
sql = (SQL_ROOT / filename).read_text(encoding='utf-8')
|
||
mutated, replacements = re.subn(
|
||
r"(\(?\s*'?502'?\s*,\s*)'菌种资料'",
|
||
r"\1'漂移菌种资料'",
|
||
sql,
|
||
count=1,
|
||
)
|
||
self.assertEqual(1, replacements)
|
||
with self.assertRaises(AssertionError):
|
||
_assert_business_menu_fields(self, mutated)
|
||
|
||
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_initializer_uses_mysql_native_menu_allocator(self) -> None:
|
||
mysql = (SQL_ROOT / 'shierfeng-fastapi.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')
|
||
|
||
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',
|
||
'biz_test_batch_user',
|
||
'biz_test_batch',
|
||
)
|
||
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_menu_promotion_patch_is_fail_closed_atomic_and_exact(self) -> None: # noqa: PLR0915
|
||
self.assertTrue(MENU_PROMOTION_PATCH_PATH.is_file(), f'missing patch: {MENU_PROMOTION_PATCH_PATH}')
|
||
raw = MENU_PROMOTION_PATCH_PATH.read_text(encoding='utf-8')
|
||
executable = _strip_sql_comments(raw).replace('`', '')
|
||
normalized = [_normalized_sql_statement(statement) for statement in _sql_statements(raw)]
|
||
|
||
for required_note in ('未自动执行', '执行前备份', '可幂等重跑', '禁止使用 mysql --force', '独立连接'):
|
||
self.assertIn(required_note, raw)
|
||
self.assertNotIn('--force', executable.lower())
|
||
self.assertNotIn('continue-on-error', raw.lower())
|
||
|
||
expected_rows = [
|
||
('1', '系统管理', 'system', None, '', '', '1', '0', 'M', '0', '0', '', 'system', '0', '1', '0', '4'),
|
||
('2', '系统监控', 'monitor', None, '', '', '1', '0', 'M', '0', '0', '', 'monitor', '0', '2', '0', '5'),
|
||
('3', '系统工具', 'tool', None, '', '', '1', '0', 'M', '0', '0', '', 'tool', '0', '3', '0', '6'),
|
||
(
|
||
'121',
|
||
'基础资料维护',
|
||
'businessMaster',
|
||
None,
|
||
'',
|
||
'',
|
||
'1',
|
||
'0',
|
||
'M',
|
||
'0',
|
||
'0',
|
||
'',
|
||
'#',
|
||
'1',
|
||
'10',
|
||
'0',
|
||
'1',
|
||
),
|
||
(
|
||
'122',
|
||
'业务管理',
|
||
'businessOperation',
|
||
None,
|
||
'',
|
||
'',
|
||
'1',
|
||
'0',
|
||
'M',
|
||
'0',
|
||
'0',
|
||
'',
|
||
'#',
|
||
'1',
|
||
'11',
|
||
'0',
|
||
'2',
|
||
),
|
||
(
|
||
'123',
|
||
'报表查询',
|
||
'businessReport',
|
||
None,
|
||
'',
|
||
'',
|
||
'1',
|
||
'0',
|
||
'M',
|
||
'0',
|
||
'0',
|
||
'',
|
||
'#',
|
||
'1',
|
||
'12',
|
||
'0',
|
||
'3',
|
||
),
|
||
]
|
||
self.assertEqual(expected_rows, [tuple(row) for row in _insert_rows(raw, 'promote_business_menus_expected')])
|
||
|
||
def unique_index(predicate: Callable[[str], bool], label: str) -> int:
|
||
matches = [index for index, statement in enumerate(normalized) if predicate(statement)]
|
||
self.assertEqual(1, len(matches), label)
|
||
return matches[0]
|
||
|
||
expected_columns = (
|
||
'menu_id',
|
||
'menu_name',
|
||
'path',
|
||
'component',
|
||
'query',
|
||
'route_name',
|
||
'is_frame',
|
||
'is_cache',
|
||
'menu_type',
|
||
'visible',
|
||
'status',
|
||
'perms',
|
||
'icon',
|
||
'old_parent_id',
|
||
'old_order_num',
|
||
'new_parent_id',
|
||
'new_order_num',
|
||
)
|
||
expected_create_index = unique_index(
|
||
lambda statement: statement.startswith('create temporary table promote_business_menus_expected '),
|
||
'expected table creation',
|
||
)
|
||
expected_insert_index = unique_index(
|
||
lambda statement: statement.startswith('insert into promote_business_menus_expected '),
|
||
'expected row insertion',
|
||
)
|
||
expected_create = normalized[expected_create_index]
|
||
for column in expected_columns:
|
||
self.assertRegex(expected_create, rf'\b{column}\b')
|
||
|
||
guard_create_index = unique_index(
|
||
lambda statement: statement.startswith('create temporary table promote_business_menus_guard '),
|
||
'guard table creation',
|
||
)
|
||
self.assertIn('primary key (guard_token)', normalized[guard_create_index])
|
||
start_indexes = [index for index, statement in enumerate(normalized) if statement == 'start transaction']
|
||
commit_indexes = [index for index, statement in enumerate(normalized) if statement == 'commit']
|
||
self.assertEqual(1, len(start_indexes))
|
||
self.assertEqual(1, len(commit_indexes))
|
||
start_index = start_indexes[0]
|
||
commit_index = commit_indexes[0]
|
||
|
||
initial_cleanup_indexes = (
|
||
unique_index(
|
||
lambda statement: statement == 'drop temporary table if exists promote_business_menus_expected',
|
||
'expected table initial cleanup',
|
||
),
|
||
unique_index(
|
||
lambda statement: statement == 'drop temporary table if exists promote_business_menus_guard',
|
||
'guard table initial cleanup',
|
||
),
|
||
)
|
||
self.assertTrue(
|
||
all(
|
||
index < start_index
|
||
for index in (
|
||
*initial_cleanup_indexes,
|
||
expected_create_index,
|
||
expected_insert_index,
|
||
guard_create_index,
|
||
)
|
||
)
|
||
)
|
||
|
||
lock_index = unique_index(
|
||
lambda statement: statement == MENU_PROMOTION_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)
|
||
if statement.startswith('insert into promote_business_menus_guard ')
|
||
]
|
||
self.assertEqual(2, len(guard_indexes))
|
||
pre_guard_index, post_guard_index = guard_indexes
|
||
self.assertEqual((pre_guard, post_guard), tuple(normalized[index] for index in guard_indexes))
|
||
|
||
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',
|
||
)
|
||
permanent_dml = []
|
||
for statement in normalized:
|
||
if not re.match(r'^(?:insert|update|delete|replace|truncate)\b', statement):
|
||
continue
|
||
if statement.startswith('insert into promote_business_menus_expected '):
|
||
continue
|
||
if statement.startswith('insert into promote_business_menus_guard '):
|
||
continue
|
||
permanent_dml.append(statement)
|
||
self.assertEqual([expected_update], permanent_dml)
|
||
self.assertNotRegex(executable.lower(), r'\bsys_role_menu\b')
|
||
|
||
self.assertLess(start_index, lock_index)
|
||
self.assertLess(lock_index, pre_guard_index)
|
||
self.assertLess(pre_guard_index, update_index)
|
||
self.assertLess(update_index, post_guard_index)
|
||
self.assertLess(post_guard_index, commit_index)
|
||
|
||
guard_drop_index = unique_index(
|
||
lambda statement: statement == 'drop temporary table promote_business_menus_guard',
|
||
'guard table final cleanup',
|
||
)
|
||
expected_drop_index = unique_index(
|
||
lambda statement: statement == 'drop temporary table promote_business_menus_expected',
|
||
'expected table final cleanup',
|
||
)
|
||
self.assertLess(commit_index, guard_drop_index)
|
||
self.assertLess(commit_index, expected_drop_index)
|
||
|
||
expected_verification = (
|
||
'select menu_id, menu_name, parent_id, order_num, path, component, menu_type, visible, status, perms '
|
||
'from sys_menu where menu_id in (1, 2, 3, 121, 122, 123) order by order_num, menu_id'
|
||
)
|
||
self.assertEqual(expected_verification, normalized[-1])
|
||
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_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()
|
||
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(LEGACY_EXPECTED_TABLES, {table for table in tables if table.startswith('biz_')})
|
||
for table in LEGACY_EXPECTED_TABLES:
|
||
with self.subTest(table=table):
|
||
columns, primary_key, unique_keys, foreign_keys, checks = _table_contract(tables[table])
|
||
self.assertEqual(LEGACY_EXPECTED_COLUMNS[table], columns)
|
||
self.assertEqual(LEGACY_EXPECTED_PRIMARY_KEYS[table], primary_key)
|
||
self.assertEqual(LEGACY_EXPECTED_UNIQUE_KEYS[table], unique_keys)
|
||
self.assertEqual(LEGACY_EXPECTED_FOREIGN_KEYS[table], foreign_keys)
|
||
self.assertEqual(LEGACY_EXPECTED_CHECKS[table], checks)
|
||
_assert_business_menu_fields(self, raw, LEGACY_PATCH_BUSINESS_MENU_FIELDS)
|
||
_assert_business_permission_inventory(self, raw, LEGACY_EXPECTED_PERMISSIONS)
|
||
patch_rows = _business_menu_rows(raw, LEGACY_BUSINESS_MENU_IDS)
|
||
mysql_rows = _business_menu_rows((SQL_ROOT / 'shierfeng-fastapi.sql').read_text(encoding='utf-8'))
|
||
compared_indexes = (0, 1, *range(4, 15))
|
||
common_menu_ids = LEGACY_BUSINESS_MENU_IDS & BUSINESS_MENU_IDS
|
||
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()
|
||
if menu_id in common_menu_ids
|
||
}
|
||
self.assertEqual(
|
||
expected_legacy_rows,
|
||
{
|
||
menu_id: tuple(row[index] for index in compared_indexes)
|
||
for menu_id, row in patch_rows.items()
|
||
if menu_id in common_menu_ids
|
||
},
|
||
)
|
||
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)
|
||
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_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(LEGACY_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(LEGACY_BUSINESS_MENU_IDS), len(verification_ids))
|
||
self.assertEqual(LEGACY_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_preflights_new_menu_ids_before_atomic_permanent_dml(self) -> None:
|
||
raw = PATCH_PATH.read_text(encoding='utf-8')
|
||
statements = _sql_statements(raw)
|
||
normalized = [_normalized_sql_statement(statement) for statement in statements]
|
||
|
||
expected_table = 'pallet_workflow_menu_seed_expected'
|
||
conflict_guard = 'pallet_workflow_menu_seed_guard'
|
||
|
||
def statement_indexes(pattern: str) -> list[int]:
|
||
return [index for index, statement in enumerate(normalized) if re.fullmatch(pattern, statement, flags=re.S)]
|
||
|
||
expected_initial_drops = statement_indexes(rf'drop temporary table if exists {expected_table}')
|
||
expected_creates = statement_indexes(rf'create temporary table {expected_table}\s*\(.*\)')
|
||
expected_inserts = statement_indexes(rf'insert into {expected_table}\s*\(.*')
|
||
guard_initial_drops = statement_indexes(rf'drop temporary table if exists {conflict_guard}')
|
||
guard_creates = statement_indexes(rf'create temporary table {conflict_guard}\s*\(.*\)')
|
||
guard_inserts = statement_indexes(rf'insert into {conflict_guard}\s*\(.*')
|
||
guard_final_drops = statement_indexes(rf'drop temporary table {conflict_guard}')
|
||
expected_final_drops = statement_indexes(rf'drop temporary table {expected_table}')
|
||
|
||
markers = {
|
||
'expected table initial cleanup': expected_initial_drops,
|
||
'expected table creation': expected_creates,
|
||
'expected seed insertion': expected_inserts,
|
||
'conflict guard initial cleanup': guard_initial_drops,
|
||
'conflict guard creation': guard_creates,
|
||
'conflict guard check': guard_inserts,
|
||
'conflict guard success cleanup': guard_final_drops,
|
||
'expected table success cleanup': expected_final_drops,
|
||
}
|
||
for name, indexes in markers.items():
|
||
with self.subTest(marker=name):
|
||
self.assertEqual(1, len(indexes), f'missing or duplicate {name}')
|
||
|
||
legacy_guard_cleanup = normalized.index('drop temporary table pallet_workflow_legacy_seed_guard')
|
||
permanent_deletes = [
|
||
index
|
||
for index, statement in enumerate(normalized)
|
||
if re.match(r'delete from sys_(?:role_menu|menu|dict_data|dict_type)\b', statement)
|
||
]
|
||
self.assertTrue(permanent_deletes)
|
||
first_permanent_delete = permanent_deletes[0]
|
||
|
||
ordered_markers = (
|
||
legacy_guard_cleanup,
|
||
expected_initial_drops[0],
|
||
expected_creates[0],
|
||
expected_inserts[0],
|
||
guard_initial_drops[0],
|
||
guard_creates[0],
|
||
guard_inserts[0],
|
||
guard_final_drops[0],
|
||
expected_final_drops[0],
|
||
first_permanent_delete,
|
||
)
|
||
self.assertEqual(tuple(sorted(ordered_markers)), ordered_markers)
|
||
|
||
expected_insert = statements[expected_inserts[0]].replace('`', '').lower()
|
||
expected_insert_match = re.fullmatch(
|
||
rf'\s*insert\s+into\s+{expected_table}\s*\((?P<columns>[^)]+)\)\s+values\s+.*',
|
||
expected_insert,
|
||
flags=re.S,
|
||
)
|
||
self.assertIsNotNone(expected_insert_match)
|
||
expected_columns = tuple(column.strip() for column in expected_insert_match.group('columns').split(','))
|
||
self.assertEqual(('menu_id', *MYSQL_MENU_STABLE_COLUMNS), expected_columns)
|
||
|
||
expected_rows = _insert_rows(statements[expected_inserts[0]], expected_table)
|
||
self.assertEqual(len(LEGACY_BUSINESS_MENU_IDS), len(expected_rows))
|
||
expected_rows_by_id = {int(row[0]): tuple(row) for row in expected_rows}
|
||
self.assertEqual(LEGACY_BUSINESS_MENU_IDS, set(expected_rows_by_id))
|
||
seed_rows_by_id = {
|
||
menu_id: tuple(row[: len(expected_columns)])
|
||
for menu_id, row in _business_menu_rows(raw, LEGACY_BUSINESS_MENU_IDS).items()
|
||
}
|
||
self.assertEqual(seed_rows_by_id, expected_rows_by_id)
|
||
|
||
guard_create = normalized[guard_creates[0]]
|
||
self.assertRegex(guard_create, r'primary key\s*\(\s*guard_token\s*\)')
|
||
guard_insert = normalized[guard_inserts[0]]
|
||
self.assertEqual(2, guard_insert.count("select 'business-menu-seed-conflict'"))
|
||
self.assertIn(
|
||
"union all select 'business-menu-seed-conflict' as guard_token where exists (",
|
||
guard_insert,
|
||
)
|
||
self.assertNotIn(' on duplicate key update ', guard_insert)
|
||
expected_comparisons = ' and '.join(
|
||
f'actual.{column} <=> expected.{column}' for column in MYSQL_MENU_STABLE_COLUMNS
|
||
)
|
||
guard_insert = re.sub(r'\(\s+', '(', guard_insert)
|
||
guard_insert = re.sub(r'\s+\)', ')', guard_insert)
|
||
self.assertIn(
|
||
f'from sys_menu as actual join {expected_table} as expected '
|
||
f'on actual.menu_id = expected.menu_id where not ({expected_comparisons})',
|
||
guard_insert,
|
||
)
|
||
|
||
preflight_sql = '; '.join(normalized[expected_creates[0] : first_permanent_delete])
|
||
self.assertNotRegex(
|
||
preflight_sql,
|
||
r'\b(?:insert\s+into|replace\s+into|update|delete\s+from)\s+sys_menu\b',
|
||
)
|
||
|
||
def test_patch_wraps_permanent_seed_dml_in_explicit_transaction(self) -> None:
|
||
statements = _sql_statements(PATCH_PATH.read_text(encoding='utf-8'))
|
||
normalized = [_normalized_sql_statement(statement) for statement in statements]
|
||
start_transactions = [index for index, statement in enumerate(normalized) if statement == 'start transaction']
|
||
commits = [index for index, statement in enumerate(normalized) if statement == 'commit']
|
||
self.assertEqual(1, len(start_transactions), 'permanent seed DML must start one explicit transaction')
|
||
self.assertEqual(1, len(commits), 'permanent seed DML must commit one explicit transaction')
|
||
start_transaction = start_transactions[0]
|
||
commit = commits[0]
|
||
|
||
permanent_deletes = [
|
||
index
|
||
for index, statement in enumerate(normalized)
|
||
if re.match(r'delete from sys_(?:role_menu|menu|dict_data|dict_type)\b', statement)
|
||
]
|
||
self.assertTrue(permanent_deletes)
|
||
menu_seed_inserts = [
|
||
index for index, statement in enumerate(normalized) if statement.startswith('insert into sys_menu values ')
|
||
]
|
||
self.assertEqual(1, len(menu_seed_inserts))
|
||
menu_seed_insert = menu_seed_inserts[0]
|
||
|
||
ordered_markers = (
|
||
normalized.index(MYSQL_MENU_SQL_MODE_SAVE),
|
||
normalized.index(MYSQL_MENU_SQL_MODE_ENFORCE),
|
||
start_transaction,
|
||
permanent_deletes[0],
|
||
permanent_deletes[-1],
|
||
menu_seed_insert,
|
||
commit,
|
||
normalized.index(MYSQL_MENU_SQL_MODE_RESTORE),
|
||
)
|
||
self.assertEqual(tuple(sorted(ordered_markers)), ordered_markers)
|
||
|
||
permanent_mutations = [*permanent_deletes, menu_seed_insert]
|
||
self.assertTrue(all(start_transaction < index < commit for index in permanent_mutations))
|
||
self.assertTrue(
|
||
all(
|
||
index < start_transaction
|
||
for index, statement in enumerate(normalized)
|
||
if re.match(r'(?:create|drop) temporary table\b', statement)
|
||
)
|
||
)
|
||
transaction_sql = '; '.join(normalized[start_transaction + 1 : commit])
|
||
self.assertNotRegex(transaction_sql, r'\b(?:create|drop|alter|truncate)\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')
|
||
|
||
def test_patch_cleanup_fails_closed_before_deleting_legacy_seed_ids(self) -> None:
|
||
sql = _strip_sql_comments(PATCH_PATH.read_text(encoding='utf-8')).replace('`', '').lower()
|
||
guard_create = re.search(
|
||
r'create\s+temporary\s+table\s+pallet_workflow_legacy_seed_guard\b',
|
||
sql,
|
||
)
|
||
guard_insert = re.search(
|
||
r'insert\s+into\s+pallet_workflow_legacy_seed_guard\b[\s\S]+?;',
|
||
sql,
|
||
)
|
||
guard_drops = list(
|
||
re.finditer(
|
||
r'drop\s+temporary\s+table(?:\s+if\s+exists)?\s+pallet_workflow_legacy_seed_guard\b',
|
||
sql,
|
||
)
|
||
)
|
||
guard_drop = guard_drops[-1] if guard_drops else None
|
||
role_cleanup = re.search(r'delete\s+from\s+sys_role_menu\b', sql)
|
||
for statement in (guard_create, guard_insert, guard_drop, role_cleanup):
|
||
self.assertIsNotNone(statement)
|
||
self.assertLess(guard_create.start(), guard_insert.start())
|
||
self.assertLess(guard_insert.start(), guard_drop.start())
|
||
self.assertLess(guard_drop.start(), role_cleanup.start())
|
||
|
||
guard_sql = guard_insert.group()
|
||
expected_menu_identities = (
|
||
(4, 'ai 管理', 0, 'ai', None, '', 'm'),
|
||
(99, '施尔丰仓库', 0, 'https://git.yujiangrubber.cn/niuniu/shierfeng', None, '', 'm'),
|
||
(118, '模型管理', 4, 'model', 'ai/model/index', 'ai:model:list', 'c'),
|
||
(119, 'ai 对话', 4, 'chat', 'ai/chat/index', 'ai:chat:list', 'c'),
|
||
(1061, '模型查询', 118, '#', '', 'ai:model:query', 'f'),
|
||
(1062, '模型新增', 118, '#', '', 'ai:model:add', 'f'),
|
||
(1063, '模型修改', 118, '#', '', 'ai:model:edit', 'f'),
|
||
(1064, '模型删除', 118, '#', '', 'ai:model:remove', 'f'),
|
||
)
|
||
for menu_id, name, parent_id, path, component, permission, menu_type in expected_menu_identities:
|
||
with self.subTest(menu_id=menu_id):
|
||
component_pattern = (
|
||
r'component\s*<=>\s*null' if component is None else rf"component\s*<=>\s*'{re.escape(component)}'"
|
||
)
|
||
identity_pattern = (
|
||
rf'menu_id\s*=\s*{menu_id}\b[\s\S]+?'
|
||
rf"menu_name\s*<=>\s*'{re.escape(name)}'[\s\S]+?"
|
||
rf'parent_id\s*<=>\s*{parent_id}\b[\s\S]+?'
|
||
rf"path\s*<=>\s*'{re.escape(path)}'[\s\S]+?"
|
||
rf'{component_pattern}[\s\S]+?'
|
||
rf"perms\s*<=>\s*'{re.escape(permission)}'[\s\S]+?"
|
||
rf"menu_type\s*<=>\s*'{menu_type}'"
|
||
)
|
||
self.assertRegex(guard_sql, identity_pattern)
|
||
|
||
self.assertRegex(
|
||
guard_sql,
|
||
r'from\s+sys_dict_data[\s\S]+?dict_code\s+between\s+33\s+and\s+69'
|
||
r"[\s\S]+?dict_type\s*<=>\s*'ai_provider_type'",
|
||
)
|
||
self.assertRegex(
|
||
guard_sql,
|
||
r'from\s+sys_dict_type[\s\S]+?dict_id\s*=\s*12\b'
|
||
r"[\s\S]+?dict_type\s*<=>\s*'ai_provider_type'",
|
||
)
|
||
|
||
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)
|
||
self.assertRegex(
|
||
dict_data_cleanup.group(),
|
||
r'dict_code\s+between\s+33\s+and\s+69\s+and\s+dict_type\s*=\s*\'ai_provider_type\'',
|
||
)
|
||
self.assertRegex(
|
||
dict_type_cleanup.group(),
|
||
r'dict_id\s*=\s*12\s+and\s+dict_type\s*=\s*\'ai_provider_type\'',
|
||
)
|
||
|
||
|
||
if __name__ == '__main__':
|
||
unittest.main()
|