314 lines
15 KiB
Python
314 lines
15 KiB
Python
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',
|
|
'/business',
|
|
'/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_target_path_selector_uses_namespace_boundaries(self) -> None:
|
|
for path in ('/master', '/master/foo', '/business', '/business/foo', '/report/foo'):
|
|
with self.subTest(path=path):
|
|
self.assertTrue(_is_target_path(path))
|
|
for path in ('/masterish', '/business-other', '/reporting'):
|
|
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), 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] = []
|
|
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(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()
|