yujiangandClaude Opus 5.5
3050ccdc4e
为 Web 端改版提供名称、统计和看板数据。只新增只读查询、扩展列表与详情 返回字段；写接口、托盘状态机和幂等键规则均未改动，未新增权限标识。 - 基础资料：列表补充名称与汇总，客户详情返回现场树，托盘详情返回最近流转， 托盘列表支持按流转阶段筛选和二维码精确匹配；新增现场、产线、隧道的上级 资料选项接口；Excel 导入支持 preview 只校验不写入 - 发货出库：列表支持 keyword、receiptState 筛选并补充名称、托盘数、包数和 菌种构成；详情明细补充托盘号、菌种、批次 - 首次入库：列表与详情补充名称、托盘数、托盘当前状态；新增 /summary - 托盘退回：列表支持 keyword 并补充名称与托盘汇总，详情补充托盘号和隧道； 新增 /summary、/site-options - 工厂装填：列表支持 keyword 并补充托盘号、仓库、操作人；新增 /summary、 /batches、/batches/detail - 客户现场使用：列表支持 keyword、fifoSkipped 并补充名称和被跳过的先到托盘号； 新增 /site-options、/site-board 以及 usage-fifo 的 /queue - 先到先用 DAO 抽出公共候选条件与排序，建议查询生成的 SQL 与改动前一致 - 新增各汇总 DAO 的单元测试，同步更新路由权限与模型字段契约测试 Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
826 lines
36 KiB
Python
826 lines
36 KiB
Python
import ast
|
|
import re
|
|
import unittest
|
|
from collections import Counter
|
|
from pathlib import Path
|
|
|
|
from fastapi.routing import APIRoute
|
|
|
|
from common.aspect.interface_auth import CheckUserInterfaceAuth
|
|
from server import create_app
|
|
|
|
BACKEND_ROOT = Path(__file__).resolve().parents[2]
|
|
FRONTEND_ROOT = BACKEND_ROOT.parent / 'shierfeng-fastapi-frontend'
|
|
CONTROLLER_NAMES = (
|
|
'strain',
|
|
'customer',
|
|
'site',
|
|
'line',
|
|
'warehouse',
|
|
'pallet',
|
|
'inbound',
|
|
'filling',
|
|
'shipment',
|
|
'shipment_receipt',
|
|
'usage',
|
|
'usage_fifo',
|
|
'return',
|
|
'report',
|
|
'test_batch',
|
|
'dashboard',
|
|
'tunnel',
|
|
'master_excel',
|
|
)
|
|
TARGET_PREFIXES = (
|
|
'/master',
|
|
'/business',
|
|
'/report',
|
|
'/system/test-batch',
|
|
)
|
|
ROUTE_HTTP_METHODS = frozenset({'DELETE', 'GET', 'HEAD', 'OPTIONS', 'PATCH', 'POST', 'PUT', 'TRACE'})
|
|
WRITE_METHODS = frozenset({'DELETE', 'PATCH', 'POST', 'PUT'})
|
|
|
|
EXPECTED_PAGE_PERMISSIONS = frozenset(
|
|
{
|
|
'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',
|
|
'system:testbatch: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: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',
|
|
'business:inbound:confirm',
|
|
'business:filling:query',
|
|
'business:filling:add',
|
|
'business:shipment:query',
|
|
'business:shipment:add',
|
|
'business:shipment:edit',
|
|
'business:shipment:confirm',
|
|
'business:shipment:receive',
|
|
'business:usage:query',
|
|
'business:usage:add',
|
|
'business:return:query',
|
|
'business:return:request',
|
|
'business:return:confirm',
|
|
'system:testbatch:query',
|
|
'system:testbatch:add',
|
|
'system:testbatch:edit',
|
|
'system:testbatch:assign',
|
|
'system:testbatch:close',
|
|
'system:testbatch:adopt',
|
|
'system:testbatch:purge',
|
|
}
|
|
)
|
|
EXPECTED_PERMISSIONS = EXPECTED_PAGE_PERMISSIONS | EXPECTED_BUTTON_PERMISSIONS
|
|
EXPECTED_WEB_PERMISSIONS = frozenset(
|
|
permission
|
|
for permission in EXPECTED_PERMISSIONS
|
|
if (
|
|
(not permission.startswith('system:testbatch:') or permission == 'system:testbatch:adopt')
|
|
and permission != 'master:tunnel:remove'
|
|
)
|
|
)
|
|
VUE_PAGE_PATHS = (
|
|
'src/views/master/strain/index.vue',
|
|
'src/views/master/customer/index.vue',
|
|
'src/views/master/site/index.vue',
|
|
'src/views/master/line/index.vue',
|
|
'src/views/master/warehouse/index.vue',
|
|
'src/views/master/pallet/index.vue',
|
|
'src/views/master/tunnel/index.vue',
|
|
'src/views/business/inbound/index.vue',
|
|
'src/views/business/filling/index.vue',
|
|
'src/views/business/shipment/index.vue',
|
|
'src/views/business/usage/index.vue',
|
|
'src/views/business/return/index.vue',
|
|
'src/views/report/palletInventory/index.vue',
|
|
'src/views/report/filling/index.vue',
|
|
'src/views/report/shipment/index.vue',
|
|
'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 = {
|
|
('/system/test-batch/options', 'GET'),
|
|
('/business/dashboard/scan-exceptions', 'GET'),
|
|
}
|
|
REMOVED_DELETE_SYMBOLS = (
|
|
'delStrain',
|
|
'delCustomer',
|
|
'delSite',
|
|
'delLine',
|
|
'delWarehouse',
|
|
'delPallet',
|
|
'delInbound',
|
|
'delShipment',
|
|
)
|
|
REMOVED_DELETE_PERMISSIONS = frozenset(
|
|
{
|
|
'master:strain:remove',
|
|
'master:customer:remove',
|
|
'master:site:remove',
|
|
'master:line:remove',
|
|
'master:warehouse:remove',
|
|
'master:pallet:remove',
|
|
'business:inbound:remove',
|
|
'business:shipment:remove',
|
|
}
|
|
)
|
|
|
|
|
|
def _build_expected_route_permissions() -> dict[tuple[str, str], str]:
|
|
expected: dict[tuple[str, str], str] = {}
|
|
for resource in ('strain', 'customer', 'site', 'line', 'warehouse', 'pallet'):
|
|
root = f'/master/{resource}'
|
|
permission_root = f'master:{resource}'
|
|
expected.update(
|
|
{
|
|
(f'{root}/list', 'GET'): f'{permission_root}:list',
|
|
(f'{root}/{{{resource}_id}}', 'GET'): f'{permission_root}:query',
|
|
(root, 'POST'): f'{permission_root}:add',
|
|
(root, 'PUT'): f'{permission_root}:edit',
|
|
(f'{root}/{{{resource}_ids}}', 'DELETE'): f'{permission_root}:list',
|
|
}
|
|
)
|
|
|
|
expected.update(
|
|
{
|
|
('/master/tunnel/list', 'GET'): 'master:tunnel:list',
|
|
('/master/tunnel/{tunnel_id}', 'GET'): 'master:tunnel:query',
|
|
('/master/tunnel', 'POST'): 'master:tunnel:add',
|
|
('/master/tunnel', 'PUT'): 'master:tunnel:edit',
|
|
('/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(
|
|
{
|
|
('/master/site/{site_id}/users', 'GET'): 'master:site:assign',
|
|
('/master/site/{site_id}/users', 'PUT'): 'master:site:assign',
|
|
('/master/warehouse/{warehouse_id}/users', 'GET'): 'master:warehouse:assign',
|
|
('/master/warehouse/{warehouse_id}/users', 'PUT'): 'master:warehouse:assign',
|
|
('/master/pallet/{pallet_id}/qrcode', 'GET'): 'master:pallet:print',
|
|
('/master/site/customer-options', 'GET'): 'master:site:list',
|
|
('/master/line/site-options', 'GET'): 'master:line:list',
|
|
('/master/tunnel/site-options', 'GET'): 'master:tunnel:list',
|
|
('/business/inbound/list', 'GET'): 'business:inbound:list',
|
|
('/business/inbound/summary', 'GET'): 'business:inbound:list',
|
|
('/business/inbound/warehouse-options', 'GET'): 'business:inbound:query',
|
|
('/business/inbound/scan/{qr_token}', 'GET'): 'business:inbound:query',
|
|
('/business/inbound/{inbound_id}', 'GET'): 'business:inbound:query',
|
|
('/business/inbound', 'POST'): 'business:inbound:add',
|
|
('/business/inbound', 'PUT'): 'business:inbound:edit',
|
|
('/business/inbound/{inbound_ids}', 'DELETE'): 'business:inbound:list',
|
|
('/business/inbound/{inbound_id}/confirm', 'PUT'): 'business:inbound:confirm',
|
|
('/business/filling/list', 'GET'): 'business:filling:list',
|
|
('/business/filling/summary', 'GET'): 'business:filling:list',
|
|
('/business/filling/batches', 'GET'): 'business:filling:list',
|
|
('/business/filling/batches/detail', 'GET'): 'business:filling:query',
|
|
('/business/filling/strain-options', 'GET'): 'business:filling:query',
|
|
('/business/filling/scan/{qr_token}', 'GET'): 'business:filling:query',
|
|
('/business/filling/{filling_id}', 'GET'): 'business:filling:query',
|
|
('/business/filling', 'POST'): 'business:filling:add',
|
|
('/business/shipment/list', 'GET'): 'business:shipment:list',
|
|
('/business/shipment/warehouse-options', 'GET'): 'business:shipment:query',
|
|
('/business/shipment/site-options', 'GET'): 'business:shipment:query',
|
|
('/business/shipment/scan/{qr_token}', 'GET'): 'business:shipment:query',
|
|
('/business/shipment/{shipment_id}', 'GET'): 'business:shipment:query',
|
|
('/business/shipment/{shipment_id}/labels', 'GET'): 'business:shipment:query',
|
|
('/business/shipment-receipt/list', 'GET'): 'business:shipment:receive',
|
|
('/business/shipment-receipt/{shipment_id}', 'GET'): 'business:shipment:receive',
|
|
('/business/shipment-receipt/{shipment_id}/scan/{identifier}', 'GET'): 'business:shipment:receive',
|
|
('/business/shipment-receipt/{shipment_id}/confirm', 'PUT'): 'business:shipment:receive',
|
|
('/business/usage-fifo/recommendation', 'GET'): 'business:usage:query',
|
|
('/business/usage-fifo/queue', 'GET'): 'business:usage:query',
|
|
('/business/shipment', 'POST'): 'business:shipment:add',
|
|
('/business/shipment', 'PUT'): 'business:shipment:edit',
|
|
('/business/shipment/{shipment_ids}', 'DELETE'): 'business:shipment:list',
|
|
('/business/shipment/{shipment_id}/confirm', 'PUT'): 'business:shipment:confirm',
|
|
('/business/usage/list', 'GET'): 'business:usage:list',
|
|
('/business/usage/line-options', 'GET'): 'business:usage:query',
|
|
('/business/usage/tunnel-options', 'GET'): 'business:usage:query',
|
|
('/business/usage/site-options', 'GET'): 'business:usage:list',
|
|
('/business/usage/site-board', 'GET'): 'business:usage:query',
|
|
('/business/usage/enter-tunnel', 'POST'): 'business:usage:add',
|
|
('/business/usage/scan/{qr_token}', 'GET'): 'business:usage:query',
|
|
('/business/usage/{usage_id}', 'GET'): 'business:usage:query',
|
|
('/business/usage', 'POST'): 'business:usage:add',
|
|
('/business/return/list', 'GET'): 'business:return:list',
|
|
('/business/return/summary', 'GET'): 'business:return:list',
|
|
('/business/return/site-options', 'GET'): 'business:return:list',
|
|
('/business/return/warehouse-options', 'GET'): 'business:return:query',
|
|
('/business/return/scan/{qr_token}', 'GET'): 'business:return:query',
|
|
('/business/return/{return_id}', 'GET'): 'business:return:query',
|
|
('/business/return/request', 'POST'): 'business:return:request',
|
|
('/business/return/{return_id}/confirm', 'PUT'): 'business:return:confirm',
|
|
('/report/pallet-inventory/list', 'GET'): 'report:palletinventory:list',
|
|
('/report/filling/list', 'GET'): 'report:filling:list',
|
|
('/report/shipment/list', 'GET'): 'report:shipment:list',
|
|
('/report/consumption/list', 'GET'): 'report:consumption:list',
|
|
('/report/return/list', 'GET'): 'report:return:list',
|
|
('/report/pallet-trace/list', 'GET'): 'report:pallettrace:list',
|
|
('/system/test-batch/list', 'GET'): 'system:testbatch:list',
|
|
('/system/test-batch', 'POST'): 'system:testbatch:add',
|
|
('/system/test-batch/{batch_id}', 'PUT'): 'system:testbatch:edit',
|
|
('/system/test-batch/{batch_id}/users', 'PUT'): 'system:testbatch:assign',
|
|
('/system/test-batch/{batch_id}/close', 'PUT'): 'system:testbatch:close',
|
|
('/system/test-batch/{batch_id}/reopen', 'PUT'): 'system:testbatch:close',
|
|
('/system/test-batch/{batch_id}/adoption/preview', 'POST'): 'system:testbatch:adopt',
|
|
('/system/test-batch/{batch_id}/adoption/execute', 'POST'): 'system:testbatch:adopt',
|
|
('/system/test-batch/{batch_id}/purge/preview', 'POST'): 'system:testbatch:purge',
|
|
('/system/test-batch/{batch_id}/purge', 'POST'): 'system:testbatch:purge',
|
|
('/system/test-batch/{batch_id}', 'GET'): 'system:testbatch:query',
|
|
}
|
|
)
|
|
return expected
|
|
|
|
|
|
EXPECTED_ROUTE_PERMISSIONS = _build_expected_route_permissions()
|
|
STATIC_READS_BEFORE_DETAIL = {
|
|
'/master/site/{site_id}': ('/master/site/customer-options',),
|
|
'/master/line/{line_id}': ('/master/line/site-options',),
|
|
'/master/tunnel/{tunnel_id}': ('/master/tunnel/site-options',),
|
|
'/business/inbound/{inbound_id}': (
|
|
'/business/inbound/summary',
|
|
'/business/inbound/warehouse-options',
|
|
'/business/inbound/scan/{qr_token}',
|
|
),
|
|
'/business/filling/{filling_id}': (
|
|
'/business/filling/summary',
|
|
'/business/filling/batches',
|
|
'/business/filling/batches/detail',
|
|
'/business/filling/strain-options',
|
|
'/business/filling/scan/{qr_token}',
|
|
),
|
|
'/business/shipment/{shipment_id}': (
|
|
'/business/shipment/warehouse-options',
|
|
'/business/shipment/site-options',
|
|
'/business/shipment/scan/{qr_token}',
|
|
),
|
|
'/business/usage/{usage_id}': (
|
|
'/business/usage/line-options',
|
|
'/business/usage/tunnel-options',
|
|
'/business/usage/site-options',
|
|
'/business/usage/site-board',
|
|
'/business/usage/scan/{qr_token}',
|
|
),
|
|
'/business/return/{return_id}': (
|
|
'/business/return/summary',
|
|
'/business/return/site-options',
|
|
'/business/return/warehouse-options',
|
|
'/business/return/scan/{qr_token}',
|
|
),
|
|
'/system/test-batch/{batch_id}': (
|
|
'/system/test-batch/options',
|
|
'/system/test-batch/list',
|
|
),
|
|
}
|
|
EXACT_QUERY_PARAMETERS = {
|
|
('/business/inbound/scan/{qr_token}', 'GET'): {'warehouseId'},
|
|
('/business/shipment/scan/{qr_token}', 'GET'): {'sourceWarehouseId', 'targetSiteId'},
|
|
('/business/usage/line-options', 'GET'): {'siteId'},
|
|
('/business/usage/tunnel-options', 'GET'): {'siteId'},
|
|
('/business/usage/site-board', 'GET'): {'siteId'},
|
|
('/business/usage-fifo/queue', 'GET'): {'siteId'},
|
|
('/business/return/scan/{qr_token}', 'GET'): {'mode', 'returnId'},
|
|
}
|
|
REQUIRED_QUERY_PARAMETERS = {
|
|
('/business/inbound/list', 'GET'): {'warehouseId'},
|
|
('/business/shipment/list', 'GET'): {'sourceWarehouseId', 'targetSiteId'},
|
|
}
|
|
|
|
|
|
def _is_target_path(path: str) -> bool:
|
|
return any(path == prefix or path.startswith(f'{prefix}/') for prefix in TARGET_PREFIXES)
|
|
|
|
|
|
def _auth_dependencies(route: APIRoute) -> list[CheckUserInterfaceAuth]:
|
|
return [
|
|
dependency.dependency
|
|
for dependency in route.dependencies
|
|
if isinstance(dependency.dependency, CheckUserInterfaceAuth)
|
|
]
|
|
|
|
|
|
def _assert_controller_permission_literals(test_case: unittest.TestCase, literals: list[str]) -> None:
|
|
expected = Counter(EXPECTED_ROUTE_PERMISSIONS.values())
|
|
test_case.assertEqual(131, expected.total())
|
|
test_case.assertEqual(expected, Counter(literals))
|
|
test_case.assertEqual(EXPECTED_PERMISSIONS, frozenset(expected))
|
|
|
|
|
|
def _sanitize_vue_source(source: str) -> tuple[str, str]: # noqa: PLR0912
|
|
code = list(source)
|
|
structure = list(source)
|
|
index = 0
|
|
while index < len(source):
|
|
if source.startswith('<!--', index):
|
|
end = source.find('-->', index + 4)
|
|
if end == -1:
|
|
raise AssertionError('Vue 源码包含未闭合 HTML 注释')
|
|
end += 3
|
|
for offset in range(index, end):
|
|
if source[offset] not in '\r\n':
|
|
code[offset] = structure[offset] = ' '
|
|
index = end
|
|
continue
|
|
|
|
quote = source[index]
|
|
if quote in '\'"`':
|
|
end = index + 1
|
|
while end < len(source) and source[end] != quote:
|
|
end += 2 if source[end] == '\\' else 1
|
|
if end >= len(source):
|
|
raise AssertionError('Vue 源码包含未闭合字符串')
|
|
for offset in range(index, end + 1):
|
|
if source[offset] not in '\r\n':
|
|
structure[offset] = ' '
|
|
index = end + 1
|
|
continue
|
|
|
|
is_line_comment = source.startswith('//', index)
|
|
is_block_comment = source.startswith('/*', index)
|
|
if is_line_comment or is_block_comment:
|
|
marker = '\n' if is_line_comment else '*/'
|
|
end = source.find(marker, index + 2)
|
|
if end == -1 and is_block_comment:
|
|
raise AssertionError('Vue 源码包含未闭合块注释')
|
|
end = len(source) if end == -1 else end + (0 if is_line_comment else 2)
|
|
for offset in range(index, end):
|
|
if source[offset] not in '\r\n':
|
|
code[offset] = structure[offset] = ' '
|
|
index = end
|
|
continue
|
|
index += 1
|
|
return ''.join(code), ''.join(structure)
|
|
|
|
|
|
def _parse_static_permission_array(expression: str) -> list[str]:
|
|
try:
|
|
parsed = ast.parse(expression.strip(), mode='eval').body
|
|
except SyntaxError as exc:
|
|
raise AssertionError('权限门禁必须使用静态字符串数组') from exc
|
|
if not isinstance(parsed, ast.List) or not parsed.elts:
|
|
raise AssertionError('权限门禁必须使用非空静态字符串数组')
|
|
permissions = []
|
|
for element in parsed.elts:
|
|
if not isinstance(element, ast.Constant) or not isinstance(element.value, str):
|
|
raise AssertionError('权限门禁数组禁止动态表达式')
|
|
if PERMISSION_LITERAL_PATTERN.fullmatch(element.value) is None:
|
|
raise AssertionError(f'非法权限字面量: {element.value}')
|
|
permissions.append(element.value)
|
|
return permissions
|
|
|
|
|
|
def _extract_static_has_permi_and(expression: str) -> list[str]:
|
|
code, structure = _sanitize_vue_source(expression)
|
|
permission_tokens = list(re.finditer(r'\bhasPermi(?:And|Or)\b', structure))
|
|
if any(token.group(0) == 'hasPermiOr' for token in permission_tokens):
|
|
raise AssertionError('禁止使用 proxy.$auth.hasPermiOr，权限门禁必须使用约定的静态 AND 语义')
|
|
canonical_call_prefix = r'(?<![\w$.])proxy\.\$auth\.hasPermiAnd\s*\('
|
|
call_tokens = list(re.finditer(canonical_call_prefix, structure))
|
|
if len(permission_tokens) != len(call_tokens):
|
|
raise AssertionError('hasPermiAnd 必须使用 canonical proxy.$auth.hasPermiAnd(static-array) 形式')
|
|
call_pattern = re.compile(canonical_call_prefix + r'\s*(?P<array>\[[^\[\]]*\])\s*\)')
|
|
calls = list(call_pattern.finditer(structure))
|
|
if len(calls) != len(call_tokens):
|
|
raise AssertionError('hasPermiAnd 必须直接传入静态字符串数组')
|
|
|
|
permissions = []
|
|
for call in calls:
|
|
start, end = call.span('array')
|
|
permissions.extend(_parse_static_permission_array(code[start:end]))
|
|
return permissions
|
|
|
|
|
|
def _extract_guarded_vue_permissions(source: str) -> frozenset[str]:
|
|
code, structure = _sanitize_vue_source(source)
|
|
permissions = []
|
|
|
|
directive_tokens = list(re.finditer(r'\bv-hasPermi\b', structure))
|
|
directive_pattern = re.compile(r"""v-hasPermi\s*=\s*(?:"(?P<double>[^"]*)"|'(?P<single>[^']*)')""")
|
|
for token in directive_tokens:
|
|
match = directive_pattern.match(code, token.start())
|
|
if match is None:
|
|
raise AssertionError('v-hasPermi 必须直接绑定静态字符串数组')
|
|
permissions.extend(_parse_static_permission_array(match.group('double') or match.group('single')))
|
|
|
|
permissions.extend(_extract_static_has_permi_and(source))
|
|
v_if_pattern = re.compile(r"""v-if\s*=\s*(?:"(?P<double>[^"]*)"|'(?P<single>[^']*)')""")
|
|
for token in re.finditer(r'\bv-if\b', structure):
|
|
match = v_if_pattern.match(code, token.start())
|
|
if match is None:
|
|
raise AssertionError('v-if 必须使用可静态解析的属性值')
|
|
expression = match.group('double') or match.group('single')
|
|
permissions.extend(_extract_static_has_permi_and(expression))
|
|
return frozenset(permissions)
|
|
|
|
|
|
def _assert_exact_permission_layers(test_case: unittest.TestCase, layers: dict[str, set | frozenset]) -> None:
|
|
for name, actual in layers.items():
|
|
test_case.assertEqual(EXPECTED_PERMISSIONS, frozenset(actual), name)
|
|
|
|
|
|
class ApiPermissionContractTest(unittest.TestCase):
|
|
@classmethod
|
|
def setUpClass(cls) -> None:
|
|
cls.app = create_app()
|
|
cls.schema = cls.app.openapi()
|
|
cls.target_routes = [
|
|
route for route in cls.app.routes if isinstance(route, APIRoute) and _is_target_path(route.path_format)
|
|
]
|
|
|
|
def test_target_path_selector_uses_namespace_boundaries(self) -> None:
|
|
for path in (
|
|
'/master',
|
|
'/master/foo',
|
|
'/business',
|
|
'/business/foo',
|
|
'/report/foo',
|
|
'/system/test-batch',
|
|
'/system/test-batch/options',
|
|
):
|
|
with self.subTest(path=path):
|
|
self.assertTrue(_is_target_path(path))
|
|
for path in ('/masterish', '/business-other', '/reporting', '/system/test-batches'):
|
|
with self.subTest(path=path):
|
|
self.assertFalse(_is_target_path(path))
|
|
|
|
def test_openapi_has_exact_target_operations_and_no_ai_routes(self) -> None:
|
|
actual_operations = {
|
|
(path, method.upper())
|
|
for path, path_item in self.schema['paths'].items()
|
|
if _is_target_path(path)
|
|
for method in path_item
|
|
if method.upper() in ROUTE_HTTP_METHODS
|
|
}
|
|
|
|
self.assertEqual(set(EXPECTED_ROUTE_PERMISSIONS) | UNGUARDED_AUTHENTICATED_ROUTES, actual_operations)
|
|
self.assertEqual(133, 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(
|
|
route.path_format
|
|
for route in self.app.routes
|
|
if getattr(route, 'path', None) == '/ai' or getattr(route, 'path', '').startswith('/ai/')
|
|
)
|
|
self.assertEqual([], openapi_ai_paths)
|
|
self.assertEqual([], app_ai_paths)
|
|
|
|
def test_route_dependencies_have_exact_permissions(self) -> None:
|
|
actual_permissions: dict[tuple[str, str], str] = {}
|
|
operation_keys: list[tuple[str, str]] = []
|
|
for route in self.target_routes:
|
|
methods = route.methods & ROUTE_HTTP_METHODS
|
|
self.assertEqual(1, len(methods), route.path_format)
|
|
key = (route.path_format, next(iter(methods)))
|
|
operation_keys.append(key)
|
|
|
|
auth_dependencies = _auth_dependencies(route)
|
|
if key in UNGUARDED_AUTHENTICATED_ROUTES:
|
|
self.assertEqual([], auth_dependencies, key)
|
|
else:
|
|
self.assertEqual(1, len(auth_dependencies), key)
|
|
actual_permissions[key] = auth_dependencies[0].perm
|
|
|
|
self.assertEqual(len(operation_keys), len(set(operation_keys)))
|
|
self.assertEqual(EXPECTED_ROUTE_PERMISSIONS, actual_permissions)
|
|
self.assertEqual(EXPECTED_PERMISSIONS, frozenset(actual_permissions.values()))
|
|
|
|
def test_controller_ast_has_exact_literal_permission_set(self) -> None:
|
|
literals: list[str] = []
|
|
malformed_calls: list[str] = []
|
|
controller_paths = sorted((BACKEND_ROOT / 'module_business/controller').glob('*_controller.py'))
|
|
actual_controller_names = {source_path.stem.removesuffix('_controller') for source_path in controller_paths}
|
|
self.assertEqual(set(CONTROLLER_NAMES), actual_controller_names)
|
|
|
|
for source_path in controller_paths:
|
|
tree = ast.parse(source_path.read_text(encoding='utf-8'), filename=str(source_path))
|
|
for node in ast.walk(tree):
|
|
if not (
|
|
isinstance(node, ast.Call)
|
|
and isinstance(node.func, ast.Name)
|
|
and node.func.id == 'UserInterfaceAuthDependency'
|
|
):
|
|
continue
|
|
if not (node.args and isinstance(node.args[0], ast.Constant) and isinstance(node.args[0].value, str)):
|
|
malformed_calls.append(f'{source_path.name}:{node.lineno}')
|
|
continue
|
|
literals.append(node.args[0].value)
|
|
|
|
self.assertEqual([], malformed_calls)
|
|
self.assertEqual(19, len(EXPECTED_PAGE_PERMISSIONS))
|
|
self.assertEqual(62, len(EXPECTED_BUTTON_PERMISSIONS))
|
|
self.assertFalse(EXPECTED_PAGE_PERMISSIONS & EXPECTED_BUTTON_PERMISSIONS)
|
|
self.assertEqual(81, len(EXPECTED_PERMISSIONS))
|
|
_assert_controller_permission_literals(self, literals)
|
|
self.assertEqual(
|
|
{
|
|
'add',
|
|
'adopt',
|
|
'assign',
|
|
'close',
|
|
'confirm',
|
|
'edit',
|
|
'export',
|
|
'import',
|
|
'list',
|
|
'print',
|
|
'purge',
|
|
'query',
|
|
'remove',
|
|
'request',
|
|
'receive',
|
|
},
|
|
{permission.rsplit(':', maxsplit=1)[1] for permission in literals},
|
|
)
|
|
|
|
def test_controller_ast_permission_contract_rejects_duplicate_call(self) -> None:
|
|
literals = list(EXPECTED_ROUTE_PERMISSIONS.values())
|
|
self.assertEqual(131, len(literals))
|
|
literals.append(literals[0])
|
|
|
|
with self.assertRaises(AssertionError):
|
|
_assert_controller_permission_literals(self, literals)
|
|
|
|
def test_vue_guard_parser_ignores_comments_and_unrelated_constants(self) -> None:
|
|
source = """
|
|
<!-- v-hasPermi="['master:strain:query']" -->
|
|
<script setup>
|
|
const staleExample = 'master:strain:query'
|
|
</script>
|
|
"""
|
|
|
|
self.assertEqual(frozenset(), _extract_guarded_vue_permissions(source))
|
|
|
|
def test_vue_pages_have_exact_guarded_permission_contract(self) -> None:
|
|
actual_permissions = set()
|
|
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)
|
|
self.assertTrue(guarded, relative_path)
|
|
actual_permissions.update(guarded)
|
|
|
|
self.assertEqual(EXPECTED_WEB_PERMISSIONS, frozenset(actual_permissions))
|
|
|
|
def test_frontend_has_no_physical_delete_api_or_page_contract(self) -> None:
|
|
relative_paths = (
|
|
'src/api/master/strain.js',
|
|
'src/api/master/customer.js',
|
|
'src/api/master/site.js',
|
|
'src/api/master/line.js',
|
|
'src/api/master/warehouse.js',
|
|
'src/api/master/pallet.js',
|
|
'src/api/business/inbound.js',
|
|
'src/api/business/shipment.js',
|
|
*VUE_PAGE_PATHS,
|
|
)
|
|
for relative_path in relative_paths:
|
|
source = (FRONTEND_ROOT / relative_path).read_text(encoding='utf-8')
|
|
code, structure = _sanitize_vue_source(source)
|
|
with self.subTest(relative_path=relative_path):
|
|
for symbol in REMOVED_DELETE_SYMBOLS:
|
|
self.assertIsNone(re.search(rf'\b{symbol}\b', structure), symbol)
|
|
self.assertFalse(REMOVED_DELETE_PERMISSIONS & set(PERMISSION_LITERAL_PATTERN.findall(code)))
|
|
|
|
def test_sensitive_test_batch_logs_only_include_batch_id_and_response_summary(self) -> None:
|
|
source_path = BACKEND_ROOT / 'module_business/controller/test_batch_controller.py'
|
|
tree = ast.parse(source_path.read_text(encoding='utf-8'), filename=str(source_path))
|
|
functions = {node.name: node for node in tree.body if isinstance(node, ast.AsyncFunctionDef)}
|
|
sensitive_handlers = {
|
|
'preview_test_batch_adoption',
|
|
'execute_test_batch_adoption',
|
|
'preview_test_batch_purge',
|
|
'purge_test_batch',
|
|
}
|
|
self.assertEqual(sensitive_handlers, sensitive_handlers & functions.keys())
|
|
|
|
for handler_name in sensitive_handlers:
|
|
log_calls = [
|
|
decorator
|
|
for decorator in functions[handler_name].decorator_list
|
|
if isinstance(decorator, ast.Call)
|
|
and isinstance(decorator.func, ast.Name)
|
|
and decorator.func.id == 'Log'
|
|
]
|
|
with self.subTest(handler=handler_name):
|
|
self.assertEqual(1, len(log_calls))
|
|
keywords = {keyword.arg: keyword.value for keyword in log_calls[0].keywords}
|
|
self.assertEqual('include', ast.literal_eval(keywords['request_log_mode']))
|
|
self.assertEqual('summary', ast.literal_eval(keywords['response_log_mode']))
|
|
include_fields = keywords['request_include_fields']
|
|
self.assertIsInstance(include_fields, ast.Tuple)
|
|
self.assertEqual(1, len(include_fields.elts))
|
|
field_call = include_fields.elts[0]
|
|
self.assertIsInstance(field_call, ast.Call)
|
|
self.assertEqual('batch_id', ast.literal_eval(field_call.args[0]))
|
|
self.assertIsInstance(field_call.func, ast.Attribute)
|
|
self.assertEqual('field', field_call.func.attr)
|
|
root = field_call.func.value
|
|
self.assertIsInstance(root, ast.Attribute)
|
|
self.assertEqual('PATH_PARAMS', root.attr)
|
|
self.assertIsInstance(root.value, ast.Name)
|
|
self.assertEqual('RequestLogFieldRoot', root.value.id)
|
|
|
|
def test_vue_guard_parser_rejects_dynamic_arrays_and_has_permi_or(self) -> None:
|
|
dynamic_sources = (
|
|
'<button v-hasPermi="permissionArray">动态权限</button>',
|
|
'<button v-if="proxy.$auth.hasPermiAnd(permissionArray)">动态权限</button>',
|
|
'<button v-if="proxy.$auth.hasPermiAnd([\'master:strain:list\', permission])">动态权限</button>',
|
|
)
|
|
for source in dynamic_sources:
|
|
with self.subTest(source=source), self.assertRaises(AssertionError):
|
|
_extract_guarded_vue_permissions(source)
|
|
|
|
with self.assertRaisesRegex(AssertionError, 'hasPermiOr'):
|
|
_extract_guarded_vue_permissions(
|
|
'<button v-if="proxy.$auth.hasPermiOr([\'master:strain:query\'])">错误门禁</button>'
|
|
)
|
|
|
|
def test_vue_guard_parser_rejects_extra_has_permi_or_after_valid_guards(self) -> None:
|
|
source = """
|
|
<button v-hasPermi="['master:strain:list']">合法门禁</button>
|
|
<button v-if="proxy.$auth.hasPermiAnd(['master:strain:list'])">合法组合门禁</button>
|
|
<button v-if="proxy.$auth.hasPermiOr(['master:strain:query'])">非约定门禁</button>
|
|
"""
|
|
|
|
with self.assertRaisesRegex(AssertionError, 'hasPermiOr'):
|
|
_extract_guarded_vue_permissions(source)
|
|
|
|
def test_vue_guard_parser_rejects_noncanonical_permission_calls(self) -> None:
|
|
valid_guard = '<button v-hasPermi="[\'master:strain:list\']">合法门禁</button>'
|
|
invalid_calls = (
|
|
"proxy . $auth . hasPermiOr(['master:strain:query'])",
|
|
'proxy?.$auth?.hasPermiAnd(permissionArray)',
|
|
"proxy['$auth'].hasPermiOr(['master:strain:query'])",
|
|
)
|
|
for invalid_call in invalid_calls:
|
|
source = f'{valid_guard}<button v-if="{invalid_call}">非约定门禁</button>'
|
|
with (
|
|
self.subTest(invalid_call=invalid_call),
|
|
self.assertRaisesRegex(
|
|
AssertionError,
|
|
'hasPermi|canonical',
|
|
),
|
|
):
|
|
_extract_guarded_vue_permissions(source)
|
|
|
|
def test_vue_guard_parser_rejects_noncanonical_permission_receivers(self) -> None:
|
|
valid_guard = '<button v-hasPermi="[\'master:strain:list\']">合法门禁</button>'
|
|
invalid_calls = (
|
|
"myproxy.$auth.hasPermiAnd(['master:strain:query'])",
|
|
"foo.proxy.$auth.hasPermiAnd(['master:strain:query'])",
|
|
"$proxy.$auth.hasPermiAnd(['master:strain:query'])",
|
|
"this.proxy.$auth.hasPermiAnd(['master:strain:query'])",
|
|
)
|
|
for invalid_call in invalid_calls:
|
|
source = f'{valid_guard}<button v-if="{invalid_call}">错误接收者</button>'
|
|
with (
|
|
self.subTest(invalid_call=invalid_call),
|
|
self.assertRaisesRegex(AssertionError, 'canonical'),
|
|
):
|
|
_extract_guarded_vue_permissions(source)
|
|
|
|
def test_independent_expected_permissions_reject_layers_missing_same_permission(self) -> None:
|
|
missing = 'master:strain:query'
|
|
mutated = EXPECTED_PERMISSIONS - {missing}
|
|
layers = {
|
|
'mysql': mutated,
|
|
'controller': mutated,
|
|
'vue': mutated,
|
|
}
|
|
|
|
with self.assertRaises(AssertionError):
|
|
_assert_exact_permission_layers(self, layers)
|
|
|
|
def test_static_scan_and_option_routes_precede_dynamic_details(self) -> None:
|
|
route_order = {
|
|
route.path_format: index
|
|
for index, route in enumerate(self.app.routes)
|
|
if isinstance(route, APIRoute) and 'GET' in route.methods
|
|
}
|
|
for detail_path, static_paths in STATIC_READS_BEFORE_DETAIL.items():
|
|
with self.subTest(detail_path=detail_path):
|
|
self.assertIn(detail_path, route_order)
|
|
for static_path in static_paths:
|
|
self.assertIn(static_path, route_order)
|
|
self.assertLess(route_order[static_path], route_order[detail_path])
|
|
|
|
def test_openapi_parameter_names_are_exact(self) -> None:
|
|
for path, method in set(EXPECTED_ROUTE_PERMISSIONS) | UNGUARDED_AUTHENTICATED_ROUTES:
|
|
parameters = self.schema['paths'][path][method.lower()].get('parameters', [])
|
|
actual_path_parameters = {parameter['name'] for parameter in parameters if parameter['in'] == 'path'}
|
|
self.assertEqual(set(re.findall(r'{([^{}]+)}', path)), actual_path_parameters, (path, method))
|
|
|
|
for key, expected_query_parameters in EXACT_QUERY_PARAMETERS.items():
|
|
path, method = key
|
|
parameters = self.schema['paths'][path][method.lower()].get('parameters', [])
|
|
actual_query_parameters = {parameter['name'] for parameter in parameters if parameter['in'] == 'query'}
|
|
self.assertEqual(expected_query_parameters, actual_query_parameters, key)
|
|
|
|
for key, required_query_parameters in REQUIRED_QUERY_PARAMETERS.items():
|
|
path, method = key
|
|
parameters = self.schema['paths'][path][method.lower()].get('parameters', [])
|
|
actual_query_parameters = {parameter['name'] for parameter in parameters if parameter['in'] == 'query'}
|
|
self.assertLessEqual(required_query_parameters, actual_query_parameters, key)
|
|
|
|
for path, method in EXPECTED_ROUTE_PERMISSIONS:
|
|
if not path.startswith('/report/'):
|
|
continue
|
|
parameters = self.schema['paths'][path][method.lower()].get('parameters', [])
|
|
self.assertNotIn('mode', {parameter['name'] for parameter in parameters})
|
|
|
|
def test_every_write_route_uses_real_auth_dependency(self) -> None:
|
|
expected_write_permissions = {
|
|
key: permission for key, permission in EXPECTED_ROUTE_PERMISSIONS.items() if key[1] in WRITE_METHODS
|
|
}
|
|
actual_write_routes: dict[tuple[str, str], APIRoute] = {}
|
|
for route in self.target_routes:
|
|
for method in route.methods & WRITE_METHODS:
|
|
actual_write_routes[(route.path_format, method)] = route
|
|
|
|
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):
|
|
auth_dependencies = _auth_dependencies(actual_write_routes[key])
|
|
self.assertEqual(1, len(auth_dependencies))
|
|
self.assertIsInstance(auth_dependencies[0], CheckUserInterfaceAuth)
|
|
self.assertEqual(expected_permission, auth_dependencies[0].perm)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|