测试:校验菜单接口权限一致性

This commit is contained in:
焦龙言
2026-07-27 14:59:36 +08:00
parent 9b5f479f0d
commit 33d73ca0ef
3 changed files with 783 additions and 48 deletions
@@ -1,6 +1,7 @@
import ast
import re
import unittest
from collections import Counter
from pathlib import Path
from fastapi.routing import APIRoute
@@ -9,6 +10,7 @@ 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',
@@ -31,28 +33,96 @@ TARGET_PREFIXES = (
ROUTE_HTTP_METHODS = frozenset({'DELETE', 'GET', 'HEAD', 'OPTIONS', 'PATCH', 'POST', 'PUT', 'TRACE'})
WRITE_METHODS = frozenset({'DELETE', 'PATCH', 'POST', 'PUT'})
EXPECTED_PERMISSION_ACTIONS = {
'master:strain': ('list', 'query', 'add', 'edit', 'remove'),
'master:customer': ('list', 'query', 'add', 'edit', 'remove'),
'master:site': ('list', 'query', 'add', 'edit', 'remove', 'assign'),
'master:line': ('list', 'query', 'add', 'edit', 'remove'),
'master:warehouse': ('list', 'query', 'add', 'edit', 'remove', 'assign'),
'master:pallet': ('list', 'query', 'add', 'edit', 'remove', 'print'),
'business:inbound': ('list', 'query', 'add', 'edit', 'remove', 'confirm'),
'business:filling': ('list', 'query', 'add'),
'business:shipment': ('list', 'query', 'add', 'edit', 'remove', 'confirm'),
'business:usage': ('list', 'query', 'add'),
'business:return': ('list', 'query', 'request', 'confirm'),
'report:palletinventory': ('list',),
'report:filling': ('list',),
'report:shipment': ('list',),
'report:consumption': ('list',),
'report:return': ('list',),
'report:pallettrace': ('list',),
}
EXPECTED_PERMISSIONS = frozenset(
f'{resource}:{action}' for resource, actions in EXPECTED_PERMISSION_ACTIONS.items() for action in actions
EXPECTED_PAGE_PERMISSIONS = frozenset(
{
'master:strain:list',
'master:customer:list',
'master:site:list',
'master:line:list',
'master:warehouse:list',
'master:pallet: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:strain:remove',
'master:customer:query',
'master:customer:add',
'master:customer:edit',
'master:customer:remove',
'master:site:query',
'master:site:add',
'master:site:edit',
'master:site:remove',
'master:site:assign',
'master:line:query',
'master:line:add',
'master:line:edit',
'master:line:remove',
'master:warehouse:query',
'master:warehouse:add',
'master:warehouse:edit',
'master:warehouse:remove',
'master:warehouse:assign',
'master:pallet:query',
'master:pallet:add',
'master:pallet:edit',
'master:pallet:remove',
'master:pallet:print',
'business:inbound:query',
'business:inbound:add',
'business:inbound:edit',
'business:inbound:remove',
'business:inbound:confirm',
'business:filling:query',
'business:filling:add',
'business:shipment:query',
'business:shipment:add',
'business:shipment:edit',
'business:shipment:remove',
'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
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/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',
)
PERMISSION_LITERAL_PATTERN = re.compile(r'(?:master|business|report):[a-z]+:[a-z]+')
def _build_expected_route_permissions() -> dict[tuple[str, str], str]:
@@ -169,6 +239,119 @@ 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(74, 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)
call_tokens = list(re.finditer(r'proxy\.\$auth\.hasPermiAnd\s*\(', structure))
call_pattern = re.compile(r'proxy\.\$auth\.hasPermiAnd\s*\(\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:
@@ -246,13 +429,76 @@ class ApiPermissionContractTest(unittest.TestCase):
literals.append(node.args[0].value)
self.assertEqual([], malformed_calls)
self.assertEqual(17, len(EXPECTED_PAGE_PERMISSIONS))
self.assertEqual(44, len(EXPECTED_BUTTON_PERMISSIONS))
self.assertFalse(EXPECTED_PAGE_PERMISSIONS & EXPECTED_BUTTON_PERMISSIONS)
self.assertEqual(61, len(EXPECTED_PERMISSIONS))
self.assertEqual(EXPECTED_PERMISSIONS, frozenset(literals))
_assert_controller_permission_literals(self, literals)
self.assertEqual(
{'add', 'assign', 'confirm', 'edit', 'list', 'print', 'query', 'remove', 'request'},
{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(74, 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(17, len(VUE_PAGE_PATHS))
self.assertEqual(17, 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)
_assert_exact_permission_layers(self, {'vue': actual_permissions})
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)
guarded = _extract_guarded_vue_permissions(
'<button v-if="proxy.$auth.hasPermiOr([\'master:strain:query\'])">错误门禁</button>'
)
self.assertEqual(frozenset(), guarded)
with self.assertRaises(AssertionError):
_assert_exact_permission_layers(self, {'vue': guarded})
def test_independent_expected_permissions_reject_four_layers_missing_same_permission(self) -> None:
missing = 'master:strain:query'
mutated = EXPECTED_PERMISSIONS - {missing}
layers = {
'mysql': mutated,
'postgresql': 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: index