From 80c22881226a2d18814cd1b53ad578fc01a89204 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:23:04 +0800 Subject: [PATCH] =?UTF-8?q?=E4=BF=AE=E5=A4=8D=EF=BC=9A=E6=94=AF=E6=8C=81?= =?UTF-8?q?=E5=B7=B2=E7=A1=AE=E8=AE=A4=E9=80=80=E5=9B=9E=E7=9A=84=E5=B9=B6?= =?UTF-8?q?=E5=8F=91=E5=B9=82=E7=AD=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../module_business/service/return_service.py | 44 ++++--- .../module_business/test_return_service.py | 121 +++++++++++++++++- 2 files changed, 142 insertions(+), 23 deletions(-) diff --git a/shierfeng-fastapi-backend/module_business/service/return_service.py b/shierfeng-fastapi-backend/module_business/service/return_service.py index ea98792..41ec3ef 100644 --- a/shierfeng-fastapi-backend/module_business/service/return_service.py +++ b/shierfeng-fastapi-backend/module_business/service/return_service.py @@ -446,14 +446,13 @@ class ReturnService: pallet_ids: list[int], source_site_id: int, current_user: CurrentUserModel, + inventories: list[BizPalletInventory], ) -> 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) inventory_by_id = {inventory.pallet_id: inventory for inventory in inventories} if set(inventory_by_id) != set(ordered_ids): cls._raise_pallet_not_accessible(current_user, '存在不存在或库存缺失的托盘') @@ -461,13 +460,11 @@ class ReturnService: ordered_inventories = [inventory_by_id[pallet_id] for pallet_id in ordered_ids] 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) + if any( + inventory.warehouse_id is not None or inventory.flow_state not in cls._REQUESTABLE_STATES + for inventory in ordered_inventories + ): + raise ServiceException(message='托盘当前状态不允许发起退回') pallets = await InventoryDao.get_pallets_by_ids(query_db, ordered_ids) pallet_by_id = {pallet.pallet_id: pallet for pallet in pallets} @@ -485,7 +482,7 @@ class ReturnService: ) for filling_id in filling_ids: 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 + return ordered_inventories, ordered_pallets, filling_by_id @classmethod def _validate_locked_request_pallets( @@ -618,7 +615,7 @@ class ReturnService: await query_db.rollback() @classmethod - async def request_return_services( # noqa: PLR0915 + async def request_return_services( # noqa: PLR0912, PLR0915 cls, query_db: AsyncSession, page_object: ReturnRequestModel, @@ -672,12 +669,18 @@ class ReturnService: await query_db.rollback() return cls._success('申请成功', return_id) - inventories, pallets, filling_by_id, fresh_site_ids = await cls._lock_request_pallets( - query_db, - ordered_ids, - page_object.source_site_id, - current_user, - ) + inventories = await InventoryDao.get_inventories_for_update(query_db, ordered_ids) + try: + await AccessService.assert_site_access( + query_db, + current_user, + page_object.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 page_object.source_site_id not in fresh_site_ids: + cls._raise_pallet_not_accessible(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, @@ -693,6 +696,13 @@ class ReturnService: await query_db.rollback() return cls._success('申请成功', fresh_existing_id) + inventories, pallets, filling_by_id = await cls._lock_request_pallets( + query_db, + ordered_ids, + page_object.source_site_id, + current_user, + inventories, + ) await cls._lock_new_request_route(query_db, page_object, current_user) cls._validate_locked_request_pallets( inventories, 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 02840a9..22f6f7e 100644 --- a/shierfeng-fastapi-backend/tests/module_business/test_return_service.py +++ b/shierfeng-fastapi-backend/tests/module_business/test_return_service.py @@ -696,16 +696,18 @@ class ReturnRequestTransactionRedTest(unittest.IsolatedAsyncioTestCase): key_read = AsyncMock(side_effect=read_key) scope_read = AsyncMock(side_effect=read_scope) + pallet_locks = AsyncMock(side_effect=lock_pallets) + filling_locks = AsyncMock(side_effect=lock_filling) 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(InventoryDao, 'get_pallets_by_ids', new=pallet_locks), patch.object( OperationDao, 'get_active_filling_for_update', - new=AsyncMock(side_effect=lock_filling), + new=filling_locks, ), patch.object(AccessService, 'assert_site_access', new=AsyncMock(side_effect=check_access)), patch.object(MasterDao, 'get_site', new=AsyncMock(side_effect=lock_site)), @@ -728,8 +730,6 @@ class ReturnRequestTransactionRedTest(unittest.IsolatedAsyncioTestCase): 'inventory-lock', 'access-fresh', 'scope-fresh', - 'pallet-lock', - 'filling-lock', 'key-fresh', 'site-lock', 'mapping-lock', @@ -739,6 +739,8 @@ class ReturnRequestTransactionRedTest(unittest.IsolatedAsyncioTestCase): ) self.assertEqual(2, key_read.await_count) self.assertEqual(2, scope_read.await_count) + pallet_locks.assert_not_awaited() + filling_locks.assert_not_awaited() add_order.assert_not_awaited() add_items.assert_not_awaited() add_event.assert_not_awaited() @@ -779,6 +781,109 @@ class ReturnRequestTransactionRedTest(unittest.IsolatedAsyncioTestCase): self.db.commit.assert_not_awaited() self.assertEqual(2, self.db.rollback.await_count) + async def test_concurrent_confirmed_winner_returns_before_inventory_business_validation(self) -> None: + winner = make_return_order( + status=ReturnStatus.CONFIRMED.value, + request_hash=ReturnService._build_request_hash(self.page), + ) + relocated = [ + make_inventory( + 2, + site_id=None, + warehouse_id=7, + line_id=None, + state=PalletFlowState.FILLED_IN_WAREHOUSE.value, + ), + make_inventory( + 5, + site_id=None, + warehouse_id=7, + line_id=None, + remaining=0, + filling_id=None, + state=PalletFlowState.EMPTY_IN_WAREHOUSE.value, + ), + ] + key_read = AsyncMock(side_effect=[None, winner]) + pallet_locks = AsyncMock() + filling_locks = AsyncMock() + 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=key_read), + patch.object(InventoryDao, 'get_inventories_for_update', new=AsyncMock(return_value=relocated)), + patch.object(InventoryDao, 'get_pallets_by_ids', new=pallet_locks), + patch.object(OperationDao, 'get_active_filling_for_update', new=filling_locks), + patch.object(MasterDao, 'get_site', new=AsyncMock(return_value=SimpleNamespace(status='0'))), + patch.object(OperationDao, 'add_return_order', new=AsyncMock()) as add_order, + patch.object(InventoryDao, 'add_event', new=AsyncMock()) as add_event, + ): + try: + result = await ReturnService.request_return_services(self.db, self.page, self.user) + except PermissionException as exc: + self.fail(f'已确认赢家在 fresh key 查询前被库存位置提前隐藏：{exc.message}') + + self.assertEqual(13, result.result) + self.assertEqual(2, key_read.await_count) + pallet_locks.assert_not_awaited() + filling_locks.assert_not_awaited() + add_order.assert_not_awaited() + add_event.assert_not_awaited() + self.db.commit.assert_not_awaited() + self.assertEqual(2, self.db.rollback.await_count) + + async def test_concurrent_confirmed_winner_different_payload_is_stable_conflict(self) -> None: + winner = make_return_order( + status=ReturnStatus.CONFIRMED.value, + request_hash=ReturnService._build_request_hash(self.page), + ) + conflict_page = self.page.model_copy(update={'target_warehouse_id': 8}) + relocated = [ + make_inventory( + 2, + site_id=None, + warehouse_id=7, + line_id=None, + state=PalletFlowState.FILLED_IN_WAREHOUSE.value, + ), + make_inventory( + 5, + site_id=None, + warehouse_id=7, + line_id=None, + remaining=0, + filling_id=None, + state=PalletFlowState.EMPTY_IN_WAREHOUSE.value, + ), + ] + key_read = AsyncMock(side_effect=[None, winner]) + pallet_locks = AsyncMock() + filling_locks = AsyncMock() + 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=key_read), + patch.object(InventoryDao, 'get_inventories_for_update', new=AsyncMock(return_value=relocated)), + patch.object(InventoryDao, 'get_pallets_by_ids', new=pallet_locks), + patch.object(OperationDao, 'get_active_filling_for_update', new=filling_locks), + patch.object(OperationDao, 'add_return_order', new=AsyncMock()) as add_order, + ): + try: + await ReturnService.request_return_services(self.db, conflict_page, self.user) + except Exception as exc: + raised = exc + else: + raised = None + + self.assertIsInstance(raised, ServiceException) + self.assertEqual('申请退回失败，幂等键已被其他请求使用', getattr(raised, 'message', None)) + self.assertEqual(2, key_read.await_count) + pallet_locks.assert_not_awaited() + filling_locks.assert_not_awaited() + 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 = [ @@ -903,12 +1008,14 @@ class ReturnRequestTransactionRedTest(unittest.IsolatedAsyncioTestCase): with self.subTest(case=name): db = MagicMock(spec=AsyncSession) pallet_locks = AsyncMock(return_value=[make_pallet(2), make_pallet(5)]) + scope_read = AsyncMock(return_value=[9]) + key_read = AsyncMock(return_value=None) with ( - patch.object(AccessDao, 'list_user_site_ids', new=AsyncMock(return_value=[9])), + patch.object(AccessDao, 'list_user_site_ids', new=scope_read), patch.object( OperationDao, 'get_return_by_request_idempotency_key', - new=AsyncMock(return_value=None), + new=key_read, ), patch.object(AccessService, 'assert_site_access', new=AsyncMock()), patch.object( @@ -926,6 +1033,8 @@ class ReturnRequestTransactionRedTest(unittest.IsolatedAsyncioTestCase): raised = None self.assertIsInstance(raised, PermissionException) messages.append(getattr(raised, 'message', '未抛出异常')) + self.assertEqual(2, scope_read.await_count) + self.assertEqual(2, key_read.await_count) pallet_locks.assert_not_awaited() db.commit.assert_not_awaited() db.rollback.assert_awaited()