From 12260fdf8375cc4eab63e09389cf1e6610d5bef1 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 00:08:57 +0800 Subject: [PATCH] =?UTF-8?q?=E4=BF=AE=E5=A4=8D=EF=BC=9A=E5=AE=8C=E5=96=84?= =?UTF-8?q?=E9=80=80=E5=9B=9E=E5=B9=B6=E5=8F=91=E5=B9=82=E7=AD=89=E4=B8=8E?= =?UTF-8?q?=E4=BF=A1=E6=81=AF=E9=9A=90=E8=97=8F?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../module_business/service/return_service.py | 124 +++-- .../module_business/test_return_service.py | 471 +++++++++++++++++- 2 files changed, 556 insertions(+), 39 deletions(-) diff --git a/shierfeng-fastapi-backend/module_business/service/return_service.py b/shierfeng-fastapi-backend/module_business/service/return_service.py index 16cff8a..ea98792 100644 --- a/shierfeng-fastapi-backend/module_business/service/return_service.py +++ b/shierfeng-fastapi-backend/module_business/service/return_service.py @@ -377,7 +377,7 @@ class ReturnService: cls._raise_pallet_not_accessible(current_user) matching_items = [item for item in return_order.items if item.pallet_id == pallet.pallet_id] if len(matching_items) != 1: - raise ServiceException(message='扫码托盘不属于该退回单') + cls._raise_pallet_not_accessible(current_user, '扫码托盘不属于该退回单') inventory = await MasterDao.get_pallet_inventory(query_db, pallet.pallet_id) if inventory is None: cls._raise_pallet_not_accessible(current_user, '托盘库存不存在') @@ -445,21 +445,37 @@ class ReturnService: query_db: AsyncSession, pallet_ids: list[int], source_site_id: int, - ) -> tuple[list[BizPalletInventory], list[BizPallet], dict[int, BizFillingRecord]]: + current_user: CurrentUserModel, + ) -> tuple[ + list[BizPalletInventory], + list[BizPallet], + dict[int, BizFillingRecord | None], + list[int] | None, + ]: ordered_ids = cls._validate_pallet_ids(pallet_ids) inventories = await InventoryDao.get_inventories_for_update(query_db, ordered_ids) - pallets = await InventoryDao.get_pallets_by_ids(query_db, ordered_ids) inventory_by_id = {inventory.pallet_id: inventory for inventory in inventories} - pallet_by_id = {pallet.pallet_id: pallet for pallet in pallets} - if set(inventory_by_id) != set(ordered_ids) or set(pallet_by_id) != set(ordered_ids): - raise ServiceException(message='存在不存在或库存缺失的托盘') + if set(inventory_by_id) != set(ordered_ids): + cls._raise_pallet_not_accessible(current_user, '存在不存在或库存缺失的托盘') ordered_inventories = [inventory_by_id[pallet_id] for pallet_id in ordered_ids] - ordered_pallets = [pallet_by_id[pallet_id] for pallet_id in ordered_ids] - for pallet, inventory in zip(ordered_pallets, ordered_inventories, strict=True): - cls._validate_request_inventory(pallet, inventory, source_site_id) + if any(inventory.site_id != source_site_id for inventory in ordered_inventories): + cls._raise_pallet_not_accessible(current_user, '同一退回单托盘必须属于同一现场') + try: + await AccessService.assert_site_access(query_db, current_user, source_site_id) + except PermissionException: + cls._raise_pallet_not_accessible(current_user) + fresh_site_ids = await cls._site_scope(query_db, current_user) + if fresh_site_ids is not None and source_site_id not in fresh_site_ids: + cls._raise_pallet_not_accessible(current_user) - filling_by_id = {} + pallets = await InventoryDao.get_pallets_by_ids(query_db, ordered_ids) + pallet_by_id = {pallet.pallet_id: pallet for pallet in pallets} + if set(pallet_by_id) != set(ordered_ids): + cls._raise_pallet_not_accessible(current_user, '存在不存在或库存缺失的托盘') + ordered_pallets = [pallet_by_id[pallet_id] for pallet_id in ordered_ids] + + filling_by_id: dict[int, BizFillingRecord | None] = {} filling_ids = sorted( { inventory.active_filling_id @@ -468,14 +484,31 @@ class ReturnService: } ) for filling_id in filling_ids: - filling = await OperationDao.get_active_filling_for_update(query_db, filling_id) - if filling is None: - raise ServiceException(message='托盘活动装填周期不存在或已关闭') - filling_by_id[filling_id] = filling - for inventory in ordered_inventories: + filling_by_id[filling_id] = await OperationDao.get_active_filling_for_update(query_db, filling_id) + return ordered_inventories, ordered_pallets, filling_by_id, fresh_site_ids + + @classmethod + def _validate_locked_request_pallets( + cls, + inventories: list[BizPalletInventory], + pallets: list[BizPallet], + filling_by_id: dict[int, BizFillingRecord | None], + source_site_id: int, + current_user: CurrentUserModel, + ) -> None: + for pallet, inventory in zip(pallets, inventories, strict=True): + cls._validate_request_inventory(pallet, inventory, source_site_id) + for inventory in inventories: if inventory.active_filling_id is not None: - cls._validate_active_filling(filling_by_id[inventory.active_filling_id], inventory) - return ordered_inventories, ordered_pallets, filling_by_id + filling = filling_by_id[inventory.active_filling_id] + if filling is None: + cls._raise_pallet_not_accessible(current_user, '托盘活动装填周期不存在或已关闭') + try: + cls._validate_active_filling(filling, inventory) + except ServiceException: + if not cls._is_admin(current_user): + cls._raise_pallet_not_accessible(current_user) + raise @staticmethod def _build_return_items( @@ -585,7 +618,7 @@ class ReturnService: await query_db.rollback() @classmethod - async def request_return_services( + async def request_return_services( # noqa: PLR0915 cls, query_db: AsyncSession, page_object: ReturnRequestModel, @@ -595,30 +628,34 @@ class ReturnService: user_id, user_name = cls._current_identity(current_user) ordered_ids = cls._validate_pallet_ids(page_object.pallet_ids) request_hash = cls._build_request_hash(page_object) - site_ids = await cls._site_scope(query_db, current_user) - if site_ids is not None and page_object.source_site_id not in site_ids: - raise PermissionException(data='', message='无权访问该客户现场') - await AccessService.assert_site_access( - query_db, - current_user, - page_object.source_site_id, - ) - request_warehouse_ids = None if site_ids is None else [] + preflight_site_ids = await cls._site_scope(query_db, current_user) + if preflight_site_ids is not None and page_object.source_site_id not in preflight_site_ids: + cls._raise_pallet_not_accessible(current_user) + try: + await AccessService.assert_site_access( + query_db, + current_user, + page_object.source_site_id, + ) + except PermissionException: + cls._raise_pallet_not_accessible(current_user) + request_warehouse_ids = None if preflight_site_ids is None else [] existing = await OperationDao.get_return_by_request_idempotency_key( query_db, page_object.idempotency_key, - site_ids=site_ids, + site_ids=preflight_site_ids, warehouse_ids=request_warehouse_ids, ) if existing is not None and not cls._is_exact_request_result(existing, page_object, request_hash): raise ServiceException(message='申请退回失败，幂等键已被其他请求使用') + preflight_existing_id = int(existing.return_id) if existing is not None else None await query_db.rollback() - if existing is not None: + if preflight_existing_id is not None: locked_existing = await OperationDao.get_return_for_update( query_db, - existing.return_id, - site_ids=site_ids, + preflight_existing_id, + site_ids=preflight_site_ids, warehouse_ids=request_warehouse_ids, ) if locked_existing is None: @@ -635,12 +672,35 @@ class ReturnService: await query_db.rollback() return cls._success('申请成功', return_id) - inventories, _, _ = await cls._lock_request_pallets( + inventories, pallets, filling_by_id, fresh_site_ids = await cls._lock_request_pallets( query_db, ordered_ids, page_object.source_site_id, + current_user, ) + fresh_warehouse_ids = None if fresh_site_ids is None else [] + fresh_existing = await OperationDao.get_return_by_request_idempotency_key( + query_db, + page_object.idempotency_key, + site_ids=fresh_site_ids, + warehouse_ids=fresh_warehouse_ids, + ) + if fresh_existing is not None: + if not cls._is_exact_request_result(fresh_existing, page_object, request_hash): + raise ServiceException(message='申请退回失败，幂等键已被其他请求使用') + fresh_existing_id = int(fresh_existing.return_id) + await cls._lock_and_reauthorize_request_site(query_db, fresh_existing, current_user) + await query_db.rollback() + return cls._success('申请成功', fresh_existing_id) + await cls._lock_new_request_route(query_db, page_object, current_user) + cls._validate_locked_request_pallets( + inventories, + pallets, + filling_by_id, + page_object.source_site_id, + current_user, + ) for inventory in inventories: PalletStateMachine.ensure_transition( PalletFlowState(inventory.flow_state), diff --git a/shierfeng-fastapi-backend/tests/module_business/test_return_service.py b/shierfeng-fastapi-backend/tests/module_business/test_return_service.py index 31bae1b..02840a9 100644 --- a/shierfeng-fastapi-backend/tests/module_business/test_return_service.py +++ b/shierfeng-fastapi-backend/tests/module_business/test_return_service.py @@ -278,6 +278,26 @@ class ReturnDaoScopeRedTest(unittest.IsolatedAsyncioTestCase): self.assertIn('biz_return_order.target_warehouse_id IN (7)', sql) self.assertIn(' OR ', sql) + async def test_request_key_read_is_scoped_and_never_locks_missing_unique_key(self) -> None: + result = MagicMock() + result.scalar_one_or_none.return_value = None + self.db.execute.return_value = result + + await OperationDao.get_return_by_request_idempotency_key( + self.db, + 'request-key', + site_ids=[9], + warehouse_ids=[], + ) + + statement = self.db.execute.await_args.args[0] + self.assertTrue(statement.get_final_froms()[0].compare(BizReturnOrder.__table__)) + for dialect in (mysql.dialect(), postgresql.dialect()): + sql = compile_sql(statement, dialect) + self.assertIn("biz_return_order.request_idempotency_key = 'request-key'", sql) + self.assertIn('biz_return_order.source_site_id IN (9)', sql) + self.assertNotIn('FOR UPDATE', sql) + @unittest.skipUnless(ReturnService is not None, 'ReturnService 尚未实现') class ReturnServiceContractRedTest(unittest.TestCase): @@ -561,6 +581,51 @@ class ReturnReadAndScanRedTest(unittest.IsolatedAsyncioTestCase): await ReturnService.scan_return_pallet_services(self.db, 'qr-2', query, self.user) self.assertEqual('无权访问该退回单', hidden_error.exception.message) + async def test_confirm_scan_hides_missing_and_foreign_pallet_equally_for_normal_user(self) -> None: + order = make_return_order(snapshots=((2, 102, 8),)) + query = ReturnScanQueryModel(mode='confirm', returnId=13) + messages = [] + for pallet in (None, make_pallet(5)): + with self.subTest(pallet_exists=pallet is not None): + inventory_read = AsyncMock() + with ( + patch.object(AccessDao, 'list_user_warehouse_ids', new=AsyncMock(return_value=[7])), + patch.object(OperationDao, 'get_return_by_id', new=AsyncMock(return_value=order)), + patch.object(InventoryDao, 'get_pallet_by_qr_token', new=AsyncMock(return_value=pallet)), + patch.object(MasterDao, 'get_pallet_inventory', new=inventory_read), + ): + try: + await ReturnService.scan_return_pallet_services(self.db, 'qr-5', query, self.user) + except Exception as exc: + raised = exc + else: + raised = None + self.assertIsInstance(raised, PermissionException) + messages.append(getattr(raised, 'message', '未抛出异常')) + inventory_read.assert_not_awaited() + + self.assertEqual(['无权访问该托盘', '无权访问该托盘'], messages) + + async def test_admin_confirm_scan_keeps_precise_missing_and_foreign_pallet_errors(self) -> None: + admin = make_user(1) + order = make_return_order(snapshots=((2, 102, 8),)) + query = ReturnScanQueryModel(mode='confirm', returnId=13) + cases = ( + (None, '托盘不存在'), + (make_pallet(5), '扫码托盘不属于该退回单'), + ) + for pallet, expected_message in cases: + with self.subTest(expected_message=expected_message): + with ( + patch.object(OperationDao, 'get_return_by_id', new=AsyncMock(return_value=order)), + patch.object(InventoryDao, 'get_pallet_by_qr_token', new=AsyncMock(return_value=pallet)), + patch.object(MasterDao, 'get_pallet_inventory', new=AsyncMock()) as inventory_read, + self.assertRaises(ServiceException) as raised, + ): + await ReturnService.scan_return_pallet_services(self.db, 'qr-5', query, admin) + self.assertEqual(expected_message, raised.exception.message) + inventory_read.assert_not_awaited() + @unittest.skipUnless(ReturnService is not None, 'ReturnService 尚未实现') class ReturnRequestTransactionRedTest(unittest.IsolatedAsyncioTestCase): @@ -575,6 +640,337 @@ class ReturnRequestTransactionRedTest(unittest.IsolatedAsyncioTestCase): idempotencyKey='request-key', ) + async def test_concurrent_same_request_returns_winner_after_inventory_wait(self) -> None: # noqa: PLR0915 + request_hash = ReturnService._build_request_hash(self.page) + winner = make_return_order(request_hash=request_hash) + inventories = [ + make_inventory(2, state=PalletFlowState.RETURNING.value), + make_inventory(5, remaining=0, state=PalletFlowState.RETURNING.value), + ] + actions: list[str] = [] + key_count = 0 + scope_count = 0 + access_count = 0 + rollback_count = 0 + + async def read_key(*_args: Any, **_kwargs: Any) -> BizReturnOrder | None: + nonlocal key_count + key_count += 1 + actions.append('key-preflight' if key_count == 1 else 'key-fresh') + return None if key_count == 1 else winner + + async def read_scope(*_args: Any) -> list[int]: + nonlocal scope_count + scope_count += 1 + actions.append('scope-preflight' if scope_count == 1 else 'scope-fresh') + return [9] + + async def check_access(*_args: Any, **kwargs: Any) -> None: + nonlocal access_count + access_count += 1 + if kwargs.get('for_update', False): + actions.append('mapping-lock') + else: + actions.append('access-preflight' if access_count == 1 else 'access-fresh') + + async def rollback() -> None: + nonlocal rollback_count + rollback_count += 1 + actions.append('rollback-preflight' if rollback_count == 1 else 'rollback-result') + + async def lock_inventories(*_args: Any) -> list[BizPalletInventory]: + actions.append('inventory-lock') + return inventories + + async def lock_pallets(*_args: Any) -> list[BizPallet]: + actions.append('pallet-lock') + return [make_pallet(2), make_pallet(5)] + + async def lock_filling(*_args: Any) -> BizFillingRecord: + actions.append('filling-lock') + return make_filling(2) + + async def lock_site(*_args: Any, **_kwargs: Any) -> SimpleNamespace: + actions.append('site-lock') + return SimpleNamespace(status='0') + + key_read = AsyncMock(side_effect=read_key) + scope_read = AsyncMock(side_effect=read_scope) + self.db.rollback.side_effect = rollback + with ( + patch.object(AccessDao, 'list_user_site_ids', new=scope_read), + patch.object(OperationDao, 'get_return_by_request_idempotency_key', new=key_read), + patch.object(InventoryDao, 'get_inventories_for_update', new=AsyncMock(side_effect=lock_inventories)), + patch.object(InventoryDao, 'get_pallets_by_ids', new=AsyncMock(side_effect=lock_pallets)), + patch.object( + OperationDao, + 'get_active_filling_for_update', + new=AsyncMock(side_effect=lock_filling), + ), + patch.object(AccessService, 'assert_site_access', new=AsyncMock(side_effect=check_access)), + patch.object(MasterDao, 'get_site', new=AsyncMock(side_effect=lock_site)), + patch.object(OperationDao, 'add_return_order', new=AsyncMock()) as add_order, + patch.object(OperationDao, 'add_return_items', new=AsyncMock()) as add_items, + patch.object(InventoryDao, 'add_event', new=AsyncMock()) as add_event, + ): + try: + result = await ReturnService.request_return_services(self.db, self.page, self.user) + except ServiceException as exc: + self.fail(f'锁等待后的精确重试被库存状态提前拒绝：{exc.message}') + + self.assertEqual(13, result.result) + self.assertEqual( + [ + 'scope-preflight', + 'access-preflight', + 'key-preflight', + 'rollback-preflight', + 'inventory-lock', + 'access-fresh', + 'scope-fresh', + 'pallet-lock', + 'filling-lock', + 'key-fresh', + 'site-lock', + 'mapping-lock', + 'rollback-result', + ], + actions, + ) + self.assertEqual(2, key_read.await_count) + self.assertEqual(2, scope_read.await_count) + add_order.assert_not_awaited() + add_items.assert_not_awaited() + add_event.assert_not_awaited() + self.db.commit.assert_not_awaited() + self.assertEqual([PalletFlowState.RETURNING.value] * 2, [item.flow_state for item in inventories]) + + async def test_concurrent_same_key_different_payload_is_stable_idempotency_conflict(self) -> None: + winner = make_return_order(request_hash=ReturnService._build_request_hash(self.page)) + conflict_page = self.page.model_copy(update={'target_warehouse_id': 8}) + inventories = [ + make_inventory(2, state=PalletFlowState.RETURNING.value), + make_inventory(5, remaining=0, state=PalletFlowState.RETURNING.value), + ] + key_read = AsyncMock(side_effect=[None, winner]) + with ( + patch.object(AccessDao, 'list_user_site_ids', new=AsyncMock(return_value=[9])), + patch.object(OperationDao, 'get_return_by_request_idempotency_key', new=key_read), + patch.object(AccessService, 'assert_site_access', new=AsyncMock()), + patch.object(InventoryDao, 'get_inventories_for_update', new=AsyncMock(return_value=inventories)), + patch.object( + InventoryDao, + 'get_pallets_by_ids', + new=AsyncMock(return_value=[make_pallet(2), make_pallet(5)]), + ), + patch.object( + OperationDao, + 'get_active_filling_for_update', + new=AsyncMock(return_value=make_filling(2)), + ), + patch.object(OperationDao, 'add_return_order', new=AsyncMock()) as add_order, + self.assertRaises(ServiceException) as raised, + ): + await ReturnService.request_return_services(self.db, conflict_page, self.user) + + self.assertEqual('申请退回失败，幂等键已被其他请求使用', raised.exception.message) + self.assertEqual(2, key_read.await_count) + add_order.assert_not_awaited() + self.db.commit.assert_not_awaited() + self.assertEqual(2, self.db.rollback.await_count) + + async def test_concurrent_winner_uses_fresh_expanded_site_scope(self) -> None: + winner = make_return_order(request_hash=ReturnService._build_request_hash(self.page)) + inventories = [ + make_inventory(2, state=PalletFlowState.RETURNING.value), + make_inventory(5, remaining=0, state=PalletFlowState.RETURNING.value), + ] + scope_read = AsyncMock(side_effect=[[9], [9, 10]]) + key_read = AsyncMock(side_effect=[None, winner]) + with ( + patch.object(AccessDao, 'list_user_site_ids', new=scope_read), + patch.object(OperationDao, 'get_return_by_request_idempotency_key', new=key_read), + patch.object(AccessService, 'assert_site_access', new=AsyncMock()), + patch.object(InventoryDao, 'get_inventories_for_update', new=AsyncMock(return_value=inventories)), + patch.object( + InventoryDao, + 'get_pallets_by_ids', + new=AsyncMock(return_value=[make_pallet(2), make_pallet(5)]), + ), + patch.object( + OperationDao, + 'get_active_filling_for_update', + new=AsyncMock(return_value=make_filling(2)), + ), + patch.object(MasterDao, 'get_site', new=AsyncMock(return_value=SimpleNamespace(status='0'))), + ): + result = await ReturnService.request_return_services(self.db, self.page, self.user) + + self.assertEqual(13, result.result) + self.assertEqual( + [ + {'site_ids': [9], 'warehouse_ids': []}, + {'site_ids': [9, 10], 'warehouse_ids': []}, + ], + [call.kwargs for call in key_read.await_args_list], + ) + self.db.commit.assert_not_awaited() + self.assertEqual(2, self.db.rollback.await_count) + + async def test_request_hides_site_revoked_during_inventory_wait(self) -> None: + inventories = [ + make_inventory(2), + make_inventory(5, remaining=0, state=PalletFlowState.AT_SITE_EMPTY.value), + ] + scope_read = AsyncMock(side_effect=[[9], []]) + key_read = AsyncMock(return_value=None) + inventory_locks = AsyncMock(return_value=inventories) + pallet_locks = AsyncMock(return_value=[make_pallet(2), make_pallet(5)]) + with ( + patch.object(AccessDao, 'list_user_site_ids', new=scope_read), + patch.object(OperationDao, 'get_return_by_request_idempotency_key', new=key_read), + patch.object(AccessService, 'assert_site_access', new=AsyncMock()), + patch.object(InventoryDao, 'get_inventories_for_update', new=inventory_locks), + patch.object(InventoryDao, 'get_pallets_by_ids', new=pallet_locks), + self.assertRaises(PermissionException) as raised, + ): + await ReturnService.request_return_services(self.db, self.page, self.user) + + self.assertEqual('无权访问该托盘', raised.exception.message) + inventory_locks.assert_awaited_once_with(self.db, [2, 5]) + pallet_locks.assert_not_awaited() + self.assertEqual(2, scope_read.await_count) + self.assertEqual(1, key_read.await_count) + self.db.commit.assert_not_awaited() + self.assertEqual(2, self.db.rollback.await_count) + + async def test_preflight_existing_uses_only_copied_id_after_rollback(self) -> None: + request_hash = ReturnService._build_request_hash(self.page) + + class ExpiringReturnProbe: + def __init__(self) -> None: + self.expired = False + self.return_id_reads = 0 + self.request_idempotency_key = 'request-key' + self.request_hash = request_hash + self.source_site_id = 9 + self.target_warehouse_id = 7 + self.status = ReturnStatus.REQUESTED.value + + @property + def return_id(self) -> int: + if self.expired: + raise AssertionError('rollback 后禁止访问预检 return ORM') + self.return_id_reads += 1 + return 13 + + probe = ExpiringReturnProbe() + fresh = make_return_order(request_hash=request_hash) + + async def rollback() -> None: + probe.expired = True + + self.db.rollback.side_effect = rollback + order_lock = AsyncMock(return_value=fresh) + with ( + patch.object(AccessDao, 'list_user_site_ids', new=AsyncMock(return_value=[9])), + patch.object(AccessService, 'assert_site_access', new=AsyncMock()), + patch.object( + OperationDao, + 'get_return_by_request_idempotency_key', + new=AsyncMock(return_value=probe), + ), + patch.object(OperationDao, 'get_return_for_update', new=order_lock), + patch.object(MasterDao, 'get_site', new=AsyncMock(return_value=SimpleNamespace(status='0'))), + patch.object(InventoryDao, 'get_inventories_for_update', new=AsyncMock()) as inventory_locks, + ): + result = await ReturnService.request_return_services(self.db, self.page, self.user) + + self.assertEqual(13, result.result) + self.assertTrue(probe.expired) + self.assertEqual(1, probe.return_id_reads) + order_lock.assert_awaited_once_with(self.db, 13, site_ids=[9], warehouse_ids=[]) + inventory_locks.assert_not_awaited() + self.assertEqual(2, self.db.rollback.await_count) + + async def test_request_hides_missing_and_foreign_site_pallets_equally_for_normal_user(self) -> None: + cases = ( + ('missing', []), + ('foreign-site', [make_inventory(2, site_id=10), make_inventory(5, site_id=10)]), + ) + messages = [] + for name, inventories in cases: + with self.subTest(case=name): + db = MagicMock(spec=AsyncSession) + pallet_locks = AsyncMock(return_value=[make_pallet(2), make_pallet(5)]) + with ( + patch.object(AccessDao, 'list_user_site_ids', new=AsyncMock(return_value=[9])), + patch.object( + OperationDao, + 'get_return_by_request_idempotency_key', + new=AsyncMock(return_value=None), + ), + patch.object(AccessService, 'assert_site_access', new=AsyncMock()), + patch.object( + InventoryDao, + 'get_inventories_for_update', + new=AsyncMock(return_value=inventories), + ), + patch.object(InventoryDao, 'get_pallets_by_ids', new=pallet_locks), + ): + try: + await ReturnService.request_return_services(db, self.page, self.user) + except Exception as exc: + raised = exc + else: + raised = None + self.assertIsInstance(raised, PermissionException) + messages.append(getattr(raised, 'message', '未抛出异常')) + pallet_locks.assert_not_awaited() + db.commit.assert_not_awaited() + db.rollback.assert_awaited() + + self.assertEqual(['无权访问该托盘', '无权访问该托盘'], messages) + + async def test_request_hides_invalid_active_filling_association_for_normal_user(self) -> None: + inventories = [ + make_inventory(2), + make_inventory(5, remaining=0, state=PalletFlowState.AT_SITE_EMPTY.value), + ] + with ( + patch.object(AccessDao, 'list_user_site_ids', new=AsyncMock(return_value=[9])), + patch.object( + OperationDao, + 'get_return_by_request_idempotency_key', + new=AsyncMock(return_value=None), + ), + patch.object(AccessService, 'assert_site_access', new=AsyncMock()), + patch.object(InventoryDao, 'get_inventories_for_update', new=AsyncMock(return_value=inventories)), + patch.object( + InventoryDao, + 'get_pallets_by_ids', + new=AsyncMock(return_value=[make_pallet(2), make_pallet(5)]), + ), + patch.object( + OperationDao, + 'get_active_filling_for_update', + new=AsyncMock(return_value=make_filling(5)), + ), + patch.object(MasterDao, 'get_site', new=AsyncMock(return_value=SimpleNamespace(status='0'))), + patch.object(MasterDao, 'get_warehouse', new=AsyncMock(return_value=SimpleNamespace(status='0'))), + ): + try: + await ReturnService.request_return_services(self.db, self.page, self.user) + except Exception as exc: + raised = exc + else: + raised = None + + self.assertIsInstance(raised, PermissionException) + self.assertEqual('无权访问该托盘', getattr(raised, 'message', None)) + self.db.commit.assert_not_awaited() + self.db.rollback.assert_awaited() + async def test_request_freezes_all_pallets_saves_snapshots_and_appends_events(self) -> None: inventories = [ make_inventory(2), @@ -603,12 +999,16 @@ class ReturnRequestTransactionRedTest(unittest.IsolatedAsyncioTestCase): return event with ( - patch.object(AccessDao, 'list_user_site_ids', new=AsyncMock(return_value=[9, 9])), + patch.object( + AccessDao, + 'list_user_site_ids', + new=AsyncMock(return_value=[9, 9]), + ) as scope_read, patch.object( OperationDao, 'get_return_by_request_idempotency_key', new=AsyncMock(return_value=None), - ), + ) as key_read, patch.object(InventoryDao, 'get_inventories_for_update', new=AsyncMock(return_value=inventories)) as locks, patch.object(InventoryDao, 'get_pallets_by_ids', new=AsyncMock(return_value=pallets)), patch.object( @@ -657,6 +1057,12 @@ class ReturnRequestTransactionRedTest(unittest.IsolatedAsyncioTestCase): self.assertTrue(result.is_success) self.assertEqual(13, result.result) + self.assertEqual(2, scope_read.await_count) + self.assertEqual(2, key_read.await_count) + self.assertEqual( + [{'site_ids': [9], 'warehouse_ids': []}, {'site_ids': [9], 'warehouse_ids': []}], + [call.kwargs for call in key_read.await_args_list], + ) locks.assert_awaited_once_with(self.db, [2, 5]) site_access.assert_awaited_with(self.db, self.user, 9, for_update=True) warehouse_access.assert_not_awaited() @@ -676,7 +1082,7 @@ class ReturnRequestTransactionRedTest(unittest.IsolatedAsyncioTestCase): self.assertEqual([4, 4], [event.after_line_id for event in events]) self.db.commit.assert_awaited_once() - async def test_request_rejects_mixed_site_and_rolls_back_whole_order(self) -> None: + async def test_request_hides_mixed_site_and_rolls_back_whole_order(self) -> None: inventories = [make_inventory(2), make_inventory(5, site_id=10)] with ( patch.object(AccessDao, 'list_user_site_ids', new=AsyncMock(return_value=[9])), @@ -696,15 +1102,45 @@ class ReturnRequestTransactionRedTest(unittest.IsolatedAsyncioTestCase): new=AsyncMock(side_effect=[make_filling(2), make_filling(5)]), ), patch.object(OperationDao, 'add_return_order', new=AsyncMock()) as add_order, - self.assertRaises(ServiceException) as raised, + self.assertRaises(PermissionException) as raised, ): await ReturnService.request_return_services(self.db, self.page, self.user) - self.assertIn('同一现场', raised.exception.message) + self.assertEqual('无权访问该托盘', raised.exception.message) add_order.assert_not_awaited() self.db.commit.assert_not_awaited() self.db.rollback.assert_awaited() + async def test_admin_request_keeps_precise_missing_and_foreign_site_errors(self) -> None: + admin = make_user(1) + cases = ( + ([], '存在不存在或库存缺失的托盘'), + ([make_inventory(2, site_id=10), make_inventory(5, site_id=10)], '同一退回单托盘必须属于同一现场'), + ) + for inventories, expected_message in cases: + with self.subTest(expected_message=expected_message): + db = MagicMock(spec=AsyncSession) + with ( + patch.object( + OperationDao, + 'get_return_by_request_idempotency_key', + new=AsyncMock(return_value=None), + ), + patch.object(AccessService, 'assert_site_access', new=AsyncMock()), + patch.object( + InventoryDao, + 'get_inventories_for_update', + new=AsyncMock(return_value=inventories), + ), + patch.object(InventoryDao, 'get_pallets_by_ids', new=AsyncMock()) as pallet_locks, + self.assertRaises(ServiceException) as raised, + ): + await ReturnService.request_return_services(db, self.page, admin) + self.assertEqual(expected_message, raised.exception.message) + pallet_locks.assert_not_awaited() + db.commit.assert_not_awaited() + db.rollback.assert_awaited() + async def test_request_same_key_same_hash_returns_original_and_different_hash_conflicts(self) -> None: request_hash = ReturnService._build_request_hash(self.page) exact = make_return_order(request_hash=request_hash) @@ -743,12 +1179,13 @@ class ReturnRequestTransactionRedTest(unittest.IsolatedAsyncioTestCase): conflict_db.rollback.assert_awaited() async def test_request_checks_site_mapping_before_locking_pallets(self) -> None: + key_read = AsyncMock(return_value=None) with ( patch.object(AccessDao, 'list_user_site_ids', new=AsyncMock(return_value=[9])), patch.object( OperationDao, 'get_return_by_request_idempotency_key', - new=AsyncMock(return_value=None), + new=key_read, ), patch.object( AccessService, @@ -770,10 +1207,30 @@ class ReturnRequestTransactionRedTest(unittest.IsolatedAsyncioTestCase): raised = None self.assertIsInstance(raised, PermissionException) - self.assertEqual('无权访问该客户现场', raised.message) + self.assertEqual('无权访问该托盘', raised.message) + key_read.assert_not_awaited() pallet_locks.assert_not_awaited() self.db.rollback.assert_awaited() + async def test_request_rejects_unscoped_requested_site_before_inventory_locks(self) -> None: + key_read = AsyncMock(return_value=None) + access = AsyncMock() + inventory_locks = AsyncMock() + with ( + patch.object(AccessDao, 'list_user_site_ids', new=AsyncMock(return_value=[])), + patch.object(OperationDao, 'get_return_by_request_idempotency_key', new=key_read), + patch.object(AccessService, 'assert_site_access', new=access), + patch.object(InventoryDao, 'get_inventories_for_update', new=inventory_locks), + self.assertRaises(PermissionException) as raised, + ): + await ReturnService.request_return_services(self.db, self.page, self.user) + + self.assertEqual('无权访问该托盘', raised.exception.message) + key_read.assert_not_awaited() + access.assert_not_awaited() + inventory_locks.assert_not_awaited() + self.db.rollback.assert_awaited_once() + async def test_admin_request_idempotency_scope_is_unrestricted(self) -> None: admin = make_user(1) request_hash = ReturnService._build_request_hash(self.page)