niuniu/ShiErFeng
4
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

为基础资料维护页面增加导入导出

Browse Source
This commit is contained in:
yely
2026-10-09 21:13:43 +08:00
parent a62803ccf0
commit a8af1f542b
20 changed files with 2134 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files
shierfeng-fastapi-backend/tests/module_business/test_api_permission_contract.py
+36 -8
View File
@@ -29,6 +29,7 @@ CONTROLLER_NAMES = (
'test_batch',
'dashboard',
'tunnel',
'master_excel',
)
TARGET_PREFIXES = (
'/master',
@@ -85,10 +86,24 @@ EXPECTED_BUTTON_PERMISSIONS = frozenset(
'master:pallet:add',
'master:pallet:edit',
'master:pallet:print',
'master:strain:import',
'master:strain:export',
'master:customer:import',
'master:customer:export',
'master:site:import',
'master:site:export',
'master:line:import',
'master:line:export',
'master:warehouse:import',
'master:warehouse:export',
'master:pallet:import',
'master:pallet:export',
'master:tunnel:query',
'master:tunnel:add',
'master:tunnel:edit',
'master:tunnel:remove',
'master:tunnel:import',
'master:tunnel:export',
'business:inbound:query',
'business:inbound:add',
'business:inbound:edit',
@@ -142,6 +157,7 @@ VUE_PAGE_PATHS = (
'src/views/report/consumption/index.vue',
'src/views/report/return/index.vue',
'src/views/report/palletTrace/index.vue',
'src/components/MasterExcelActions/index.vue',
)
PERMISSION_LITERAL_PATTERN = re.compile(r'(?:master|business|report|system):[a-z]+:[a-z]+')
UNGUARDED_AUTHENTICATED_ROUTES = {
@@ -196,6 +212,16 @@ def _build_expected_route_permissions() -> dict[tuple[str, str], str]:
('/master/tunnel/{tunnel_ids}', 'DELETE'): 'master:tunnel:remove',
}
)
for resource in ('strain', 'customer', 'site', 'line', 'warehouse', 'pallet', 'tunnel'):
root = f'/master/{resource}'
permission_root = f'master:{resource}'
expected.update(
{
(f'{root}/importData', 'POST'): f'{permission_root}:import',
(f'{root}/importTemplate', 'POST'): f'{permission_root}:import',
(f'{root}/export', 'POST'): f'{permission_root}:export',
}
)
expected.update(
{
@@ -323,7 +349,7 @@ def _auth_dependencies(route: APIRoute) -> list[CheckUserInterfaceAuth]:
def _assert_controller_permission_literals(test_case: unittest.TestCase, literals: list[str]) -> None:
expected = Counter(EXPECTED_ROUTE_PERMISSIONS.values())
test_case.assertEqual(98, expected.total())
test_case.assertEqual(119, expected.total())
test_case.assertEqual(expected, Counter(literals))
test_case.assertEqual(EXPECTED_PERMISSIONS, frozenset(expected))
@@ -475,7 +501,7 @@ class ApiPermissionContractTest(unittest.TestCase):
}
self.assertEqual(set(EXPECTED_ROUTE_PERMISSIONS) | UNGUARDED_AUTHENTICATED_ROUTES, actual_operations)
self.assertEqual(100, len(actual_operations))
self.assertEqual(121, len(actual_operations))
openapi_ai_paths = sorted(path for path in self.schema['paths'] if path == '/ai' or path.startswith('/ai/'))
app_ai_paths = sorted(
@@ -529,9 +555,9 @@ class ApiPermissionContractTest(unittest.TestCase):
self.assertEqual([], malformed_calls)
self.assertEqual(19, len(EXPECTED_PAGE_PERMISSIONS))
self.assertEqual(48, len(EXPECTED_BUTTON_PERMISSIONS))
self.assertEqual(62, len(EXPECTED_BUTTON_PERMISSIONS))
self.assertFalse(EXPECTED_PAGE_PERMISSIONS & EXPECTED_BUTTON_PERMISSIONS)
self.assertEqual(67, len(EXPECTED_PERMISSIONS))
self.assertEqual(81, len(EXPECTED_PERMISSIONS))
_assert_controller_permission_literals(self, literals)
self.assertEqual(
{
@@ -541,6 +567,8 @@ class ApiPermissionContractTest(unittest.TestCase):
'close',
'confirm',
'edit',
'export',
'import',
'list',
'print',
'purge',
@@ -554,7 +582,7 @@ class ApiPermissionContractTest(unittest.TestCase):
def test_controller_ast_permission_contract_rejects_duplicate_call(self) -> None:
literals = list(EXPECTED_ROUTE_PERMISSIONS.values())
self.assertEqual(98, len(literals))
self.assertEqual(119, len(literals))
literals.append(literals[0])
with self.assertRaises(AssertionError):
@@ -572,8 +600,8 @@ class ApiPermissionContractTest(unittest.TestCase):
def test_vue_pages_have_exact_guarded_permission_contract(self) -> None:
actual_permissions = set()
self.assertEqual(18, len(VUE_PAGE_PATHS))
self.assertEqual(18, len(set(VUE_PAGE_PATHS)))
self.assertEqual(19, len(VUE_PAGE_PATHS))
self.assertEqual(19, len(set(VUE_PAGE_PATHS)))
for relative_path in VUE_PAGE_PATHS:
source = (FRONTEND_ROOT / relative_path).read_text(encoding='utf-8')
guarded = _extract_guarded_vue_permissions(source)
@@ -758,7 +786,7 @@ class ApiPermissionContractTest(unittest.TestCase):
for method in route.methods & WRITE_METHODS:
actual_write_routes[(route.path_format, method)] = route
self.assertEqual(46, len(expected_write_permissions))
self.assertEqual(67, len(expected_write_permissions))
self.assertEqual(set(expected_write_permissions), set(actual_write_routes))
for key, expected_permission in expected_write_permissions.items():
with self.subTest(operation=key):
shierfeng-fastapi-backend/tests/module_business/test_master_excel_service.py
+320
View File
@@ -0,0 +1,320 @@
import io
import unittest
from types import SimpleNamespace
from unittest.mock import AsyncMock, patch
from openpyxl import Workbook, load_workbook
from sqlalchemy.exc import IntegrityError
from sqlalchemy.ext.asyncio import AsyncSession
from exceptions.exception import PermissionException, ServiceException
from module_business.dao.master_dao import MasterDao
from module_business.domain.test_batch_scope import TestBatchScope
from module_business.entity.do.master_do import BizPallet
from module_business.entity.do.operation_do import BizPalletInventory
from module_business.entity.vo.master_vo import StrainModel, StrainPageQueryModel
from module_business.service.master_excel_service import MasterExcelService, _LISTERS
from module_business.service.master_service import MasterService
from tests.module_business.test_batch_fixtures import make_admin, make_current_user, make_test_scope
class _Upload:
def __init__(self, content: bytes) -> None:
self.content = content
self.closed = False
async def read(self) -> bytes:
return self.content
async def close(self) -> None:
self.closed = True
def workbook_bytes(headers: list[str], rows: list[list[object]]) -> bytes:
workbook = Workbook()
worksheet = workbook.active
worksheet.append(headers)
for row in rows:
worksheet.append(row)
buffer = io.BytesIO()
workbook.save(buffer)
return buffer.getvalue()
def sheet_rows(binary: bytes) -> list[tuple[object, ...]]:
workbook = load_workbook(io.BytesIO(binary), data_only=False)
try:
return [tuple(cell.value for cell in row) for row in workbook.active.iter_rows()]
finally:
workbook.close()
def compile_sql(statement: object) -> str:
return ' '.join(str(statement.compile()).lower().split())
class MasterExcelServiceTest(unittest.IsolatedAsyncioTestCase):
def setUp(self) -> None:
self.db = AsyncMock(spec=AsyncSession)
self.admin = make_admin()
self.scope = TestBatchScope.formal()
self.executed: list[object] = []
async def execute(statement: object) -> None:
self.executed.append(statement)
self.db.execute = execute
async def _import(self, resource: str, headers: list[str], rows: list[list[object]], *, update_support: bool = False, scope=None, user=None):
return await MasterExcelService.import_resource(
self.db,
resource, # type: ignore[arg-type]
_Upload(workbook_bytes(headers, rows)),
update_support,
user or self.admin,
scope or self.scope,
)
async def test_add_skip_update_and_other_scope(self) -> None:
headers = ['菌种编码', '菌种名称', '状态', '备注']
with patch.object(MasterDao, 'find_strain_by_code', new=AsyncMock(return_value=None)), patch.object(
MasterDao, 'add_strain', new=AsyncMock()
) as add_strain:
added = await self._import('strain', headers, [[1001, '香菇', '正常', '首批']])
self.assertIn('新增1条', added.message)
saved = add_strain.await_args.args[1]
self.assertEqual('1001', saved.strain_code)
self.assertEqual('香菇', saved.strain_name)
self.assertEqual('0', saved.status)
self.assertIsNone(add_strain.await_args.args[2])
self.db.commit.assert_awaited_once()
self.db.commit.reset_mock()
existing = SimpleNamespace(strain_id=3, strain_code='S001', test_batch_id=None)
with patch.object(MasterDao, 'find_strain_by_code', new=AsyncMock(return_value=existing)), patch.object(
MasterDao, 'get_strain', new=AsyncMock(return_value=existing)
), patch.object(MasterDao, 'update_strain', new=AsyncMock()) as update_strain:
skipped = await self._import('strain', headers, [['S001', '香菇', '正常', '']])
self.assertIn('跳过1条', skipped.message)
self.assertIn('已跳过', skipped.message)
update_strain.assert_not_awaited()
self.db.rollback.assert_awaited()
self.db.commit.assert_not_awaited()
self.db.rollback.reset_mock()
with patch.object(MasterDao, 'find_strain_by_code', new=AsyncMock(return_value=existing)), patch.object(
MasterDao, 'get_strain', new=AsyncMock(return_value=existing)
), patch.object(MasterDao, 'update_strain', new=AsyncMock()) as update_strain:
updated = await self._import('strain', headers, [['S001', '新香菇', '停用', '改备注']], update_support=True)
self.assertIn('更新1条', updated.message)
model = update_strain.await_args.args[1]
self.assertEqual(3, model.strain_id)
self.assertEqual('S001', model.strain_code)
self.assertEqual('新香菇', model.strain_name)
self.assertEqual('1', model.status)
self.db.commit.assert_awaited_once()
self.db.commit.reset_mock()
outside = SimpleNamespace(strain_id=9, strain_code='S009', test_batch_id=4)
with patch.object(MasterDao, 'find_strain_by_code', new=AsyncMock(return_value=outside)), patch.object(
MasterDao, 'get_strain', new=AsyncMock(return_value=None)
), patch.object(MasterDao, 'add_strain', new=AsyncMock()) as add_strain:
rejected = await self._import('strain', headers, [['S009', '越界', '正常', '']])
self.assertIn('已存在于其他数据范围', rejected.message)
add_strain.assert_not_awaited()
self.db.commit.assert_not_awaited()
self.db.rollback.assert_awaited()
async def test_reordered_headers_and_row_error_still_commit_success_rows(self) -> None:
headers = ['备注', '状态', '菌种名称', '菌种编码']
with patch.object(MasterDao, 'find_strain_by_code', new=AsyncMock(return_value=None)), patch.object(
MasterDao, 'add_strain', new=AsyncMock()
) as add_strain:
result = await self._import('strain', headers, [['备注', '正常', '香菇', 'S001'], ['坏', '启用', '平菇', 'S002']])
self.assertEqual(1, add_strain.await_count)
self.assertIn('新增1条', result.message)
self.assertIn('第3行', result.message)
self.assertIn('状态只能填写正常或停用', result.message)
self.db.commit.assert_awaited_once()
async def test_duplicate_identity_rejects_whole_file_before_write(self) -> None:
headers = ['菌种编码', '菌种名称', '状态', '备注']
with patch.object(MasterDao, 'find_strain_by_code', new=AsyncMock()) as find_strain:
with self.assertRaises(ServiceException) as caught:
await self._import('strain', headers, [['S001', '甲', '正常', ''], ['s001', '乙', '停用', '']])
self.assertIn('重复编码', caught.exception.message)
find_strain.assert_not_awaited()
self.db.commit.assert_not_awaited()
self.db.rollback.assert_not_awaited()
async def test_header_mismatch_empty_file_and_size_limit(self) -> None:
with self.assertRaises(ServiceException) as mismatch:
await self._import('strain', ['菌种编码', '多余'], [['S001', '甲']])
self.assertIn('表头不匹配', mismatch.exception.message)
with self.assertRaises(ServiceException) as empty:
await MasterExcelService.import_resource(
self.db, 'strain', _Upload(b''), False, self.admin, self.scope
)
self.assertEqual('导入文件不能为空', empty.exception.message)
with self.assertRaises(ServiceException) as oversized:
await MasterExcelService.import_resource(
self.db, 'strain', _Upload(b'x' * (5 * 1024 * 1024 + 1)), False, self.admin, self.scope
)
self.assertIn('5MB', oversized.exception.message)
self.db.commit.assert_not_awaited()
async def test_missing_parent_and_ambiguous_site_code_are_row_errors(self) -> None:
site_headers = ['客户编码', '现场编码', '现场名称', '现场地址', '状态', '备注']
with patch.object(MasterDao, 'find_customer_by_code', new=AsyncMock(return_value=None)), patch.object(
MasterDao, 'add_site', new=AsyncMock()
) as add_site:
missing = await self._import('site', site_headers, [['C404', 'SITE1', '现场', '地址', '正常', '']])
self.assertIn('目标客户C404不存在', missing.message)
add_site.assert_not_awaited()
line_headers = ['现场编码', '产线编码', '产线名称', '状态', '备注']
sites = [
SimpleNamespace(site_id=1, site_code='SITE1', status='0', test_batch_id=None),
SimpleNamespace(site_id=2, site_code='SITE1', status='0', test_batch_id=None),
]
with patch.object(MasterDao, 'list_sites_by_code', new=AsyncMock(return_value=sites)), patch.object(
MasterDao, 'add_line', new=AsyncMock()
) as add_line:
ambiguous = await self._import('line', line_headers, [['SITE1', 'L001', '一线', '正常', '']])
self.assertIn('现场编码SITE1不唯一', ambiguous.message)
add_line.assert_not_awaited()
self.db.commit.assert_not_awaited()
async def test_database_error_rolls_back_and_permission_error_is_not_wrapped(self) -> None:
headers = ['菌种编码', '菌种名称', '状态', '备注']
with patch.object(MasterDao, 'find_strain_by_code', new=AsyncMock(return_value=None)), patch.object(
MasterDao, 'add_strain', new=AsyncMock(side_effect=IntegrityError('statement', {}, RuntimeError('duplicate')))
):
with self.assertRaises(ServiceException) as caught:
await self._import('strain', headers, [['S001', '香菇', '正常', '']])
self.assertEqual('导入失败，数据已全部回滚', caught.exception.message)
self.db.rollback.assert_awaited()
self.db.commit.assert_not_awaited()
self.db.rollback.reset_mock()
with self.assertRaises(PermissionException) as denied:
await self._import(
'strain',
headers,
[['S001', '香菇', '正常', '']],
scope=make_test_scope(9),
user=make_current_user(admin=False),
)
self.assertEqual('仅超级管理员可执行测试批次管理操作', denied.exception.message)
self.db.rollback.assert_awaited()
async def test_import_uses_current_test_batch(self) -> None:
headers = ['菌种编码', '菌种名称', '状态', '备注']
scope = make_test_scope(9)
with patch.object(MasterService, '_prepare_test_write', new=AsyncMock()) as prepare, patch.object(
MasterDao, 'find_strain_by_code', new=AsyncMock(return_value=None)
), patch.object(MasterDao, 'add_strain', new=AsyncMock()) as add_strain:
await self._import('strain', headers, [['S001', '香菇', '正常', '']], scope=scope, user=self.admin)
prepare.assert_awaited_once()
self.assertIs(scope, prepare.await_args.args[2])
self.assertEqual(9, add_strain.await_args.args[2])
async def test_new_pallet_gets_fixed_capacity_qr_and_empty_inventory(self) -> None:
async def flush() -> None:
for call in self.db.add.call_args_list:
record = call.args[0]
if isinstance(record, BizPallet) and record.pallet_id is None:
record.pallet_id = 15
self.db.flush = flush
with patch.object(MasterDao, 'find_pallet_by_code', new=AsyncMock(return_value=None)):
result = await self._import('pallet', ['托盘编号', '状态', '备注'], [['P001', '停用', '新盘']])
added = [call.args[0] for call in self.db.add.call_args_list]
pallet = next(record for record in added if isinstance(record, BizPallet))
inventory = next(record for record in added if isinstance(record, BizPalletInventory))
self.assertEqual(100, pallet.capacity_packages)
self.assertEqual('1', pallet.asset_status)
self.assertTrue(pallet.qr_token)
self.assertNotEqual('P001', pallet.qr_token)
self.assertEqual(15, inventory.pallet_id)
self.assertEqual(0, inventory.remaining_packages)
self.assertEqual('UNSTOCKED', inventory.flow_state)
self.assertIsNone(inventory.test_batch_id)
self.assertIn('新增1条', result.message)
self.db.commit.assert_awaited_once()
async def test_pallet_update_keeps_capacity_and_qr_token(self) -> None:
pallet = SimpleNamespace(
pallet_id=5,
pallet_code='P001',
qr_token='qr-keep',
capacity_packages=80,
asset_status='0',
remark='旧',
test_batch_id=None,
)
inventory = SimpleNamespace(pallet_id=5, remaining_packages=3, test_batch_id=None)
with patch.object(MasterDao, 'find_pallet_by_code', new=AsyncMock(return_value=pallet)), patch.object(
MasterDao, 'get_pallet', new=AsyncMock(return_value=pallet)
), patch.object(MasterDao, 'get_pallet_inventory', new=AsyncMock(return_value=inventory)):
result = await self._import(
'pallet', ['托盘编号', '状态', '备注'], [['P001', '停用', '新备注']], update_support=True
)
self.assertIn('更新1条', result.message)
self.assertEqual('qr-keep', pallet.qr_token)
self.assertEqual(80, pallet.capacity_packages)
sql = compile_sql(self.executed[-1])
self.assertNotIn('qr_token', sql)
self.assertNotIn('pallet_code', sql)
self.assertIn('capacity_packages', sql)
self.assertIn('asset_status', sql)
self.db.commit.assert_awaited_once()
async def test_tunnel_update_sql_does_not_touch_current_trace(self) -> None:
site = SimpleNamespace(site_id=2, site_code='SITE1', status='0', test_batch_id=None)
tunnel = SimpleNamespace(tunnel_id=8, tunnel_code='T001', site_id=2, status='0', test_batch_id=None)
with patch.object(MasterDao, 'list_sites_by_code', new=AsyncMock(return_value=[site])), patch.object(
MasterDao, 'find_tunnel_by_code', new=AsyncMock(return_value=tunnel)
), patch.object(MasterDao, 'get_tunnel', new=AsyncMock(return_value=tunnel)), patch.object(
MasterDao, 'get_site', new=AsyncMock(return_value=site)
):
result = await self._import(
'tunnel',
['现场编码', '隧道编号', '隧道名称', '状态', '备注'],
[['SITE1', 'T001', '新隧道', '停用', '改备注']],
update_support=True,
)
self.assertIn('更新1条', result.message)
sql = compile_sql(self.executed[-1])
self.assertNotIn('current_trace_no', sql)
self.assertNotIn('current_trace_started_at', sql)
self.assertIn('tunnel_name', sql)
self.assertIn('status', sql)
self.db.commit.assert_awaited_once()
async def test_export_uses_chinese_status_and_empty_result_keeps_headers(self) -> None:
rows = [
StrainModel(strainId=1, strainCode='S001', strainName='香菇', status='0', remark='甲'),
StrainModel(strainId=2, strainCode='S002', strainName='平菇', status='1', remark=None),
]
query = StrainPageQueryModel()
list_strain = AsyncMock(return_value=rows)
with patch.dict(_LISTERS, {'strain': list_strain}):
binary = await MasterExcelService.export_resource(self.db, 'strain', query, self.scope)
self.assertFalse(list_strain.await_args.kwargs['is_page'])
self.assertEqual(
[
('菌种编码', '菌种名称', '状态', '备注'),
('S001', '香菇', '正常', '甲'),
('S002', '平菇', '停用', None),
],
[row[:4] for row in sheet_rows(binary)],
)
with patch.dict(_LISTERS, {'strain': AsyncMock(return_value=[])}):
empty = await MasterExcelService.export_resource(self.db, 'strain', query, self.scope)
exported = sheet_rows(empty)
self.assertEqual([('菌种编码', '菌种名称', '状态', '备注')], [row[:4] for row in exported])
self.assertEqual(1, len(exported))
shierfeng-fastapi-backend/tests/module_business/test_sql_contract.py
+65 -4
View File
@@ -11,6 +11,7 @@ MENU_PROMOTION_PATCH_PATH = SQL_ROOT / 'patches/20260727_promote_business_menus_
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
@@ -798,7 +799,27 @@ LEGACY_PATCH_BUSINESS_MENU_FIELDS = LEGACY_EXPECTED_BUSINESS_MENU_FIELDS | {
# 同样在菜单旧快照生成后加入收货权限，避免改变历史迁移的权限清单。
EXPECTED_BUTTONS[1119] = ('收货核对', 510, 6, 'business:shipment:receive')
EXPECTED_BUSINESS_MENU_FIELDS[1119] = (510, '收货核对', '', '', 6, 'business:shipment:receive', '')
EXPECTED_BUTTON_PERMISSIONS |= {'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),
@@ -842,6 +863,7 @@ BUSINESS_MENU_IDS = (
| {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 = (
@@ -1508,6 +1530,45 @@ class ShipmentReceiptSqlContractTest(unittest.TestCase):
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 = (
@@ -1615,10 +1676,10 @@ class SqlContractTest(unittest.TestCase):
def test_initializers_define_fixed_business_menus_and_permissions(self) -> None:
self.assertEqual(19, len(EXPECTED_PAGE_PERMISSIONS))
self.assertEqual(47, len(EXPECTED_BUTTON_PERMISSIONS))
self.assertEqual(61, len(EXPECTED_BUTTON_PERMISSIONS))
self.assertFalse(EXPECTED_PAGE_PERMISSIONS & EXPECTED_BUTTON_PERMISSIONS)
self.assertEqual(66, len(EXPECTED_PERMISSIONS))
self.assertEqual(69, len(EXPECTED_BUSINESS_MENU_FIELDS))
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')