From f6e994d95f06023e1239f7e55b9c6bd516807081 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 17:56:58 +0800 Subject: [PATCH] =?UTF-8?q?=E4=BF=AE=E5=A4=8D=EF=BC=9A=E5=BC=BA=E5=8C=96?= =?UTF-8?q?=E8=A3=85=E5=A1=AB=E4=BA=8B=E5=8A=A1=E4=B8=80=E8=87=B4=E6=80=A7?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../module_business/dao/master_dao.py | 18 ++- .../service/filling_service.py | 43 ++++-- .../module_business/test_filling_service.py | 125 ++++++++++++++++-- .../module_business/test_master_service.py | 18 +++ 4 files changed, 182 insertions(+), 22 deletions(-) diff --git a/shierfeng-fastapi-backend/module_business/dao/master_dao.py b/shierfeng-fastapi-backend/module_business/dao/master_dao.py index 82d1687..6366d22 100644 --- a/shierfeng-fastapi-backend/module_business/dao/master_dao.py +++ b/shierfeng-fastapi-backend/module_business/dao/master_dao.py @@ -117,7 +117,7 @@ class MasterDao: ) -> Any | None: statement = select(model).where(id_column == resource_id) if for_update: - statement = statement.with_for_update() + statement = statement.with_for_update().execution_options(populate_existing=True) result = await db.execute(statement) return result.scalar_one_or_none() @@ -187,8 +187,20 @@ class MasterDao: return await cls._paginate(db, statement, query_object, is_page) @classmethod - async def get_strain(cls, db: AsyncSession, strain_id: int) -> BizStrain | None: - return await cls._get(db, BizStrain, BizStrain.strain_id, strain_id) + async def get_strain( + cls, + db: AsyncSession, + strain_id: int, + *, + for_update: bool = False, + ) -> BizStrain | None: + return await cls._get( + db, + BizStrain, + BizStrain.strain_id, + strain_id, + for_update=for_update, + ) @classmethod async def find_strain_by_code(cls, db: AsyncSession, strain_code: str) -> BizStrain | None: diff --git a/shierfeng-fastapi-backend/module_business/service/filling_service.py b/shierfeng-fastapi-backend/module_business/service/filling_service.py index c8e4515..4326179 100644 --- a/shierfeng-fastapi-backend/module_business/service/filling_service.py +++ b/shierfeng-fastapi-backend/module_business/service/filling_service.py @@ -96,6 +96,20 @@ class FillingService: raise ServiceException(message=admin_message) raise PermissionException(data='', message='无权访问该托盘') + @classmethod + async def _assert_pallet_warehouse_access( + cls, + query_db: AsyncSession, + current_user: CurrentUserModel, + warehouse_id: int, + ) -> None: + try: + await AccessService.assert_warehouse_access(query_db, current_user, warehouse_id) + except PermissionException: + if cls._is_admin(current_user): + raise + cls._raise_pallet_not_accessible(current_user) + @staticmethod def _validate_empty_warehouse_pallet( pallet: BizPallet, @@ -232,7 +246,7 @@ class FillingService: cls._raise_pallet_not_accessible(current_user, '托盘库存不存在') if inventory.warehouse_id is None: cls._raise_pallet_not_accessible(current_user, '托盘不在仓库') - await AccessService.assert_warehouse_access( + await cls._assert_pallet_warehouse_access( query_db, current_user, inventory.warehouse_id, @@ -279,8 +293,6 @@ class FillingService: warehouse_ids=warehouse_ids, ) if recovered is None: - if not cls._is_admin(current_user): - raise PermissionException(data='', message='无权访问该装填记录') raise ServiceException(message='装填失败，数据冲突') await AccessService.assert_warehouse_access( query_db, @@ -330,11 +342,6 @@ class FillingService: current_user, page_object.warehouse_id, ) - warehouse = await MasterDao.get_warehouse(query_db, page_object.warehouse_id) - if warehouse is None: - raise ServiceException(message='仓库不存在') - if warehouse.status != '0': - raise ServiceException(message='仓库已停用') scanned_pallet = await InventoryDao.get_pallet_by_qr_token( query_db, @@ -351,7 +358,7 @@ class FillingService: cls._raise_pallet_not_accessible(current_user, '托盘库存不存在') if inventory.warehouse_id is None: cls._raise_pallet_not_accessible(current_user, '托盘不在仓库') - await AccessService.assert_warehouse_access( + await cls._assert_pallet_warehouse_access( query_db, current_user, inventory.warehouse_id, @@ -380,6 +387,23 @@ class FillingService: current_user, ) + warehouse = await MasterDao.get_warehouse( + query_db, + page_object.warehouse_id, + for_update=True, + ) + if warehouse is None: + raise ServiceException(message='仓库不存在') + if warehouse.status != '0': + raise ServiceException(message='仓库已停用') + strain = cls._validate_active_strain( + await MasterDao.get_strain( + query_db, + page_object.strain_id, + for_update=True, + ) + ) + cls._validate_empty_warehouse_pallet( pallet, inventory, @@ -391,7 +415,6 @@ class FillingService: PalletFlowState(inventory.flow_state), PalletFlowState.FILLED_IN_WAREHOUSE, ) - strain = cls._validate_active_strain(await MasterDao.get_strain(query_db, page_object.strain_id)) current_time = datetime.now() filling = BizFillingRecord( diff --git a/shierfeng-fastapi-backend/tests/module_business/test_filling_service.py b/shierfeng-fastapi-backend/tests/module_business/test_filling_service.py index ef4a0a0..0d07755 100644 --- a/shierfeng-fastapi-backend/tests/module_business/test_filling_service.py +++ b/shierfeng-fastapi-backend/tests/module_business/test_filling_service.py @@ -332,10 +332,12 @@ class TestFillingReadServices(unittest.IsolatedAsyncioTestCase): new=AsyncMock(return_value=make_inventory(flow_state=PalletFlowState.FILLED_IN_WAREHOUSE.value)), ), patch.object(AccessService, 'assert_warehouse_access', new=access), - self.assertRaises(PermissionException), + self.assertRaises(PermissionException) as raised, ): await FillingService.scan_filling_pallet_services(self.db, 'qr-2', self.user) access.assert_awaited_once_with(self.db, self.user, 7) + self.assertEqual('', raised.exception.data) + self.assertEqual('无权访问该托盘', raised.exception.message) async def test_scan_hides_nonexistent_pallet_from_normal_user_but_admin_gets_precise_error(self) -> None: for user, exception_type in ((self.user, PermissionException), (make_user(1), ServiceException)): @@ -381,7 +383,11 @@ class TestFillPalletService(unittest.IsolatedAsyncioTestCase): actions.append('pallet-lock') return [self.pallet] - async def get_strain(*_args: Any) -> BizStrain: + async def get_warehouse(*_args: Any, **_kwargs: Any) -> SimpleNamespace: + actions.append('warehouse-read') + return self.warehouse + + async def get_strain(*_args: Any, **_kwargs: Any) -> BizStrain: actions.append('strain-read') return self.strain @@ -408,6 +414,8 @@ class TestFillPalletService(unittest.IsolatedAsyncioTestCase): captured['filling'].__dict__.pop('filling_id', None) self.db.commit.side_effect = commit + warehouse_read = AsyncMock(side_effect=get_warehouse) + strain_read = AsyncMock(side_effect=get_strain) with ( patch.object(AccessDao, 'list_user_warehouse_ids', new=AsyncMock(return_value=[7])), patch.object(AccessService, 'assert_warehouse_access', new=AsyncMock()), @@ -415,8 +423,8 @@ class TestFillPalletService(unittest.IsolatedAsyncioTestCase): patch.object(InventoryDao, 'get_pallet_by_qr_token', new=AsyncMock(return_value=self.pallet)), patch.object(InventoryDao, 'get_inventory_for_update', new=AsyncMock(side_effect=lock_inventory)), patch.object(InventoryDao, 'get_pallets_by_ids', new=AsyncMock(side_effect=lock_pallets)), - patch.object(MasterDao, 'get_warehouse', new=AsyncMock(return_value=self.warehouse)), - patch.object(MasterDao, 'get_strain', new=AsyncMock(side_effect=get_strain)), + patch.object(MasterDao, 'get_warehouse', new=warehouse_read), + patch.object(MasterDao, 'get_strain', new=strain_read), patch.object(OperationDao, 'add_filling_record', new=AsyncMock(side_effect=add_filling)), patch.object(InventoryDao, 'add_event', new=AsyncMock(side_effect=add_event)), patch('module_business.service.filling_service.datetime') as mocked_datetime, @@ -425,9 +433,19 @@ class TestFillPalletService(unittest.IsolatedAsyncioTestCase): result = await FillingService.fill_pallet_services(self.db, self.page, self.user) self.assertEqual( - ['inventory-lock', 'pallet-lock', 'strain-read', 'filling-flush', 'event-flush', 'commit'], + [ + 'inventory-lock', + 'pallet-lock', + 'warehouse-read', + 'strain-read', + 'filling-flush', + 'event-flush', + 'commit', + ], actions, ) + warehouse_read.assert_awaited_once_with(self.db, 7, for_update=True) + strain_read.assert_awaited_once_with(self.db, 11, for_update=True) filling = captured['filling'] self.assertTrue(filling.filling_no.startswith('FL-')) self.assertEqual(2, filling.pallet_id) @@ -514,6 +532,49 @@ class TestFillPalletService(unittest.IsolatedAsyncioTestCase): async def test_fill_rejects_packages_over_persisted_pallet_capacity(self) -> None: await self._assert_fill_rejected(page=make_page(initial_packages=21)) + async def test_fill_rejects_stopped_warehouse_only_after_core_locks(self) -> None: + actions: list[str] = [] + + async def lock_inventory(*_args: Any) -> BizPalletInventory: + actions.append('inventory-lock') + return self.inventory + + async def lock_pallets(*_args: Any) -> list[BizPallet]: + actions.append('pallet-lock') + return [self.pallet] + + async def lock_warehouse(*_args: Any, **_kwargs: Any) -> SimpleNamespace: + actions.append('warehouse-read') + return SimpleNamespace(warehouse_id=7, status='1') + + warehouse_read = AsyncMock(side_effect=lock_warehouse) + strain_read = AsyncMock() + add_filling = AsyncMock() + add_event = AsyncMock() + with ( + patch.object(AccessDao, 'list_user_warehouse_ids', new=AsyncMock(return_value=[7])), + patch.object(AccessService, 'assert_warehouse_access', new=AsyncMock()), + patch.object(OperationDao, 'get_filling_by_idempotency_key', new=AsyncMock(return_value=None)), + patch.object(InventoryDao, 'get_pallet_by_qr_token', new=AsyncMock(return_value=self.pallet)), + patch.object(InventoryDao, 'get_inventory_for_update', new=AsyncMock(side_effect=lock_inventory)), + patch.object(InventoryDao, 'get_pallets_by_ids', new=AsyncMock(side_effect=lock_pallets)), + patch.object(MasterDao, 'get_warehouse', new=warehouse_read), + patch.object(MasterDao, 'get_strain', new=strain_read), + patch.object(OperationDao, 'add_filling_record', new=add_filling), + patch.object(InventoryDao, 'add_event', new=add_event), + self.assertRaises(ServiceException) as raised, + ): + await FillingService.fill_pallet_services(self.db, self.page, self.user) + + self.assertEqual('仓库已停用', raised.exception.message) + self.assertEqual(['inventory-lock', 'pallet-lock', 'warehouse-read'], actions) + warehouse_read.assert_awaited_once_with(self.db, 7, for_update=True) + strain_read.assert_not_awaited() + add_filling.assert_not_awaited() + add_event.assert_not_awaited() + self.db.commit.assert_not_awaited() + self.db.rollback.assert_awaited_once() + async def test_fill_rejects_stopped_pallet_and_stopped_or_missing_strain(self) -> None: await self._assert_fill_rejected(pallet=make_pallet(asset_status='1')) self.db.reset_mock() @@ -556,6 +617,8 @@ class TestFillPalletService(unittest.IsolatedAsyncioTestCase): existing = make_filling() inventory_lock = AsyncMock() access = AsyncMock() + warehouse_read = AsyncMock() + strain_read = AsyncMock() async def rollback() -> None: existing.__dict__.pop('filling_id', None) @@ -568,12 +631,16 @@ class TestFillPalletService(unittest.IsolatedAsyncioTestCase): OperationDao, 'get_filling_by_idempotency_key', new=AsyncMock(return_value=existing) ) as get_key, patch.object(InventoryDao, 'get_inventory_for_update', new=inventory_lock), + patch.object(MasterDao, 'get_warehouse', new=warehouse_read), + patch.object(MasterDao, 'get_strain', new=strain_read), ): result = await FillingService.fill_pallet_services(self.db, self.page, self.user) get_key.assert_awaited_once_with(self.db, 'fill-key', warehouse_ids=[7]) access.assert_awaited_once_with(self.db, self.user, 7) inventory_lock.assert_not_awaited() + warehouse_read.assert_not_awaited() + strain_read.assert_not_awaited() self.db.commit.assert_not_awaited() self.db.rollback.assert_awaited_once() self.assertTrue(result.is_success) @@ -615,6 +682,9 @@ class TestFillPalletService(unittest.IsolatedAsyncioTestCase): raise PermissionException(data='', message='无权访问该仓库') access = AsyncMock(side_effect=assert_access) + strain_read = AsyncMock() + add_filling = AsyncMock() + add_event = AsyncMock() with ( patch.object(AccessDao, 'list_user_warehouse_ids', new=AsyncMock(return_value=[7])), patch.object(AccessService, 'assert_warehouse_access', new=access), @@ -623,12 +693,20 @@ class TestFillPalletService(unittest.IsolatedAsyncioTestCase): patch.object(InventoryDao, 'get_inventory_for_update', new=AsyncMock(return_value=inventory)), patch.object(InventoryDao, 'get_pallets_by_ids', new=pallet_lock), patch.object(MasterDao, 'get_warehouse', new=AsyncMock(return_value=self.warehouse)), - self.assertRaises(PermissionException), + patch.object(MasterDao, 'get_strain', new=strain_read), + patch.object(OperationDao, 'add_filling_record', new=add_filling), + patch.object(InventoryDao, 'add_event', new=add_event), + self.assertRaises(PermissionException) as raised, ): await FillingService.fill_pallet_services(self.db, self.page, self.user) self.assertEqual([7, foreign_warehouse_id], [call.args[2] for call in access.await_args_list]) + self.assertEqual('', raised.exception.data) + self.assertEqual('无权访问该托盘', raised.exception.message) pallet_lock.assert_not_awaited() + strain_read.assert_not_awaited() + add_filling.assert_not_awaited() + add_event.assert_not_awaited() self.db.rollback.assert_awaited_once() async def test_concurrent_same_key_recovers_after_waiting_for_inventory_lock(self) -> None: @@ -655,6 +733,8 @@ class TestFillPalletService(unittest.IsolatedAsyncioTestCase): add_filling = AsyncMock() add_event = AsyncMock() + warehouse_read = AsyncMock(return_value=self.warehouse) + strain_read = AsyncMock() with ( patch.object(AccessDao, 'list_user_warehouse_ids', new=AsyncMock(return_value=[7])), patch.object(AccessService, 'assert_warehouse_access', new=AsyncMock()), @@ -667,7 +747,8 @@ class TestFillPalletService(unittest.IsolatedAsyncioTestCase): patch.object(InventoryDao, 'get_inventory_for_update', new=AsyncMock(side_effect=lock_inventory)), patch.object(InventoryDao, 'get_pallets_by_ids', new=AsyncMock(side_effect=lock_pallet)), patch.object(OperationDao, 'get_active_filling_for_update', new=AsyncMock(side_effect=lock_active)), - patch.object(MasterDao, 'get_warehouse', new=AsyncMock(return_value=self.warehouse)), + patch.object(MasterDao, 'get_warehouse', new=warehouse_read), + patch.object(MasterDao, 'get_strain', new=strain_read), patch.object(OperationDao, 'add_filling_record', new=add_filling), patch.object(InventoryDao, 'add_event', new=add_event), ): @@ -677,6 +758,8 @@ class TestFillPalletService(unittest.IsolatedAsyncioTestCase): self.assertEqual(2, get_key.await_count) self.assertEqual(2, self.db.rollback.await_count) self.assertEqual(31, result.result) + warehouse_read.assert_not_awaited() + strain_read.assert_not_awaited() add_filling.assert_not_awaited() add_event.assert_not_awaited() self.db.commit.assert_not_awaited() @@ -762,7 +845,30 @@ class TestFillPalletService(unittest.IsolatedAsyncioTestCase): access.assert_any_await(self.db, self.user, 7) self.assertTrue(result.is_success) - async def test_integrity_error_without_scoped_idempotent_result_is_hidden_and_never_succeeds(self) -> None: + async def test_integrity_error_recovery_rejects_revoked_warehouse_access(self) -> None: + warehouse_scope = AsyncMock(side_effect=[[7], []]) + key_reads = AsyncMock(return_value=None) + with ( + patch.object(AccessDao, 'list_user_warehouse_ids', new=warehouse_scope), + patch.object(AccessService, 'assert_warehouse_access', new=AsyncMock()), + patch.object(OperationDao, 'get_filling_by_idempotency_key', new=key_reads), + patch.object(InventoryDao, 'get_pallet_by_qr_token', new=AsyncMock(return_value=self.pallet)), + patch.object(InventoryDao, 'get_inventory_for_update', new=AsyncMock(return_value=self.inventory)), + patch.object(InventoryDao, 'get_pallets_by_ids', new=AsyncMock(return_value=[self.pallet])), + patch.object(MasterDao, 'get_warehouse', new=AsyncMock(return_value=self.warehouse)), + patch.object(MasterDao, 'get_strain', new=AsyncMock(return_value=self.strain)), + patch.object(OperationDao, 'add_filling_record', new=AsyncMock(side_effect=integrity_error())), + self.assertRaises(PermissionException) as raised, + ): + await FillingService.fill_pallet_services(self.db, self.page, self.user) + + self.assertEqual('无权访问该仓库', raised.exception.message) + self.assertEqual(2, warehouse_scope.await_count) + self.assertEqual(1, key_reads.await_count) + self.assertEqual(2, self.db.rollback.await_count) + self.db.commit.assert_not_awaited() + + async def test_integrity_error_without_idempotent_result_is_reported_as_data_conflict(self) -> None: with ( patch.object(AccessDao, 'list_user_warehouse_ids', new=AsyncMock(return_value=[7])), patch.object(AccessService, 'assert_warehouse_access', new=AsyncMock()), @@ -777,9 +883,10 @@ class TestFillPalletService(unittest.IsolatedAsyncioTestCase): patch.object(MasterDao, 'get_warehouse', new=AsyncMock(return_value=self.warehouse)), patch.object(MasterDao, 'get_strain', new=AsyncMock(return_value=self.strain)), patch.object(OperationDao, 'add_filling_record', new=AsyncMock(side_effect=integrity_error())), - self.assertRaises(PermissionException), + self.assertRaises(ServiceException) as raised, ): await FillingService.fill_pallet_services(self.db, self.page, self.user) + self.assertEqual('装填失败，数据冲突', raised.exception.message) self.assertEqual(2, self.db.rollback.await_count) self.db.commit.assert_not_awaited() diff --git a/shierfeng-fastapi-backend/tests/module_business/test_master_service.py b/shierfeng-fastapi-backend/tests/module_business/test_master_service.py index 781cfce..a927755 100644 --- a/shierfeng-fastapi-backend/tests/module_business/test_master_service.py +++ b/shierfeng-fastapi-backend/tests/module_business/test_master_service.py @@ -502,6 +502,24 @@ class TestMasterDao(unittest.IsolatedAsyncioTestCase): self.assertIn(f'FROM {table}', sql) self.assertIn(f'{table}.{id_field} = 9', sql) + async def test_locking_master_gets_use_current_reads_and_refresh_identity_map(self) -> None: + strain_lock_parameter = inspect.signature(MasterDao.get_strain).parameters.get('for_update') + self.assertIsNotNone(strain_lock_parameter) + self.assertIs(inspect.Parameter.KEYWORD_ONLY, strain_lock_parameter.kind) + self.assertFalse(strain_lock_parameter.default) + + for resource in ('warehouse', 'strain'): + with self.subTest(resource=resource): + result = MagicMock() + result.scalar_one_or_none.return_value = None + self.db.execute = AsyncMock(return_value=result) + + await getattr(MasterDao, f'get_{resource}')(self.db, 9, for_update=True) + + statement = self.db.execute.await_args.args[0] + self.assertIn('FOR UPDATE', compile_sql(statement)) + self.assertIs(True, statement.get_execution_options().get('populate_existing')) + async def test_unique_queries_use_global_or_parent_scoped_predicates(self) -> None: cases = ( ('strain', ('S001',), 'biz_strain', ('strain_code',)),