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', ) 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:tunnel:query', 'master:tunnel:add', 'master:tunnel:edit', 'master:tunnel:remove', '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', ) 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', } ) 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', ('/business/inbound/list', '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/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/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/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/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 = { '/business/inbound/{inbound_id}': ( '/business/inbound/warehouse-options', '/business/inbound/scan/{qr_token}', ), '/business/filling/{filling_id}': ( '/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/scan/{qr_token}', ), '/business/return/{return_id}': ( '/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/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(98, 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 + 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'(?\[[^\[\]]*\])\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[^"]*)"|'(?P[^']*)')""") 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[^"]*)"|'(?P[^']*)')""") 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(100, 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(48, len(EXPECTED_BUTTON_PERMISSIONS)) self.assertFalse(EXPECTED_PAGE_PERMISSIONS & EXPECTED_BUTTON_PERMISSIONS) self.assertEqual(67, len(EXPECTED_PERMISSIONS)) _assert_controller_permission_literals(self, literals) self.assertEqual( { 'add', 'adopt', 'assign', 'close', 'confirm', 'edit', '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(98, 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 = """ """ self.assertEqual(frozenset(), _extract_guarded_vue_permissions(source)) 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))) 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 = ( '', '', '', ) 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( '' ) def test_vue_guard_parser_rejects_extra_has_permi_or_after_valid_guards(self) -> None: source = """ """ with self.assertRaisesRegex(AssertionError, 'hasPermiOr'): _extract_guarded_vue_permissions(source) def test_vue_guard_parser_rejects_noncanonical_permission_calls(self) -> None: valid_guard = '' 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}' 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 = '' 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}' 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(46, 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()