From 66890ca63c537c6be83936574b55db2c6c3d8e21 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 18:16:34 +0800 Subject: [PATCH] =?UTF-8?q?=E4=BF=AE=E5=A4=8D=EF=BC=9A=E8=A3=85=E5=A1=AB?= =?UTF-8?q?=E5=89=8D=E5=A4=8D=E6=A0=B8=E4=BB=93=E5=BA=93=E6=9D=83=E9=99=90?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../module_business/dao/access_dao.py | 24 +++-- .../module_business/service/access_service.py | 13 ++- .../service/filling_service.py | 20 ++++- .../module_business/test_access_service.py | 52 +++++++++++ .../module_business/test_filling_service.py | 89 ++++++++++++++++++- 5 files changed, 184 insertions(+), 14 deletions(-) diff --git a/shierfeng-fastapi-backend/module_business/dao/access_dao.py b/shierfeng-fastapi-backend/module_business/dao/access_dao.py index 5935034..0745432 100644 --- a/shierfeng-fastapi-backend/module_business/dao/access_dao.py +++ b/shierfeng-fastapi-backend/module_business/dao/access_dao.py @@ -8,16 +8,22 @@ class AccessDao: """仓库与客户现场数据权限数据库操作层。""" @classmethod - async def user_has_warehouse(cls, db: AsyncSession, user_id: int, warehouse_id: int) -> bool: + async def user_has_warehouse( + cls, + db: AsyncSession, + user_id: int, + warehouse_id: int, + *, + for_update: bool = False, + ) -> bool: """判断用户是否关联指定仓库。""" - mapped_user_id = ( - await db.execute( - select(BizUserWarehouse.user_id).where( - BizUserWarehouse.user_id == user_id, - BizUserWarehouse.warehouse_id == warehouse_id, - ) - ) - ).scalar_one_or_none() + statement = select(BizUserWarehouse.user_id).where( + BizUserWarehouse.user_id == user_id, + BizUserWarehouse.warehouse_id == warehouse_id, + ) + if for_update: + statement = statement.with_for_update() + mapped_user_id = (await db.execute(statement)).scalar_one_or_none() return mapped_user_id is not None diff --git a/shierfeng-fastapi-backend/module_business/service/access_service.py b/shierfeng-fastapi-backend/module_business/service/access_service.py index 5f73764..fe9b0b8 100644 --- a/shierfeng-fastapi-backend/module_business/service/access_service.py +++ b/shierfeng-fastapi-backend/module_business/service/access_service.py @@ -14,11 +14,22 @@ class AccessService: db: AsyncSession, current_user: CurrentUserModel, warehouse_id: int, + *, + for_update: bool = False, ) -> None: """校验当前用户是否有权访问指定仓库。""" if current_user.user.admin: return - if not await AccessDao.user_has_warehouse(db, current_user.user.user_id, warehouse_id): + if for_update: + has_warehouse = await AccessDao.user_has_warehouse( + db, + current_user.user.user_id, + warehouse_id, + for_update=True, + ) + else: + has_warehouse = await AccessDao.user_has_warehouse(db, current_user.user.user_id, warehouse_id) + if not has_warehouse: raise PermissionException(data='', message='无权访问该仓库') @classmethod diff --git a/shierfeng-fastapi-backend/module_business/service/filling_service.py b/shierfeng-fastapi-backend/module_business/service/filling_service.py index 4326179..aa35539 100644 --- a/shierfeng-fastapi-backend/module_business/service/filling_service.py +++ b/shierfeng-fastapi-backend/module_business/service/filling_service.py @@ -102,9 +102,19 @@ class FillingService: query_db: AsyncSession, current_user: CurrentUserModel, warehouse_id: int, + *, + for_update: bool = False, ) -> None: try: - await AccessService.assert_warehouse_access(query_db, current_user, warehouse_id) + if for_update: + await AccessService.assert_warehouse_access( + query_db, + current_user, + warehouse_id, + for_update=True, + ) + else: + await AccessService.assert_warehouse_access(query_db, current_user, warehouse_id) except PermissionException: if cls._is_admin(current_user): raise @@ -389,13 +399,19 @@ class FillingService: warehouse = await MasterDao.get_warehouse( query_db, - page_object.warehouse_id, + inventory.warehouse_id, for_update=True, ) if warehouse is None: raise ServiceException(message='仓库不存在') if warehouse.status != '0': raise ServiceException(message='仓库已停用') + await cls._assert_pallet_warehouse_access( + query_db, + current_user, + inventory.warehouse_id, + for_update=True, + ) strain = cls._validate_active_strain( await MasterDao.get_strain( query_db, diff --git a/shierfeng-fastapi-backend/tests/module_business/test_access_service.py b/shierfeng-fastapi-backend/tests/module_business/test_access_service.py index 57ce566..a354af0 100644 --- a/shierfeng-fastapi-backend/tests/module_business/test_access_service.py +++ b/shierfeng-fastapi-backend/tests/module_business/test_access_service.py @@ -1,3 +1,4 @@ +import inspect import unittest from types import SimpleNamespace from unittest.mock import AsyncMock, MagicMock, patch @@ -36,6 +37,36 @@ class AccessServiceTest(unittest.IsolatedAsyncioTestCase): user_has_warehouse.assert_awaited_once_with(db, 8, 99) + @patch('module_business.service.access_service.AccessDao.user_has_warehouse', new_callable=AsyncMock) + async def test_locking_warehouse_access_is_forwarded(self, user_has_warehouse: AsyncMock) -> None: + lock_parameter = inspect.signature(AccessService.assert_warehouse_access).parameters.get('for_update') + self.assertIsNotNone(lock_parameter) + self.assertIs(inspect.Parameter.KEYWORD_ONLY, lock_parameter.kind) + self.assertIs(False, lock_parameter.default) + + user_has_warehouse.return_value = True + user = SimpleNamespace(user=SimpleNamespace(user_id=8, admin=False)) + db = AsyncMock(spec=AsyncSession) + + await AccessService.assert_warehouse_access(db, user, 99, for_update=True) + + user_has_warehouse.assert_awaited_once_with(db, 8, 99, for_update=True) + + @patch('module_business.service.access_service.AccessDao.user_has_warehouse', new_callable=AsyncMock) + async def test_admin_bypasses_locking_warehouse_mapping(self, user_has_warehouse: AsyncMock) -> None: + lock_parameter = inspect.signature(AccessService.assert_warehouse_access).parameters.get('for_update') + self.assertIsNotNone(lock_parameter) + user = SimpleNamespace(user=SimpleNamespace(user_id=1, admin=True)) + + await AccessService.assert_warehouse_access( + AsyncMock(spec=AsyncSession), + user, + 99, + for_update=True, + ) + + user_has_warehouse.assert_not_awaited() + @patch('module_business.service.access_service.AccessDao.user_has_warehouse', new_callable=AsyncMock) async def test_unmapped_warehouse_is_forbidden(self, user_has_warehouse: AsyncMock) -> None: user_has_warehouse.return_value = False @@ -127,6 +158,27 @@ class AccessDaoTest(unittest.IsolatedAsyncioTestCase): ), ) + async def test_user_has_warehouse_current_read_is_optional_keyword_only_lock(self) -> None: + lock_parameter = inspect.signature(AccessDao.user_has_warehouse).parameters.get('for_update') + self.assertIsNotNone(lock_parameter) + self.assertIs(inspect.Parameter.KEYWORD_ONLY, lock_parameter.kind) + self.assertIs(False, lock_parameter.default) + + for for_update in (False, True): + with self.subTest(for_update=for_update): + result = MagicMock() + result.scalar_one_or_none.return_value = 8 + db = self._db_with_result(result) + + if for_update: + await AccessDao.user_has_warehouse(db, 8, 99, for_update=True) + else: + await AccessDao.user_has_warehouse(db, 8, 99) + + statement = db.execute.await_args.args[0] + sql = ' '.join(str(statement.compile(compile_kwargs={'literal_binds': True})).split()) + self.assertIs(for_update, 'FOR UPDATE' in sql) + async def test_user_has_site_returns_mapping_existence(self) -> None: for scalar, expected in ((8, True), (None, False)): with self.subTest(scalar=scalar): 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 0d07755..dd2de6e 100644 --- a/shierfeng-fastapi-backend/tests/module_business/test_filling_service.py +++ b/shierfeng-fastapi-backend/tests/module_business/test_filling_service.py @@ -391,6 +391,16 @@ class TestFillPalletService(unittest.IsolatedAsyncioTestCase): actions.append('strain-read') return self.strain + async def assert_access( + _db: AsyncSession, + _user: CurrentUserModel, + _warehouse_id: int, + *, + for_update: bool = False, + ) -> None: + if for_update: + actions.append('warehouse-access-read') + async def add_filling(_db: AsyncSession, filling: BizFillingRecord) -> BizFillingRecord: actions.append('filling-flush') filling.filling_id = 31 @@ -414,11 +424,12 @@ class TestFillPalletService(unittest.IsolatedAsyncioTestCase): captured['filling'].__dict__.pop('filling_id', None) self.db.commit.side_effect = commit + access = AsyncMock(side_effect=assert_access) 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()), + patch.object(AccessService, 'assert_warehouse_access', new=access), 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)), @@ -437,6 +448,7 @@ class TestFillPalletService(unittest.IsolatedAsyncioTestCase): 'inventory-lock', 'pallet-lock', 'warehouse-read', + 'warehouse-access-read', 'strain-read', 'filling-flush', 'event-flush', @@ -444,6 +456,10 @@ class TestFillPalletService(unittest.IsolatedAsyncioTestCase): ], actions, ) + self.assertEqual( + [False, False, True], [call.kwargs.get('for_update', False) for call in access.await_args_list] + ) + self.assertEqual([7, 7, 7], [call.args[2] for call in access.await_args_list]) 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'] @@ -575,6 +591,73 @@ class TestFillPalletService(unittest.IsolatedAsyncioTestCase): self.db.commit.assert_not_awaited() self.db.rollback.assert_awaited_once() + async def test_fill_rechecks_actual_warehouse_mapping_after_warehouse_lock(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 self.warehouse + + async def assert_access( + _db: AsyncSession, + _user: CurrentUserModel, + _warehouse_id: int, + *, + for_update: bool = False, + ) -> None: + if for_update: + actions.append('warehouse-access-read') + raise PermissionException(data='', message='无权访问该仓库') + + async def lock_strain(*_args: Any, **_kwargs: Any) -> BizStrain: + actions.append('strain-read') + return self.strain + + access = AsyncMock(side_effect=assert_access) + warehouse_read = AsyncMock(side_effect=lock_warehouse) + strain_read = AsyncMock(side_effect=lock_strain) + add_filling = AsyncMock(return_value=make_filling()) + add_event = AsyncMock() + with ( + patch.object(AccessDao, 'list_user_warehouse_ids', new=AsyncMock(return_value=[7])), + patch.object(AccessService, 'assert_warehouse_access', new=access), + 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(PermissionException) as raised, + ): + await FillingService.fill_pallet_services(self.db, self.page, self.user) + + self.assertEqual('', raised.exception.data) + self.assertEqual('无权访问该托盘', raised.exception.message) + self.assertEqual( + ['inventory-lock', 'pallet-lock', 'warehouse-read', 'warehouse-access-read'], + actions, + ) + self.assertEqual( + [False, False, True], [call.kwargs.get('for_update', False) for call in access.await_args_list] + ) + self.assertEqual([7, 7, 7], [call.args[2] for call in access.await_args_list]) + 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() @@ -663,9 +746,11 @@ class TestFillPalletService(unittest.IsolatedAsyncioTestCase): with ( patch.object(AccessDao, 'list_user_warehouse_ids', new=AsyncMock(return_value=[])), patch.object(OperationDao, 'get_filling_by_idempotency_key', new=get_key), - self.assertRaises(PermissionException), + self.assertRaises(PermissionException) as raised, ): await FillingService.fill_pallet_services(self.db, self.page, self.user) + self.assertEqual('', raised.exception.data) + self.assertEqual('无权访问该仓库', raised.exception.message) get_key.assert_not_awaited() self.db.rollback.assert_awaited_once()