From dc28cf6df004fcdd8e634f3835a39cb76d29bb4d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E7=84=A6=E9=BE=99=E8=A8=80?= Date: Sun, 26 Jul 2026 20:06:35 +0800 Subject: [PATCH] =?UTF-8?q?=E4=BF=AE=E5=A4=8D=EF=BC=9A=E5=BC=BA=E5=8C=96?= =?UTF-8?q?=E5=8F=91=E8=B4=A7=E6=9D=83=E9=99=90=E4=B8=8E=E5=B9=82=E7=AD=89?= =?UTF-8?q?=E4=B8=80=E8=87=B4=E6=80=A7?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../service/shipment_service.py | 166 +++- .../module_business/test_shipment_service.py | 735 +++++++++++++++++- 2 files changed, 860 insertions(+), 41 deletions(-) diff --git a/shierfeng-fastapi-backend/module_business/service/shipment_service.py b/shierfeng-fastapi-backend/module_business/service/shipment_service.py index 69a7a87..fb6b2c3 100644 --- a/shierfeng-fastapi-backend/module_business/service/shipment_service.py +++ b/shierfeng-fastapi-backend/module_business/service/shipment_service.py @@ -476,6 +476,15 @@ class ShipmentService: ordered_ids, page_object.source_warehouse_id, ) + await cls._lock_and_authorize_routes( + query_db, + current_user, + warehouse_ids={page_object.source_warehouse_id}, + site_ids={page_object.target_site_id}, + hidden_warehouse_ids=set(), + hidden_site_ids=set(), + require_active=True, + ) current_time = datetime.now() shipment = BizShipmentOrder( @@ -538,6 +547,21 @@ class ShipmentService: ordered_ids, page_object.source_warehouse_id, ) + await cls._lock_and_authorize_routes( + query_db, + current_user, + warehouse_ids={ + shipment.source_warehouse_id, + page_object.source_warehouse_id, + }, + site_ids={ + shipment.target_site_id, + page_object.target_site_id, + }, + hidden_warehouse_ids={shipment.source_warehouse_id}, + hidden_site_ids={shipment.target_site_id}, + require_active=True, + ) current_time = datetime.now() shipment.source_warehouse_id = page_object.source_warehouse_id @@ -576,6 +600,7 @@ class ShipmentService: ordered_ids = sorted(shipment_ids) warehouse_ids, site_ids = await cls._scopes(query_db, current_user) + locked_shipments = [] for shipment_id in ordered_ids: shipment = await OperationDao.get_shipment_for_update( query_db, @@ -587,6 +612,19 @@ class ShipmentService: cls._raise_shipment_not_accessible(current_user) if shipment.status != DocumentStatus.DRAFT.value: raise ServiceException(message='已确认发货单不允许删除') + locked_shipments.append(shipment) + + route_warehouse_ids = {shipment.source_warehouse_id for shipment in locked_shipments} + route_site_ids = {shipment.target_site_id for shipment in locked_shipments} + await cls._lock_and_authorize_routes( + query_db, + current_user, + warehouse_ids=route_warehouse_ids, + site_ids=route_site_ids, + hidden_warehouse_ids=route_warehouse_ids, + hidden_site_ids=route_site_ids, + require_active=False, + ) for shipment_id in ordered_ids: await OperationDao.delete_shipment(query_db, shipment_id) @@ -613,50 +651,80 @@ class ShipmentService: and shipment.confirm_request_hash == request_hash ) + @classmethod + async def _lock_and_authorize_routes( + cls, + query_db: AsyncSession, + current_user: CurrentUserModel, + *, + warehouse_ids: set[int], + site_ids: set[int], + hidden_warehouse_ids: set[int], + hidden_site_ids: set[int], + require_active: bool, + ) -> None: + for warehouse_id in sorted(warehouse_ids): + warehouse = await MasterDao.get_warehouse( + query_db, + warehouse_id, + for_update=True, + ) + if warehouse is None: + raise ServiceException(message='来源仓库不存在') + if require_active and warehouse.status != '0': + raise ServiceException(message='来源仓库已停用') + try: + await AccessService.assert_warehouse_access( + query_db, + current_user, + warehouse_id, + for_update=True, + ) + except PermissionException: + if warehouse_id in hidden_warehouse_ids: + cls._raise_shipment_not_accessible(current_user) + raise + + for site_id in sorted(site_ids): + site = await MasterDao.get_site( + query_db, + site_id, + for_update=True, + ) + if site is None: + raise ServiceException(message='目标现场不存在') + if require_active and site.status != '0': + raise ServiceException(message='目标现场已停用') + try: + await AccessService.assert_site_access( + query_db, + current_user, + site_id, + for_update=True, + ) + except PermissionException: + if site_id in hidden_site_ids: + cls._raise_shipment_not_accessible(current_user) + raise + @classmethod async def _lock_and_reauthorize_route( cls, query_db: AsyncSession, shipment: BizShipmentOrder, current_user: CurrentUserModel, + *, + require_active: bool = True, ) -> None: - warehouse = await MasterDao.get_warehouse( + await cls._lock_and_authorize_routes( query_db, - shipment.source_warehouse_id, - for_update=True, + current_user, + warehouse_ids={shipment.source_warehouse_id}, + site_ids={shipment.target_site_id}, + hidden_warehouse_ids={shipment.source_warehouse_id}, + hidden_site_ids={shipment.target_site_id}, + require_active=require_active, ) - if warehouse is None: - raise ServiceException(message='来源仓库不存在') - if warehouse.status != '0': - raise ServiceException(message='来源仓库已停用') - try: - await AccessService.assert_warehouse_access( - query_db, - current_user, - shipment.source_warehouse_id, - for_update=True, - ) - except PermissionException: - cls._raise_shipment_not_accessible(current_user) - - site = await MasterDao.get_site( - query_db, - shipment.target_site_id, - for_update=True, - ) - if site is None: - raise ServiceException(message='目标现场不存在') - if site.status != '0': - raise ServiceException(message='目标现场已停用') - try: - await AccessService.assert_site_access( - query_db, - current_user, - shipment.target_site_id, - for_update=True, - ) - except PermissionException: - cls._raise_shipment_not_accessible(current_user) @classmethod async def _recover_confirm_integrity_error( @@ -684,11 +752,26 @@ class ShipmentService: request_hash, ): raise ServiceException(message='确认发货失败，幂等键已被其他请求使用') - await cls._validate_active_route( + locked_recovered = await OperationDao.get_shipment_for_update( query_db, - recovered.source_warehouse_id, - recovered.target_site_id, + shipment_id, + warehouse_ids=warehouse_ids, + site_ids=site_ids, + ) + if locked_recovered is None: + cls._raise_shipment_not_accessible(current_user) + if not cls._is_exact_confirmed( + locked_recovered, + shipment_id, + idempotency_key, + request_hash, + ): + raise ServiceException(message='确认发货失败，幂等键已被其他请求使用') + await cls._lock_and_reauthorize_route( + query_db, + locked_recovered, current_user, + require_active=False, ) return cls._success('确认成功') finally: @@ -745,7 +828,12 @@ class ShipmentService: idempotency_key, request_hash, ): - await cls._lock_and_reauthorize_route(query_db, locked_shipment, current_user) + await cls._lock_and_reauthorize_route( + query_db, + locked_shipment, + current_user, + require_active=False, + ) await query_db.rollback() return cls._success('确认成功') raise ServiceException(message='发货单已确认，幂等信息不匹配') diff --git a/shierfeng-fastapi-backend/tests/module_business/test_shipment_service.py b/shierfeng-fastapi-backend/tests/module_business/test_shipment_service.py index a4414ba..6beda9e 100644 --- a/shierfeng-fastapi-backend/tests/module_business/test_shipment_service.py +++ b/shierfeng-fastapi-backend/tests/module_business/test_shipment_service.py @@ -745,6 +745,10 @@ class ShipmentDraftServiceTest(unittest.IsolatedAsyncioTestCase): 'pallet-lock', 'filling-lock-102', 'filling-lock-105', + 'warehouse-read', + 'warehouse-access', + 'site-read', + 'site-access', 'add-order', 'replace-items', 'commit', @@ -756,6 +760,114 @@ class ShipmentDraftServiceTest(unittest.IsolatedAsyncioTestCase): self.db.commit.assert_awaited_once() self.db.rollback.assert_not_awaited() + async def test_add_rechecks_revoked_mapping_after_pallet_locks_before_writing(self) -> None: + page = ShipmentCreateModel(sourceWarehouseId=7, targetSiteId=9, palletIds=[2]) + + async def warehouse_access(*_args: Any, **kwargs: Any) -> None: + if kwargs.get('for_update'): + raise PermissionException(data='', message='无权访问该仓库') + + add_order = AsyncMock(return_value=make_shipment(pallet_ids=(2,))) + replace_items = AsyncMock() + warehouse_mapping = AsyncMock(side_effect=warehouse_access) + with ( + patch.object(AccessService, 'assert_warehouse_access', new=warehouse_mapping), + patch.object(AccessService, 'assert_site_access', new=AsyncMock()), + patch.object( + MasterDao, + 'get_warehouse', + new=AsyncMock( + return_value=BizWarehouse( + warehouse_id=7, + warehouse_code='W-7', + warehouse_name='七号仓', + status='0', + ) + ), + ) as warehouse_read, + patch.object( + MasterDao, + 'get_site', + new=AsyncMock( + return_value=BizCustomerSite( + site_id=9, + customer_id=1, + site_code='S-9', + site_name='九号现场', + status='0', + ) + ), + ), + patch.object( + ShipmentService, + '_lock_and_validate_pallets', + new=AsyncMock(return_value=([], [], [])), + ), + patch.object(OperationDao, 'add_shipment', new=add_order), + patch.object(OperationDao, 'replace_shipment_items', new=replace_items), + self.assertRaises(PermissionException) as raised, + ): + await ShipmentService.add_shipment_services(self.db, page, self.user) + + self.assertEqual('无权访问该仓库', raised.exception.message) + self.assertEqual(2, warehouse_read.await_count) + warehouse_read.assert_awaited_with(self.db, 7, for_update=True) + self.assertEqual(2, warehouse_mapping.await_count) + warehouse_mapping.assert_awaited_with(self.db, self.user, 7, for_update=True) + add_order.assert_not_awaited() + replace_items.assert_not_awaited() + self.db.commit.assert_not_awaited() + self.db.rollback.assert_awaited_once() + + async def test_add_rechecks_disabled_master_after_pallet_locks_before_writing(self) -> None: + page = ShipmentCreateModel(sourceWarehouseId=7, targetSiteId=9, palletIds=[2]) + + async def get_warehouse(*_args: Any, **kwargs: Any) -> BizWarehouse: + return BizWarehouse( + warehouse_id=7, + warehouse_code='W-7', + warehouse_name='七号仓', + status='1' if kwargs.get('for_update') else '0', + ) + + add_order = AsyncMock(return_value=make_shipment(pallet_ids=(2,))) + replace_items = AsyncMock() + warehouse_read = AsyncMock(side_effect=get_warehouse) + with ( + patch.object(AccessService, 'assert_warehouse_access', new=AsyncMock()), + patch.object(AccessService, 'assert_site_access', new=AsyncMock()), + patch.object(MasterDao, 'get_warehouse', new=warehouse_read), + patch.object( + MasterDao, + 'get_site', + new=AsyncMock( + return_value=BizCustomerSite( + site_id=9, + customer_id=1, + site_code='S-9', + site_name='九号现场', + status='0', + ) + ), + ), + patch.object( + ShipmentService, + '_lock_and_validate_pallets', + new=AsyncMock(return_value=([], [], [])), + ), + patch.object(OperationDao, 'add_shipment', new=add_order), + patch.object(OperationDao, 'replace_shipment_items', new=replace_items), + self.assertRaises(ServiceException) as raised, + ): + await ShipmentService.add_shipment_services(self.db, page, self.user) + + self.assertEqual('来源仓库已停用', raised.exception.message) + warehouse_read.assert_awaited_with(self.db, 7, for_update=True) + add_order.assert_not_awaited() + replace_items.assert_not_awaited() + self.db.commit.assert_not_awaited() + self.db.rollback.assert_awaited_once() + async def test_edit_revalidates_current_draft_and_keeps_snapshots_null(self) -> None: page = ShipmentUpdateModel( shipmentId=12, @@ -848,6 +960,229 @@ class ShipmentDraftServiceTest(unittest.IsolatedAsyncioTestCase): self.assertEqual('FILLED_IN_WAREHOUSE', inventory.flow_state) self.db.commit.assert_awaited_once() + async def test_edit_rechecks_old_and_new_routes_before_writing(self) -> None: + scenarios = ( + ('current-warehouse-revoked', PermissionException, '无权访问该发货单'), + ('new-site-revoked', PermissionException, '无权访问该客户现场'), + ('new-site-disabled', ServiceException, '目标现场已停用'), + ) + for scenario, exception_type, expected_message in scenarios: + with self.subTest(scenario=scenario): + db = MagicMock(spec=AsyncSession) + shipment = make_shipment(pallet_ids=(2,)) + page = ShipmentUpdateModel( + shipmentId=12, + sourceWarehouseId=8, + targetSiteId=10, + palletIds=[2], + ) + current_warehouse_id = shipment.source_warehouse_id + new_site_id = page.target_site_id + + async def warehouse_access( + _db: AsyncSession, + _user: CurrentUserModel, + warehouse_id: int, + _scenario: str = scenario, + _current_warehouse_id: int = current_warehouse_id, + **kwargs: Any, + ) -> None: + if ( + _scenario == 'current-warehouse-revoked' + and warehouse_id == _current_warehouse_id + and kwargs.get('for_update') + ): + raise PermissionException(data='', message='无权访问该仓库') + + async def site_access( + _db: AsyncSession, + _user: CurrentUserModel, + site_id: int, + _scenario: str = scenario, + _new_site_id: int = new_site_id, + **kwargs: Any, + ) -> None: + if _scenario == 'new-site-revoked' and site_id == _new_site_id and kwargs.get('for_update'): + raise PermissionException(data='', message='无权访问该客户现场') + + async def get_warehouse( + _db: AsyncSession, + warehouse_id: int, + **_kwargs: Any, + ) -> BizWarehouse: + return BizWarehouse( + warehouse_id=warehouse_id, + warehouse_code=f'W-{warehouse_id}', + warehouse_name=f'{warehouse_id}号仓', + status='0', + ) + + async def get_site( + _db: AsyncSession, + site_id: int, + _scenario: str = scenario, + _new_site_id: int = new_site_id, + **kwargs: Any, + ) -> BizCustomerSite: + return BizCustomerSite( + site_id=site_id, + customer_id=1, + site_code=f'S-{site_id}', + site_name=f'{site_id}号现场', + status=( + '1' + if _scenario == 'new-site-disabled' and site_id == _new_site_id and kwargs.get('for_update') + else '0' + ), + ) + + update = AsyncMock() + replace_items = AsyncMock() + warehouse_mapping = AsyncMock(side_effect=warehouse_access) + site_mapping = AsyncMock(side_effect=site_access) + site_read = AsyncMock(side_effect=get_site) + with ( + patch.object(AccessDao, 'list_user_warehouse_ids', new=AsyncMock(return_value=[7, 8])), + patch.object(AccessDao, 'list_user_site_ids', new=AsyncMock(return_value=[9, 10])), + patch.object(OperationDao, 'get_shipment_for_update', new=AsyncMock(return_value=shipment)), + patch.object(AccessService, 'assert_warehouse_access', new=warehouse_mapping), + patch.object(AccessService, 'assert_site_access', new=site_mapping), + patch.object(MasterDao, 'get_warehouse', new=AsyncMock(side_effect=get_warehouse)), + patch.object(MasterDao, 'get_site', new=site_read), + patch.object( + ShipmentService, + '_lock_and_validate_pallets', + new=AsyncMock(return_value=([], [], [])), + ), + patch.object(OperationDao, 'update_shipment', new=update), + patch.object(OperationDao, 'replace_shipment_items', new=replace_items), + self.assertRaises(exception_type) as raised, + ): + await ShipmentService.edit_shipment_services(db, page, self.user) + + self.assertEqual(expected_message, raised.exception.message) + if scenario == 'current-warehouse-revoked': + warehouse_mapping.assert_any_await(db, self.user, 7, for_update=True) + else: + site_read.assert_any_await(db, 10, for_update=True) + if scenario == 'new-site-revoked': + site_mapping.assert_any_await(db, self.user, 10, for_update=True) + update.assert_not_awaited() + replace_items.assert_not_awaited() + db.commit.assert_not_awaited() + db.rollback.assert_awaited_once() + + async def test_edit_locks_old_and_new_routes_in_stable_grouped_order(self) -> None: + db = MagicMock(spec=AsyncSession) + shipment = make_shipment( + source_warehouse_id=8, + target_site_id=10, + pallet_ids=(2,), + ) + page = ShipmentUpdateModel( + shipmentId=12, + sourceWarehouseId=7, + targetSiteId=9, + palletIds=[2], + ) + actions: list[str] = [] + + async def warehouse_access( + _db: AsyncSession, + _user: CurrentUserModel, + warehouse_id: int, + **kwargs: Any, + ) -> None: + if kwargs.get('for_update'): + actions.append(f'warehouse-mapping-{warehouse_id}') + + async def site_access( + _db: AsyncSession, + _user: CurrentUserModel, + site_id: int, + **kwargs: Any, + ) -> None: + if kwargs.get('for_update'): + actions.append(f'site-mapping-{site_id}') + + async def get_warehouse( + _db: AsyncSession, + warehouse_id: int, + **kwargs: Any, + ) -> BizWarehouse: + if kwargs.get('for_update'): + actions.append(f'warehouse-master-{warehouse_id}') + return BizWarehouse( + warehouse_id=warehouse_id, + warehouse_code=f'W-{warehouse_id}', + warehouse_name=f'{warehouse_id}号仓', + status='0', + ) + + async def get_site( + _db: AsyncSession, + site_id: int, + **kwargs: Any, + ) -> BizCustomerSite: + if kwargs.get('for_update'): + actions.append(f'site-master-{site_id}') + return BizCustomerSite( + site_id=site_id, + customer_id=1, + site_code=f'S-{site_id}', + site_name=f'{site_id}号现场', + status='0', + ) + + async def lock_pallets(*_args: Any, **_kwargs: Any) -> tuple[list[Any], list[Any], list[Any]]: + actions.append('pallet-locks') + return [], [], [] + + async def update(*_args: Any, **_kwargs: Any) -> BizShipmentOrder: + actions.append('update') + return shipment + + async def replace(*_args: Any, **_kwargs: Any) -> list[BizShipmentItem]: + actions.append('replace') + return [] + + async def commit() -> None: + actions.append('commit') + + db.commit.side_effect = commit + with ( + patch.object(AccessDao, 'list_user_warehouse_ids', new=AsyncMock(return_value=[7, 8])), + patch.object(AccessDao, 'list_user_site_ids', new=AsyncMock(return_value=[9, 10])), + patch.object(OperationDao, 'get_shipment_for_update', new=AsyncMock(return_value=shipment)), + patch.object(AccessService, 'assert_warehouse_access', new=AsyncMock(side_effect=warehouse_access)), + patch.object(AccessService, 'assert_site_access', new=AsyncMock(side_effect=site_access)), + patch.object(MasterDao, 'get_warehouse', new=AsyncMock(side_effect=get_warehouse)), + patch.object(MasterDao, 'get_site', new=AsyncMock(side_effect=get_site)), + patch.object(ShipmentService, '_lock_and_validate_pallets', new=AsyncMock(side_effect=lock_pallets)), + patch.object(OperationDao, 'update_shipment', new=AsyncMock(side_effect=update)), + patch.object(OperationDao, 'replace_shipment_items', new=AsyncMock(side_effect=replace)), + ): + result = await ShipmentService.edit_shipment_services(db, page, self.user) + + self.assertTrue(result.is_success) + self.assertEqual( + [ + 'pallet-locks', + 'warehouse-master-7', + 'warehouse-mapping-7', + 'warehouse-master-8', + 'warehouse-mapping-8', + 'site-master-9', + 'site-mapping-9', + 'site-master-10', + 'site-mapping-10', + 'update', + 'replace', + 'commit', + ], + actions, + ) + async def test_batch_delete_preflights_all_drafts_before_any_delete(self) -> None: documents = { 11: make_shipment(11, status='DRAFT'), @@ -879,6 +1214,173 @@ class ShipmentDraftServiceTest(unittest.IsolatedAsyncioTestCase): self.db.commit.assert_not_awaited() self.db.rollback.assert_awaited_once() + async def test_batch_delete_rechecks_all_current_routes_before_first_delete(self) -> None: + documents = { + 11: make_shipment(11, source_warehouse_id=8, target_site_id=10, status='DRAFT'), + 12: make_shipment(12, source_warehouse_id=7, target_site_id=9, status='DRAFT'), + } + revoked_warehouse_id = documents[11].source_warehouse_id + parent_lock_order: list[int] = [] + + async def lock_order( + _db: AsyncSession, + shipment_id: int, + **_kwargs: Any, + ) -> BizShipmentOrder: + parent_lock_order.append(shipment_id) + return documents[shipment_id] + + async def get_warehouse( + _db: AsyncSession, + warehouse_id: int, + **_kwargs: Any, + ) -> BizWarehouse: + return BizWarehouse( + warehouse_id=warehouse_id, + warehouse_code=f'W-{warehouse_id}', + warehouse_name=f'{warehouse_id}号仓', + status='0', + ) + + async def warehouse_access( + _db: AsyncSession, + _user: CurrentUserModel, + warehouse_id: int, + **kwargs: Any, + ) -> None: + if warehouse_id == revoked_warehouse_id and kwargs.get('for_update'): + raise PermissionException(data='', message='无权访问该仓库') + + delete = AsyncMock() + warehouse_mapping = AsyncMock(side_effect=warehouse_access) + with ( + patch.object(AccessDao, 'list_user_warehouse_ids', new=AsyncMock(return_value=[7, 8])), + patch.object(AccessDao, 'list_user_site_ids', new=AsyncMock(return_value=[9, 10])), + patch.object(OperationDao, 'get_shipment_for_update', new=AsyncMock(side_effect=lock_order)), + patch.object(MasterDao, 'get_warehouse', new=AsyncMock(side_effect=get_warehouse)), + patch.object(AccessService, 'assert_warehouse_access', new=warehouse_mapping), + patch.object(OperationDao, 'delete_shipment', new=delete), + self.assertRaises(PermissionException) as raised, + ): + await ShipmentService.delete_shipment_services(self.db, [12, 11], self.user) + + self.assertEqual('无权访问该发货单', raised.exception.message) + self.assertEqual([11, 12], parent_lock_order) + self.assertEqual([7, 8], [call.args[2] for call in warehouse_mapping.await_args_list]) + self.assertTrue(all(call.kwargs.get('for_update') for call in warehouse_mapping.await_args_list)) + delete.assert_not_awaited() + self.db.commit.assert_not_awaited() + self.db.rollback.assert_awaited_once() + + async def test_batch_delete_locks_deduplicated_routes_stably_without_requiring_active(self) -> None: + db = MagicMock(spec=AsyncSession) + documents = { + 11: make_shipment(11, source_warehouse_id=8, target_site_id=10, status='DRAFT'), + 12: make_shipment(12, source_warehouse_id=7, target_site_id=10, status='DRAFT'), + 13: make_shipment(13, source_warehouse_id=8, target_site_id=9, status='DRAFT'), + } + actions: list[str] = [] + + async def lock_order( + _db: AsyncSession, + shipment_id: int, + **_kwargs: Any, + ) -> BizShipmentOrder: + actions.append(f'parent-{shipment_id}') + return documents[shipment_id] + + async def get_warehouse( + _db: AsyncSession, + warehouse_id: int, + **kwargs: Any, + ) -> BizWarehouse: + self.assertTrue(kwargs.get('for_update')) + actions.append(f'warehouse-master-{warehouse_id}') + return BizWarehouse( + warehouse_id=warehouse_id, + warehouse_code=f'W-{warehouse_id}', + warehouse_name=f'{warehouse_id}号仓', + status='1', + ) + + async def warehouse_access( + _db: AsyncSession, + _user: CurrentUserModel, + warehouse_id: int, + **kwargs: Any, + ) -> None: + self.assertTrue(kwargs.get('for_update')) + actions.append(f'warehouse-mapping-{warehouse_id}') + + async def get_site( + _db: AsyncSession, + site_id: int, + **kwargs: Any, + ) -> BizCustomerSite: + self.assertTrue(kwargs.get('for_update')) + actions.append(f'site-master-{site_id}') + return BizCustomerSite( + site_id=site_id, + customer_id=1, + site_code=f'S-{site_id}', + site_name=f'{site_id}号现场', + status='1', + ) + + async def site_access( + _db: AsyncSession, + _user: CurrentUserModel, + site_id: int, + **kwargs: Any, + ) -> None: + self.assertTrue(kwargs.get('for_update')) + actions.append(f'site-mapping-{site_id}') + + async def delete(_db: AsyncSession, shipment_id: int) -> None: + actions.append(f'delete-{shipment_id}') + + async def commit() -> None: + actions.append('commit') + + db.commit.side_effect = commit + with ( + patch.object(AccessDao, 'list_user_warehouse_ids', new=AsyncMock(return_value=[7, 8])), + patch.object(AccessDao, 'list_user_site_ids', new=AsyncMock(return_value=[9, 10])), + patch.object(OperationDao, 'get_shipment_for_update', new=AsyncMock(side_effect=lock_order)), + patch.object(MasterDao, 'get_warehouse', new=AsyncMock(side_effect=get_warehouse)), + patch.object( + AccessService, + 'assert_warehouse_access', + new=AsyncMock(side_effect=warehouse_access), + ), + patch.object(MasterDao, 'get_site', new=AsyncMock(side_effect=get_site)), + patch.object(AccessService, 'assert_site_access', new=AsyncMock(side_effect=site_access)), + patch.object(OperationDao, 'delete_shipment', new=AsyncMock(side_effect=delete)), + ): + result = await ShipmentService.delete_shipment_services(db, [13, 11, 12], self.user) + + self.assertTrue(result.is_success) + self.assertEqual( + [ + 'parent-11', + 'parent-12', + 'parent-13', + 'warehouse-master-7', + 'warehouse-mapping-7', + 'warehouse-master-8', + 'warehouse-mapping-8', + 'site-master-9', + 'site-mapping-9', + 'site-master-10', + 'site-mapping-10', + 'delete-11', + 'delete-12', + 'delete-13', + 'commit', + ], + actions, + ) + class ShipmentConfirmServiceTest(unittest.IsolatedAsyncioTestCase): def setUp(self) -> None: @@ -1161,6 +1663,153 @@ class ShipmentConfirmServiceTest(unittest.IsolatedAsyncioTestCase): self.db.commit.assert_not_awaited() self.db.rollback.assert_awaited_once() + async def test_exact_confirm_retry_ignores_later_route_disable_but_rechecks_current_access( + self, + ) -> None: + for warehouse_status, site_status in (('1', '0'), ('0', '1')): + with self.subTest(warehouse_status=warehouse_status, site_status=site_status): + db = MagicMock(spec=AsyncSession) + confirmed = make_shipment( + status='CONFIRMED', + idempotency_key='confirm-key', + request_hash=self.request_hash, + ) + for item in confirmed.items: + item.active_filling_id = 100 + item.pallet_id + item.package_qty = 8 + original_snapshots = [ + (item.pallet_id, item.active_filling_id, item.package_qty) for item in confirmed.items + ] + inventory_lock = AsyncMock() + update = AsyncMock() + add_event = AsyncMock() + warehouse_access = AsyncMock() + site_access = AsyncMock() + + with ( + patch.object(AccessDao, 'list_user_warehouse_ids', new=AsyncMock(return_value=[7])), + patch.object(AccessDao, 'list_user_site_ids', new=AsyncMock(return_value=[9])), + patch.object(OperationDao, 'get_shipment_by_id', new=AsyncMock(return_value=confirmed)), + patch.object( + OperationDao, + 'get_shipment_by_confirm_idempotency_key', + new=AsyncMock(return_value=confirmed), + ), + patch.object( + OperationDao, + 'get_shipment_for_update', + new=AsyncMock(return_value=confirmed), + ), + patch.object(InventoryDao, 'get_inventories_for_update', new=inventory_lock), + patch.object( + MasterDao, + 'get_warehouse', + new=AsyncMock( + return_value=BizWarehouse( + warehouse_id=7, + warehouse_code='W-7', + warehouse_name='七号仓', + status=warehouse_status, + ) + ), + ) as warehouse_lock, + patch.object(AccessService, 'assert_warehouse_access', new=warehouse_access), + patch.object( + MasterDao, + 'get_site', + new=AsyncMock( + return_value=BizCustomerSite( + site_id=9, + customer_id=1, + site_code='S-9', + site_name='九号现场', + status=site_status, + ) + ), + ) as site_lock, + patch.object(AccessService, 'assert_site_access', new=site_access), + patch.object(OperationDao, 'update_shipment', new=update), + patch.object(InventoryDao, 'add_event', new=add_event), + ): + result = await ShipmentService.confirm_shipment_services( + db, + 12, + self.command, + self.user, + ) + + self.assertTrue(result.is_success) + warehouse_lock.assert_awaited_once_with(db, 7, for_update=True) + warehouse_access.assert_awaited_once_with(db, self.user, 7, for_update=True) + site_lock.assert_awaited_once_with(db, 9, for_update=True) + site_access.assert_awaited_once_with(db, self.user, 9, for_update=True) + self.assertEqual( + original_snapshots, + [(item.pallet_id, item.active_filling_id, item.package_qty) for item in confirmed.items], + ) + inventory_lock.assert_not_awaited() + update.assert_not_awaited() + add_event.assert_not_awaited() + db.commit.assert_not_awaited() + db.rollback.assert_awaited_once() + + async def test_exact_confirm_retry_hides_document_when_current_mapping_is_revoked(self) -> None: + confirmed = make_shipment( + status='CONFIRMED', + idempotency_key='confirm-key', + request_hash=self.request_hash, + ) + inventory_lock = AsyncMock() + update = AsyncMock() + add_event = AsyncMock() + site_lock = AsyncMock() + warehouse_mapping = AsyncMock(side_effect=PermissionException(data='', message='无权访问该仓库')) + with ( + patch.object(AccessDao, 'list_user_warehouse_ids', new=AsyncMock(return_value=[7])), + patch.object(AccessDao, 'list_user_site_ids', new=AsyncMock(return_value=[9])), + patch.object(OperationDao, 'get_shipment_by_id', new=AsyncMock(return_value=confirmed)), + patch.object( + OperationDao, + 'get_shipment_by_confirm_idempotency_key', + new=AsyncMock(return_value=confirmed), + ), + patch.object(OperationDao, 'get_shipment_for_update', new=AsyncMock(return_value=confirmed)), + patch.object(InventoryDao, 'get_inventories_for_update', new=inventory_lock), + patch.object( + MasterDao, + 'get_warehouse', + new=AsyncMock( + return_value=BizWarehouse( + warehouse_id=7, + warehouse_code='W-7', + warehouse_name='七号仓', + status='1', + ) + ), + ) as warehouse_lock, + patch.object(AccessService, 'assert_warehouse_access', new=warehouse_mapping), + patch.object(MasterDao, 'get_site', new=site_lock), + patch.object(OperationDao, 'update_shipment', new=update), + patch.object(InventoryDao, 'add_event', new=add_event), + self.assertRaises(PermissionException) as raised, + ): + await ShipmentService.confirm_shipment_services( + self.db, + 12, + self.command, + self.user, + ) + + self.assertEqual('无权访问该发货单', raised.exception.message) + warehouse_lock.assert_awaited_once_with(self.db, 7, for_update=True) + warehouse_mapping.assert_awaited_once_with(self.db, self.user, 7, for_update=True) + site_lock.assert_not_awaited() + inventory_lock.assert_not_awaited() + update.assert_not_awaited() + add_event.assert_not_awaited() + self.db.commit.assert_not_awaited() + self.db.rollback.assert_awaited_once() + async def test_confirm_uses_locked_current_items_not_initial_rr_snapshot(self) -> None: initial = make_shipment(pallet_ids=(99,)) locked = make_shipment(pallet_ids=(2,)) @@ -1265,7 +1914,11 @@ class ShipmentConfirmServiceTest(unittest.IsolatedAsyncioTestCase): 'get_shipment_by_confirm_idempotency_key', new=AsyncMock(side_effect=[None, nonexact]), ), - patch.object(OperationDao, 'get_shipment_for_update', new=AsyncMock(side_effect=integrity_error())), + patch.object( + OperationDao, + 'get_shipment_for_update', + new=AsyncMock(side_effect=integrity_error()), + ), self.assertRaises(ServiceException) as raised, ): await ShipmentService.confirm_shipment_services(self.db, 12, self.command, self.user) @@ -1274,6 +1927,80 @@ class ShipmentConfirmServiceTest(unittest.IsolatedAsyncioTestCase): self.assertEqual(2, self.db.rollback.await_count) self.db.commit.assert_not_awaited() + async def test_integrity_recovery_locks_parent_and_rechecks_revoked_current_mapping(self) -> None: + initial = make_shipment(status='DRAFT') + recovered = make_shipment( + status='CONFIRMED', + idempotency_key='confirm-key', + request_hash=self.request_hash, + ) + parent_lock_count = 0 + + async def lock_parent(*_args: Any, **_kwargs: Any) -> BizShipmentOrder: + nonlocal parent_lock_count + parent_lock_count += 1 + if parent_lock_count == 1: + raise integrity_error() + return recovered + + async def warehouse_access(*_args: Any, **kwargs: Any) -> None: + if kwargs.get('for_update'): + raise PermissionException(data='', message='无权访问该仓库') + + parent_lock = AsyncMock(side_effect=lock_parent) + warehouse_mapping = AsyncMock(side_effect=warehouse_access) + site_lock = AsyncMock( + return_value=BizCustomerSite( + site_id=9, + customer_id=1, + site_code='S-9', + site_name='九号现场', + status='0', + ) + ) + with ( + patch.object(AccessDao, 'list_user_warehouse_ids', new=AsyncMock(side_effect=[[7], [7]])), + patch.object(AccessDao, 'list_user_site_ids', new=AsyncMock(side_effect=[[9], [9]])), + patch.object(OperationDao, 'get_shipment_by_id', new=AsyncMock(return_value=initial)), + patch.object( + OperationDao, + 'get_shipment_by_confirm_idempotency_key', + new=AsyncMock(side_effect=[None, recovered]), + ), + patch.object(OperationDao, 'get_shipment_for_update', new=parent_lock), + patch.object( + MasterDao, + 'get_warehouse', + new=AsyncMock( + return_value=BizWarehouse( + warehouse_id=7, + warehouse_code='W-7', + warehouse_name='七号仓', + status='0', + ) + ), + ) as warehouse_lock, + patch.object(AccessService, 'assert_warehouse_access', new=warehouse_mapping), + patch.object(MasterDao, 'get_site', new=site_lock), + patch.object(AccessService, 'assert_site_access', new=AsyncMock()), + self.assertRaises(PermissionException) as raised, + ): + await ShipmentService.confirm_shipment_services(self.db, 12, self.command, self.user) + + self.assertEqual('无权访问该发货单', raised.exception.message) + self.assertEqual(2, parent_lock.await_count) + parent_lock.assert_awaited_with( + self.db, + 12, + warehouse_ids=[7], + site_ids=[9], + ) + warehouse_lock.assert_awaited_once_with(self.db, 7, for_update=True) + warehouse_mapping.assert_awaited_once_with(self.db, self.user, 7, for_update=True) + site_lock.assert_not_awaited() + self.db.commit.assert_not_awaited() + self.assertEqual(2, self.db.rollback.await_count) + async def test_disabled_source_after_pallet_locks_rolls_back_before_mutation(self) -> None: shipment = make_shipment(pallet_ids=(2,)) inventory = make_inventory(2, version=4) @@ -1338,7 +2065,11 @@ class ShipmentConfirmServiceTest(unittest.IsolatedAsyncioTestCase): patch.object(AccessDao, 'list_user_site_ids', new=site_scope), patch.object(OperationDao, 'get_shipment_by_id', new=AsyncMock(return_value=initial)), patch.object(OperationDao, 'get_shipment_by_confirm_idempotency_key', new=key_read), - patch.object(OperationDao, 'get_shipment_for_update', new=AsyncMock(side_effect=integrity_error())), + patch.object( + OperationDao, + 'get_shipment_for_update', + new=AsyncMock(side_effect=[integrity_error(), recovered]), + ), patch.object(AccessService, 'assert_warehouse_access', new=AsyncMock()), patch.object(AccessService, 'assert_site_access', new=AsyncMock()), patch.object(