From dc190f1162638dc819f57508d26e80b90912ba7f Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E7=84=A6=E9=BE=99=E8=A8=80?= Date: Mon, 27 Jul 2026 02:25:39 +0800 Subject: [PATCH] =?UTF-8?q?=E6=B5=8B=E8=AF=95=EF=BC=9A=E5=9B=BA=E5=8C=96?= =?UTF-8?q?=E4=B8=9A=E5=8A=A1=E6=8E=A5=E5=8F=A3=E6=9D=83=E9=99=90=E5=A5=91?= =?UTF-8?q?=E7=BA=A6?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../test_api_permission_contract.py | 311 ++++++++++++++++++ 1 file changed, 311 insertions(+) create mode 100644 shierfeng-fastapi-backend/tests/module_business/test_api_permission_contract.py diff --git a/shierfeng-fastapi-backend/tests/module_business/test_api_permission_contract.py b/shierfeng-fastapi-backend/tests/module_business/test_api_permission_contract.py new file mode 100644 index 0000000..7962e42 --- /dev/null +++ b/shierfeng-fastapi-backend/tests/module_business/test_api_permission_contract.py @@ -0,0 +1,311 @@ +import ast +import re +import unittest +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] +CONTROLLER_NAMES = ( + 'strain', + 'customer', + 'site', + 'line', + 'warehouse', + 'pallet', + 'inbound', + 'filling', + 'shipment', + 'usage', + 'return', + 'report', +) +TARGET_PREFIXES = ( + '/master/strain', + '/master/customer', + '/master/site', + '/master/line', + '/master/warehouse', + '/master/pallet', + '/business/inbound', + '/business/filling', + '/business/shipment', + '/business/usage', + '/business/return', + '/report', +) +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 +) + + +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}: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:remove', + ('/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', 'POST'): 'business:shipment:add', + ('/business/shipment', 'PUT'): 'business:shipment:edit', + ('/business/shipment/{shipment_ids}', 'DELETE'): 'business:shipment:remove', + ('/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/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', + } + ) + 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/scan/{qr_token}', + ), + '/business/return/{return_id}': ( + '/business/return/warehouse-options', + '/business/return/scan/{qr_token}', + ), +} +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/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) + ] + + +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) + ] + + 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), actual_operations) + self.assertEqual(74, 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 + 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) + key = (route.path, next(iter(methods))) + operation_keys.append(key) + + auth_dependencies = _auth_dependencies(route) + 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] = [] + for controller_name in CONTROLLER_NAMES: + source_path = BACKEND_ROOT / 'module_business/controller' / f'{controller_name}_controller.py' + 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(61, len(EXPECTED_PERMISSIONS)) + self.assertEqual(EXPECTED_PERMISSIONS, frozenset(literals)) + self.assertEqual( + {'add', 'assign', 'confirm', 'edit', 'list', 'print', 'query', 'remove', 'request'}, + {permission.rsplit(':', maxsplit=1)[1] for permission in literals}, + ) + + def test_static_scan_and_option_routes_precede_dynamic_details(self) -> None: + route_order = { + route.path: 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 EXPECTED_ROUTE_PERMISSIONS: + 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, method)] = route + + self.assertEqual(32, 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()