diff --git a/shierfeng-fastapi-backend/common/annotation/log_annotation.py b/shierfeng-fastapi-backend/common/annotation/log_annotation.py index 4b11b59..19d95a0 100644 --- a/shierfeng-fastapi-backend/common/annotation/log_annotation.py +++ b/shierfeng-fastapi-backend/common/annotation/log_annotation.py @@ -19,7 +19,7 @@ from user_agents import parse from common.context import RequestContext from common.enums import BusinessType from config.env import AppConfig -from exceptions.exception import LoginException, ServiceException, ServiceWarning +from exceptions.exception import LoginException, PermissionException, ServiceException, ServiceWarning from module_admin.entity.vo.log_vo import LogininforModel, OperLogModel from module_admin.service.log_service import LogQueueService from utils.client_ip_util import ClientIPUtil @@ -193,6 +193,8 @@ class Log: except (LoginException, ServiceWarning) as e: logger.warning(e.message) result = ResponseUtil.failure(data=e.data, msg=e.message) + except PermissionException as e: + result = ResponseUtil.forbidden(data=e.data, msg=e.message) except ServiceException as e: logger.error(e.message) result = ResponseUtil.error(data=e.data, msg=e.message) diff --git a/shierfeng-fastapi-backend/module_business/dao/operation_dao.py b/shierfeng-fastapi-backend/module_business/dao/operation_dao.py index a844f6e..a6be1b1 100644 --- a/shierfeng-fastapi-backend/module_business/dao/operation_dao.py +++ b/shierfeng-fastapi-backend/module_business/dao/operation_dao.py @@ -51,6 +51,15 @@ class OperationDao: conditions.append(column < end_at) return conditions + @staticmethod + def _inbound_scope_conditions( + warehouse_ids: list[int] | None, + ) -> list[ColumnElement[bool]]: + if warehouse_ids is None: + return [] + scoped_warehouse_ids = sorted(set(warehouse_ids)) + return [BizInboundOrder.warehouse_id.in_(scoped_warehouse_ids) if scoped_warehouse_ids else false()] + @staticmethod async def _paginate( db: AsyncSession, @@ -98,8 +107,9 @@ class OperationDao: model: type[Any], id_column: InstrumentedAttribute, record_id: int, + *extra_conditions: ColumnElement[bool], ) -> Any | None: - result = await db.execute(select(model).where(id_column == record_id)) + result = await db.execute(select(model).where(id_column == record_id, *extra_conditions)) return result.scalar_one_or_none() @classmethod @@ -112,8 +122,15 @@ class OperationDao: item_model: type[Any], item_order_id_column: InstrumentedAttribute, item_id_column: InstrumentedAttribute, + *order_conditions: ColumnElement[bool], ) -> Any | None: - order = await cls._get_by_id(db, order_model, order_id_column, order_id) + order = await cls._get_by_id( + db, + order_model, + order_id_column, + order_id, + *order_conditions, + ) if order is None: return None @@ -185,12 +202,14 @@ class OperationDao: cls, db: AsyncSession, inbound_id: int, + warehouse_ids: list[int] | None = None, ) -> BizInboundOrder | None: inbound = await cls._get_for_update( db, BizInboundOrder, BizInboundOrder.inbound_id, inbound_id, + *cls._inbound_scope_conditions(warehouse_ids), ) if inbound is None: return None @@ -336,12 +355,7 @@ class OperationDao: is_page: bool = True, warehouse_ids: list[int] | None = None, ) -> PageModel | list[dict[str, Any]]: - conditions = [] - if warehouse_ids is not None: - scoped_warehouse_ids = sorted(set(warehouse_ids)) - conditions.append( - BizInboundOrder.warehouse_id.in_(scoped_warehouse_ids) if scoped_warehouse_ids else false() - ) + conditions = cls._inbound_scope_conditions(warehouse_ids) if query_object.inbound_no: conditions.append(BizInboundOrder.inbound_no.like(f'%{query_object.inbound_no}%')) if query_object.warehouse_id: @@ -363,6 +377,7 @@ class OperationDao: cls, db: AsyncSession, inbound_id: int, + warehouse_ids: list[int] | None = None, ) -> BizInboundOrder | None: return await cls._get_order_detail( db, @@ -372,6 +387,7 @@ class OperationDao: BizInboundItem, BizInboundItem.inbound_id, BizInboundItem.inbound_item_id, + *cls._inbound_scope_conditions(warehouse_ids), ) @classmethod diff --git a/shierfeng-fastapi-backend/module_business/service/inbound_service.py b/shierfeng-fastapi-backend/module_business/service/inbound_service.py index 0435bd0..31a9a5e 100644 --- a/shierfeng-fastapi-backend/module_business/service/inbound_service.py +++ b/shierfeng-fastapi-backend/module_business/service/inbound_service.py @@ -2,7 +2,7 @@ import hashlib import json import secrets from datetime import datetime -from typing import Any +from typing import Any, NoReturn from sqlalchemy.exc import IntegrityError from sqlalchemy.ext.asyncio import AsyncSession @@ -81,18 +81,27 @@ class InboundService: user_id, _ = cls._current_identity(current_user) return sorted(set(await AccessDao.list_user_warehouse_ids(query_db, user_id))) + @classmethod + def _raise_inbound_not_accessible( + cls, + current_user: CurrentUserModel, + ) -> NoReturn: + if cls._is_admin(current_user): + raise ServiceException(message='入库单不存在') + raise PermissionException(data='', message='无权访问该入库单') + @staticmethod async def _validate_active_warehouse( query_db: AsyncSession, warehouse_id: int, current_user: CurrentUserModel, ) -> Any: + await AccessService.assert_warehouse_access(query_db, current_user, warehouse_id) warehouse = await MasterDao.get_warehouse(query_db, warehouse_id) if warehouse is None: raise ServiceException(message='目标仓库不存在') if warehouse.status != '0': raise ServiceException(message='目标仓库已停用') - await AccessService.assert_warehouse_access(query_db, current_user, warehouse_id) return warehouse @staticmethod @@ -188,10 +197,14 @@ class InboundService: inbound_id: int, current_user: CurrentUserModel, ) -> InboundModel: - inbound = await OperationDao.get_inbound_by_id(query_db, inbound_id) + warehouse_ids = await cls._warehouse_scope(query_db, current_user) + inbound = await OperationDao.get_inbound_by_id( + query_db, + inbound_id, + warehouse_ids=warehouse_ids, + ) if inbound is None: - raise ServiceException(message='入库单不存在') - await AccessService.assert_warehouse_access(query_db, current_user, inbound.warehouse_id) + cls._raise_inbound_not_accessible(current_user) return InboundModel.model_validate(inbound) @classmethod @@ -318,14 +331,14 @@ class InboundService: ) -> CrudResponseModel: try: _, user_name = cls._current_identity(current_user) - inbound = await OperationDao.get_inbound_for_update(query_db, page_object.inbound_id) - if inbound is None: - raise ServiceException(message='入库单不存在') - await AccessService.assert_warehouse_access( + warehouse_ids = await cls._warehouse_scope(query_db, current_user) + inbound = await OperationDao.get_inbound_for_update( query_db, - current_user, - inbound.warehouse_id, + page_object.inbound_id, + warehouse_ids=warehouse_ids, ) + if inbound is None: + cls._raise_inbound_not_accessible(current_user) if inbound.status != DocumentStatus.DRAFT.value: raise ServiceException(message='已确认入库单不允许编辑') await cls._validate_active_warehouse( @@ -369,15 +382,15 @@ class InboundService: raise ServiceException(message='入库单ID不能重复') ordered_ids = sorted(inbound_ids) + warehouse_ids = await cls._warehouse_scope(query_db, current_user) for inbound_id in ordered_ids: - inbound = await OperationDao.get_inbound_for_update(query_db, inbound_id) - if inbound is None: - raise ServiceException(message='入库单不存在') - await AccessService.assert_warehouse_access( + inbound = await OperationDao.get_inbound_for_update( query_db, - current_user, - inbound.warehouse_id, + inbound_id, + warehouse_ids=warehouse_ids, ) + if inbound is None: + cls._raise_inbound_not_accessible(current_user) if inbound.status != DocumentStatus.DRAFT.value: raise ServiceException(message='已确认入库单不允许删除') @@ -448,14 +461,14 @@ class InboundService: ) -> CrudResponseModel: try: user_id, user_name = cls._current_identity(current_user) - inbound = await OperationDao.get_inbound_by_id(query_db, inbound_id) - if inbound is None: - raise ServiceException(message='入库单不存在') - await AccessService.assert_warehouse_access( + warehouse_ids = await cls._warehouse_scope(query_db, current_user) + inbound = await OperationDao.get_inbound_by_id( query_db, - current_user, - inbound.warehouse_id, + inbound_id, + warehouse_ids=warehouse_ids, ) + if inbound is None: + cls._raise_inbound_not_accessible(current_user) request_hash = cls._build_confirm_request_hash(inbound_id) idempotency_key = page_object.idempotency_key @@ -477,14 +490,13 @@ class InboundService: ): raise ServiceException(message='确认入库失败，幂等键已被其他请求使用') - locked_inbound = await OperationDao.get_inbound_for_update(query_db, inbound_id) - if locked_inbound is None: - raise ServiceException(message='入库单不存在') - await AccessService.assert_warehouse_access( + locked_inbound = await OperationDao.get_inbound_for_update( query_db, - current_user, - locked_inbound.warehouse_id, + inbound_id, + warehouse_ids=warehouse_ids, ) + if locked_inbound is None: + cls._raise_inbound_not_accessible(current_user) if locked_inbound.status == DocumentStatus.CONFIRMED.value: if cls._is_exact_confirmed( locked_inbound, diff --git a/shierfeng-fastapi-backend/tests/module_business/test_inbound_service.py b/shierfeng-fastapi-backend/tests/module_business/test_inbound_service.py index c27f863..d64b1b9 100644 --- a/shierfeng-fastapi-backend/tests/module_business/test_inbound_service.py +++ b/shierfeng-fastapi-backend/tests/module_business/test_inbound_service.py @@ -7,13 +7,14 @@ from types import SimpleNamespace from typing import Any, get_type_hints from unittest.mock import AsyncMock, MagicMock, patch -from fastapi import FastAPI +from fastapi import FastAPI, Request from fastapi.routing import APIRoute from sqlalchemy.exc import IntegrityError from sqlalchemy.ext.asyncio import AsyncSession from common.aspect.interface_auth import CheckUserInterfaceAuth from common.aspect.pre_auth import PreAuth +from common.context import RequestContext from common.vo import ( CrudResponseModel, DataResponseModel, @@ -23,6 +24,7 @@ from common.vo import ( ) from exceptions.exception import PermissionException, ServiceException from module_admin.entity.vo.user_vo import CurrentUserModel, UserInfoModel +from module_admin.service.log_service import LogQueueService from module_admin.service.login_service import LoginService from module_business.controller import inbound_controller as inbound_controller_module from module_business.dao.access_dao import AccessDao @@ -161,6 +163,13 @@ class TestInboundDraftService(unittest.IsolatedAsyncioTestCase): self.db = MagicMock(spec=AsyncSession) self.user = make_user() self.fixed_time = datetime(2026, 7, 26, 10, 30) + self.scope_patcher = patch.object( + AccessDao, + 'list_user_warehouse_ids', + new=AsyncMock(return_value=[2, 5, 7, 8, 11]), + ) + self.default_scope = self.scope_patcher.start() + self.addCleanup(self.scope_patcher.stop) async def test_add_validates_and_locks_pallets_before_writing_draft(self) -> None: page = InboundCreateModel(warehouseId=7, palletIds=[5, 2]) @@ -320,14 +329,38 @@ class TestInboundDraftService(unittest.IsolatedAsyncioTestCase): lock.assert_not_awaited() self.db.rollback.assert_awaited_once() - async def test_edit_checks_original_and_target_access_then_replaces_items(self) -> None: + async def test_add_checks_warehouse_access_before_revealing_existence(self) -> None: + page = InboundCreateModel(warehouseId=404, palletIds=[2]) + get_warehouse = AsyncMock(return_value=None) + access = AsyncMock( + side_effect=PermissionException(data='', message='无权访问该仓库'), + ) + pallet_lock = AsyncMock() + + with ( + patch.object(MasterDao, 'get_warehouse', new=get_warehouse), + patch.object(AccessService, 'assert_warehouse_access', new=access), + patch.object(InventoryDao, 'get_inventories_for_update', new=pallet_lock), + self.assertRaises(Exception) as raised, + ): + await InboundService.add_inbound_services(self.db, page, self.user) + + self.assertIsInstance(raised.exception, PermissionException) + self.assertEqual('无权访问该仓库', raised.exception.message) + access.assert_awaited_once_with(self.db, self.user, 404) + get_warehouse.assert_not_awaited() + pallet_lock.assert_not_awaited() + self.db.rollback.assert_awaited_once() + + async def test_edit_scopes_original_and_checks_target_access_then_replaces_items(self) -> None: page = InboundUpdateModel(inboundId=11, warehouseId=8, palletIds=[5, 2]) inbound = make_inbound(warehouse_id=7, pallet_ids=(3,)) + lock_order = AsyncMock(return_value=inbound) access = AsyncMock() update = AsyncMock() replace = AsyncMock() with ( - patch.object(OperationDao, 'get_inbound_for_update', new=AsyncMock(return_value=inbound)), + patch.object(OperationDao, 'get_inbound_for_update', new=lock_order), patch.object(AccessService, 'assert_warehouse_access', new=access), patch.object( MasterDao, @@ -351,10 +384,13 @@ class TestInboundDraftService(unittest.IsolatedAsyncioTestCase): service_datetime.now.return_value = self.fixed_time result = await InboundService.edit_inbound_services(self.db, page, self.user) - self.assertEqual( - [unittest.mock.call(self.db, self.user, 7), unittest.mock.call(self.db, self.user, 8)], - access.await_args_list, + self.default_scope.assert_awaited_once_with(self.db, 7) + lock_order.assert_awaited_once_with( + self.db, + 11, + warehouse_ids=[2, 5, 7, 8, 11], ) + access.assert_awaited_once_with(self.db, self.user, 8) inventory_lock.assert_awaited_once_with(self.db, [2, 5]) pallet_lock.assert_awaited_once_with(self.db, [2, 5]) self.assertEqual(8, inbound.warehouse_id) @@ -388,16 +424,56 @@ class TestInboundDraftService(unittest.IsolatedAsyncioTestCase): update.assert_not_awaited() self.db.rollback.assert_awaited_once() + async def test_edit_scopes_parent_lock_and_hides_unmatched_document(self) -> None: + page = InboundUpdateModel(inboundId=404, warehouseId=8, palletIds=[2]) + lock_order = AsyncMock(return_value=None) + get_warehouse = AsyncMock() + inventory_lock = AsyncMock() + with ( + patch.object(AccessDao, 'list_user_warehouse_ids', new=AsyncMock(return_value=[7])) as scope, + patch.object(OperationDao, 'get_inbound_for_update', new=lock_order), + patch.object(MasterDao, 'get_warehouse', new=get_warehouse), + patch.object(InventoryDao, 'get_inventories_for_update', new=inventory_lock), + self.assertRaises(Exception) as raised, + ): + await InboundService.edit_inbound_services(self.db, page, self.user) + + self.assertIsInstance(raised.exception, PermissionException) + self.assertEqual('无权访问该入库单', raised.exception.message) + scope.assert_awaited_once_with(self.db, 7) + lock_order.assert_awaited_once_with(self.db, 404, warehouse_ids=[7]) + get_warehouse.assert_not_awaited() + inventory_lock.assert_not_awaited() + self.db.rollback.assert_awaited_once() + + async def test_admin_missing_edit_keeps_precise_not_found(self) -> None: + page = InboundUpdateModel(inboundId=404, warehouseId=8, palletIds=[2]) + scope = AsyncMock() + lock_order = AsyncMock(return_value=None) + with ( + patch.object(AccessDao, 'list_user_warehouse_ids', new=scope), + patch.object(OperationDao, 'get_inbound_for_update', new=lock_order), + self.assertRaises(ServiceException) as raised, + ): + await InboundService.edit_inbound_services(self.db, page, make_user(1)) + + self.assertEqual('入库单不存在', raised.exception.message) + scope.assert_not_awaited() + lock_order.assert_awaited_once_with(self.db, 404, warehouse_ids=None) + self.db.rollback.assert_awaited_once() + async def test_batch_delete_locks_and_preflights_all_before_ordered_delete(self) -> None: actions: list[str] = [] - async def lock(_db: AsyncSession, inbound_id: int) -> BizInboundOrder: + async def lock( + _db: AsyncSession, + inbound_id: int, + warehouse_ids: list[int] | None = None, + ) -> BizInboundOrder: + self.assertEqual([2, 5, 7, 8, 11], warehouse_ids) actions.append(f'lock-{inbound_id}') return make_inbound(inbound_id=inbound_id, warehouse_id=inbound_id) - async def access(_db: AsyncSession, _user: CurrentUserModel, warehouse_id: int) -> None: - actions.append(f'access-{warehouse_id}') - async def delete(_db: AsyncSession, inbound_id: int) -> None: actions.append(f'delete-{inbound_id}') @@ -405,17 +481,20 @@ class TestInboundDraftService(unittest.IsolatedAsyncioTestCase): actions.append('commit') self.db.commit.side_effect = commit + access = AsyncMock() with ( patch.object(OperationDao, 'get_inbound_for_update', new=AsyncMock(side_effect=lock)), - patch.object(AccessService, 'assert_warehouse_access', new=AsyncMock(side_effect=access)), + patch.object(AccessService, 'assert_warehouse_access', new=access), patch.object(OperationDao, 'delete_inbound', new=AsyncMock(side_effect=delete)), ): result = await InboundService.delete_inbound_services(self.db, [5, 2], self.user) self.assertEqual( - ['lock-2', 'access-2', 'lock-5', 'access-5', 'delete-2', 'delete-5', 'commit'], + ['lock-2', 'lock-5', 'delete-2', 'delete-5', 'commit'], actions, ) + self.default_scope.assert_awaited_once_with(self.db, 7) + access.assert_not_awaited() self.assertTrue(result.is_success) self.db.rollback.assert_not_awaited() @@ -429,7 +508,9 @@ class TestInboundDraftService(unittest.IsolatedAsyncioTestCase): patch.object( OperationDao, 'get_inbound_for_update', - new=AsyncMock(side_effect=lambda _db, inbound_id: records[inbound_id]), + new=AsyncMock( + side_effect=lambda _db, inbound_id, warehouse_ids=None: records[inbound_id], + ), ), patch.object(AccessService, 'assert_warehouse_access', new=AsyncMock()), patch.object(OperationDao, 'delete_inbound', new=delete), @@ -441,6 +522,39 @@ class TestInboundDraftService(unittest.IsolatedAsyncioTestCase): self.db.commit.assert_not_awaited() self.db.rollback.assert_awaited_once() + async def test_delete_scopes_parent_lock_and_hides_unmatched_document(self) -> None: + lock_order = AsyncMock(return_value=None) + delete = AsyncMock() + with ( + patch.object(AccessDao, 'list_user_warehouse_ids', new=AsyncMock(return_value=[7])) as scope, + patch.object(OperationDao, 'get_inbound_for_update', new=lock_order), + patch.object(OperationDao, 'delete_inbound', new=delete), + self.assertRaises(Exception) as raised, + ): + await InboundService.delete_inbound_services(self.db, [404], self.user) + + self.assertIsInstance(raised.exception, PermissionException) + self.assertEqual('无权访问该入库单', raised.exception.message) + scope.assert_awaited_once_with(self.db, 7) + lock_order.assert_awaited_once_with(self.db, 404, warehouse_ids=[7]) + delete.assert_not_awaited() + self.db.rollback.assert_awaited_once() + + async def test_admin_missing_delete_keeps_precise_not_found(self) -> None: + scope = AsyncMock() + lock_order = AsyncMock(return_value=None) + with ( + patch.object(AccessDao, 'list_user_warehouse_ids', new=scope), + patch.object(OperationDao, 'get_inbound_for_update', new=lock_order), + self.assertRaises(ServiceException) as raised, + ): + await InboundService.delete_inbound_services(self.db, [404], make_user(1)) + + self.assertEqual('入库单不存在', raised.exception.message) + scope.assert_not_awaited() + lock_order.assert_awaited_once_with(self.db, 404, warehouse_ids=None) + self.db.rollback.assert_awaited_once() + async def test_all_draft_mutations_convert_integrity_errors(self) -> None: add_page = InboundCreateModel(warehouseId=7, palletIds=[2]) with ( @@ -543,28 +657,52 @@ class TestInboundReadService(unittest.IsolatedAsyncioTestCase): self.assertEqual(0, actual.total) self.assertFalse(actual.has_next) - async def test_detail_loads_then_checks_warehouse_scope(self) -> None: + async def test_detail_pushes_warehouse_scope_into_parent_query(self) -> None: inbound = make_inbound() + get_detail = AsyncMock(return_value=inbound) access = AsyncMock() with ( - patch.object(OperationDao, 'get_inbound_by_id', new=AsyncMock(return_value=inbound)), + patch.object( + AccessDao, + 'list_user_warehouse_ids', + new=AsyncMock(return_value=[7, 3, 7]), + ) as scope, + patch.object(OperationDao, 'get_inbound_by_id', new=get_detail), patch.object(AccessService, 'assert_warehouse_access', new=access), ): actual = await InboundService.inbound_detail_services(self.db, 11, self.user) - access.assert_awaited_once_with(self.db, self.user, 7) + scope.assert_awaited_once_with(self.db, 7) + get_detail.assert_awaited_once_with(self.db, 11, warehouse_ids=[3, 7]) + access.assert_not_awaited() self.assertIsInstance(actual, InboundModel) - async def test_missing_detail_fails_stably_without_access_probe(self) -> None: - access = AsyncMock() + async def test_normal_user_detail_hides_missing_or_external_document(self) -> None: + get_detail = AsyncMock(return_value=None) with ( - patch.object(OperationDao, 'get_inbound_by_id', new=AsyncMock(return_value=None)), - patch.object(AccessService, 'assert_warehouse_access', new=access), - self.assertRaises(ServiceException) as raised, + patch.object(AccessDao, 'list_user_warehouse_ids', new=AsyncMock(return_value=[7])), + patch.object(OperationDao, 'get_inbound_by_id', new=get_detail), + self.assertRaises(Exception) as raised, ): await InboundService.inbound_detail_services(self.db, 404, self.user) + + self.assertIsInstance(raised.exception, PermissionException) + self.assertEqual('无权访问该入库单', raised.exception.message) + get_detail.assert_awaited_once_with(self.db, 404, warehouse_ids=[7]) + + async def test_admin_missing_detail_keeps_precise_not_found(self) -> None: + scope = AsyncMock() + get_detail = AsyncMock(return_value=None) + with ( + patch.object(AccessDao, 'list_user_warehouse_ids', new=scope), + patch.object(OperationDao, 'get_inbound_by_id', new=get_detail), + self.assertRaises(ServiceException) as raised, + ): + await InboundService.inbound_detail_services(self.db, 404, make_user(1)) + self.assertEqual('入库单不存在', raised.exception.message) - access.assert_not_awaited() + scope.assert_not_awaited() + get_detail.assert_awaited_once_with(self.db, 404, warehouse_ids=None) async def test_warehouse_options_filter_disabled_and_unauthorized_records(self) -> None: warehouses = [ @@ -722,6 +860,58 @@ class TestInboundConfirmService(unittest.IsolatedAsyncioTestCase): self.command = ConfirmCommandModel(idempotencyKey='confirm-key') self.fixed_time = datetime(2026, 7, 26, 14, 20) self.request_hash = InboundService._build_confirm_request_hash(11) + self.scope_patcher = patch.object( + AccessDao, + 'list_user_warehouse_ids', + new=AsyncMock(return_value=[7]), + ) + self.default_scope = self.scope_patcher.start() + self.addCleanup(self.scope_patcher.stop) + + async def test_confirm_scopes_initial_parent_and_hides_unmatched_document(self) -> None: + get_detail = AsyncMock(return_value=None) + get_by_key = AsyncMock() + lock_order = AsyncMock() + inventory_lock = AsyncMock() + with ( + patch.object(AccessDao, 'list_user_warehouse_ids', new=AsyncMock(return_value=[7])) as scope, + patch.object(OperationDao, 'get_inbound_by_id', new=get_detail), + patch.object(OperationDao, 'get_inbound_by_confirm_idempotency_key', new=get_by_key), + patch.object(OperationDao, 'get_inbound_for_update', new=lock_order), + patch.object(InventoryDao, 'get_inventories_for_update', new=inventory_lock), + self.assertRaises(Exception) as raised, + ): + await InboundService.confirm_inbound_services(self.db, 404, self.command, self.user) + + self.assertIsInstance(raised.exception, PermissionException) + self.assertEqual('无权访问该入库单', raised.exception.message) + scope.assert_awaited_once_with(self.db, 7) + get_detail.assert_awaited_once_with(self.db, 404, warehouse_ids=[7]) + get_by_key.assert_not_awaited() + lock_order.assert_not_awaited() + inventory_lock.assert_not_awaited() + self.db.rollback.assert_awaited_once() + + async def test_admin_missing_confirm_keeps_precise_not_found(self) -> None: + scope = AsyncMock() + get_detail = AsyncMock(return_value=None) + get_by_key = AsyncMock() + lock_order = AsyncMock() + with ( + patch.object(AccessDao, 'list_user_warehouse_ids', new=scope), + patch.object(OperationDao, 'get_inbound_by_id', new=get_detail), + patch.object(OperationDao, 'get_inbound_by_confirm_idempotency_key', new=get_by_key), + patch.object(OperationDao, 'get_inbound_for_update', new=lock_order), + self.assertRaises(ServiceException) as raised, + ): + await InboundService.confirm_inbound_services(self.db, 404, self.command, make_user(1)) + + self.assertEqual('入库单不存在', raised.exception.message) + scope.assert_not_awaited() + get_detail.assert_awaited_once_with(self.db, 404, warehouse_ids=None) + get_by_key.assert_not_awaited() + lock_order.assert_not_awaited() + self.db.rollback.assert_awaited_once() async def _run_successful_confirm(self, *, commit_side_effect: Exception | None = None) -> dict[str, Any]: inbound = make_inbound() @@ -730,7 +920,7 @@ class TestInboundConfirmService(unittest.IsolatedAsyncioTestCase): actions: list[str] = [] events: list[BizPalletEvent] = [] - async def detail(*_args: Any) -> BizInboundOrder: + async def detail(*_args: Any, **_kwargs: Any) -> BizInboundOrder: actions.append('detail') return inbound @@ -740,7 +930,7 @@ class TestInboundConfirmService(unittest.IsolatedAsyncioTestCase): async def by_key(*_args: Any) -> None: actions.append('idempotency-read') - async def lock_order(*_args: Any) -> BizInboundOrder: + async def lock_order(*_args: Any, **_kwargs: Any) -> BizInboundOrder: actions.append('order-lock') return inbound @@ -797,10 +987,10 @@ class TestInboundConfirmService(unittest.IsolatedAsyncioTestCase): patch.object(inbound_service_module, 'datetime'), ) with ( - patches[0], - patches[1], + patches[0] as detail_read, + patches[1] as access_probe, patches[2], - patches[3], + patches[3] as order_lock, patches[4] as inventory_lock, patches[5] as pallet_lock, patches[6] as ensure_transition, @@ -817,6 +1007,9 @@ class TestInboundConfirmService(unittest.IsolatedAsyncioTestCase): 'inventories': inventories, 'events': events, 'actions': actions, + 'detail_read': detail_read, + 'access_probe': access_probe, + 'order_lock': order_lock, 'inventory_lock': inventory_lock, 'pallet_lock': pallet_lock, 'ensure_transition': ensure_transition, @@ -828,10 +1021,8 @@ class TestInboundConfirmService(unittest.IsolatedAsyncioTestCase): self.assertEqual( [ 'detail', - 'access', 'idempotency-read', 'order-lock', - 'access', 'inventory-lock', 'pallet-lock', 'transition-UNSTOCKED-EMPTY_IN_WAREHOUSE', @@ -843,6 +1034,10 @@ class TestInboundConfirmService(unittest.IsolatedAsyncioTestCase): ], captured['actions'], ) + self.default_scope.assert_awaited_once_with(self.db, 7) + captured['detail_read'].assert_awaited_once_with(self.db, 11, warehouse_ids=[7]) + captured['order_lock'].assert_awaited_once_with(self.db, 11, warehouse_ids=[7]) + captured['access_probe'].assert_not_awaited() captured['inventory_lock'].assert_awaited_once_with(self.db, [2, 5]) captured['pallet_lock'].assert_awaited_once_with(self.db, [2, 5]) captured['ensure_transition'].assert_has_calls( @@ -962,8 +1157,8 @@ class TestInboundConfirmService(unittest.IsolatedAsyncioTestCase): ): result = await InboundService.confirm_inbound_services(self.db, 11, self.command, self.user) - lock.assert_awaited_once_with(self.db, 11) - self.assertEqual(3, access.await_count) + lock.assert_awaited_once_with(self.db, 11, warehouse_ids=[7]) + access.assert_awaited_once_with(self.db, self.user, 7) inventory_lock.assert_not_awaited() self.db.commit.assert_not_awaited() self.db.rollback.assert_awaited_once() @@ -992,7 +1187,7 @@ class TestInboundConfirmService(unittest.IsolatedAsyncioTestCase): ): await InboundService.confirm_inbound_services(self.db, 11, self.command, self.user) - self.assertEqual(2, access.await_count) + access.assert_awaited_once_with(self.db, self.user, 8) lock.assert_not_awaited() self.db.rollback.assert_awaited_once() @@ -1035,7 +1230,7 @@ class TestInboundConfirmService(unittest.IsolatedAsyncioTestCase): ): result = await InboundService.confirm_inbound_services(self.db, 11, self.command, self.user) - get_detail.assert_awaited_once_with(self.db, 11) + get_detail.assert_awaited_once_with(self.db, 11, warehouse_ids=[7]) inventory_lock.assert_awaited_once_with(self.db, [5]) self.assertEqual('EMPTY_IN_WAREHOUSE', inventory.flow_state) self.assertTrue(result.is_success) @@ -1063,7 +1258,7 @@ class TestInboundConfirmService(unittest.IsolatedAsyncioTestCase): ): result = await InboundService.confirm_inbound_services(self.db, 11, self.command, self.user) - self.assertEqual(3, access.await_count) + access.assert_awaited_once_with(self.db, self.user, 7) self.assertEqual(2, key_reads.await_count) self.assertEqual(2, self.db.rollback.await_count) self.assertTrue(result.is_success) @@ -1098,6 +1293,68 @@ class TestInboundConfirmService(unittest.IsolatedAsyncioTestCase): self.assertEqual(2, self.db.rollback.await_count) +class TestInboundLogPermission(unittest.IsolatedAsyncioTestCase): + async def test_real_inbound_log_wrapper_preserves_permission_code(self) -> None: + async def receive() -> dict[str, Any]: + return { + 'type': 'http.request', + 'body': b'{"warehouseId":404,"palletIds":[2]}', + 'more_body': False, + } + + request = Request( + { + 'type': 'http', + 'method': 'POST', + 'scheme': 'http', + 'path': '/business/inbound', + 'raw_path': b'/business/inbound', + 'root_path': '', + 'query_string': b'', + 'headers': [ + (b'content-type', b'application/json'), + (b'user-agent', b'unittest'), + ], + 'client': ('127.0.0.1', 12345), + 'server': ('testserver', 80), + }, + receive=receive, + ) + user = make_user() + db = MagicMock(spec=AsyncSession) + service = AsyncMock( + side_effect=PermissionException(data='', message='无权访问该仓库'), + ) + enqueue_log = AsyncMock() + user_token = RequestContext.set_current_user(user) + try: + with ( + patch.object(InboundService, 'add_inbound_services', new=service), + patch( + 'common.annotation.log_annotation.Log._get_oper_location', + new=AsyncMock(return_value='内网IP'), + ), + patch.object(LogQueueService, 'enqueue_operation_log', new=enqueue_log), + patch('common.annotation.log_annotation.logger.exception') as exception_log, + ): + response = await inbound_controller_module.add_business_inbound( + request=request, + add_inbound=InboundCreateModel(warehouseId=404, palletIds=[2]), + query_db=db, + current_user=user, + ) + finally: + RequestContext.reset_current_user(user_token) + + payload = json.loads(response.body) + self.assertEqual(403, payload['code']) + self.assertEqual('无权访问该仓库', payload['msg']) + self.assertFalse(payload['success']) + service.assert_awaited_once() + exception_log.assert_not_called() + enqueue_log.assert_awaited_once() + + class TestInboundController(unittest.TestCase): @staticmethod def _routes() -> dict[tuple[str, str], APIRoute]: diff --git a/shierfeng-fastapi-backend/tests/module_business/test_inventory_dao.py b/shierfeng-fastapi-backend/tests/module_business/test_inventory_dao.py index 99e6099..ae5ae73 100644 --- a/shierfeng-fastapi-backend/tests/module_business/test_inventory_dao.py +++ b/shierfeng-fastapi-backend/tests/module_business/test_inventory_dao.py @@ -121,7 +121,7 @@ class TestDaoPublicContract(unittest.TestCase): def test_operation_dao_critical_parameter_names_and_defaults_are_stable(self) -> None: expected = { - 'get_inbound_for_update': ('db', 'inbound_id'), + 'get_inbound_for_update': ('db', 'inbound_id', 'warehouse_ids'), 'get_shipment_for_update': ('db', 'shipment_id'), 'get_return_for_update': ('db', 'return_id'), 'get_active_filling_for_update': ('db', 'filling_id'), @@ -131,7 +131,7 @@ class TestDaoPublicContract(unittest.TestCase): 'get_usage_by_idempotency_key': ('db', 'idempotency_key'), 'get_return_by_request_idempotency_key': ('db', 'idempotency_key'), 'get_return_by_confirm_idempotency_key': ('db', 'idempotency_key'), - 'get_inbound_by_id': ('db', 'inbound_id'), + 'get_inbound_by_id': ('db', 'inbound_id', 'warehouse_ids'), 'add_inbound': ('db', 'inbound'), 'update_inbound': ('db', 'inbound'), 'delete_inbound': ('db', 'inbound_id'), @@ -164,6 +164,14 @@ class TestDaoPublicContract(unittest.TestCase): self.assertIs(True, inbound_list_signature.parameters['is_page'].default) self.assertIsNone(inbound_list_signature.parameters['warehouse_ids'].default) + for method_name in ('get_inbound_for_update', 'get_inbound_by_id'): + with self.subTest(method=method_name): + signature = inspect.signature(getattr(OperationDao, method_name)) + warehouse_ids = signature.parameters.get('warehouse_ids') + self.assertIsNotNone(warehouse_ids) + if warehouse_ids is not None: + self.assertIsNone(warehouse_ids.default) + for method_name in ( 'get_filling_list', 'get_shipment_list', @@ -446,6 +454,30 @@ class TestOperationDaoReads(unittest.IsolatedAsyncioTestCase): ) self.assertIn('FOR UPDATE', item_sql) + async def test_inbound_for_update_applies_scope_before_locking_items(self) -> None: + self.assertIn( + 'warehouse_ids', + inspect.signature(OperationDao.get_inbound_for_update).parameters, + ) + parent_result = MagicMock() + parent_result.scalar_one_or_none.return_value = None + self.db.execute.return_value = parent_result + + actual = await OperationDao.get_inbound_for_update( + self.db, + inbound_id=11, + warehouse_ids=[7, 3, 7], + ) + + self.assertIsNone(actual) + self.db.execute.assert_awaited_once() + parent_statement = self.db.execute.await_args.args[0] + for dialect in (mysql.dialect(), postgresql.dialect()): + with self.subTest(dialect=dialect.name): + sql = compile_sql(parent_statement, dialect) + self.assertIn('biz_inbound_order.warehouse_id IN (3, 7)', sql) + self.assertIn('FOR UPDATE', sql) + async def test_idempotency_queries_use_the_exact_declared_column(self) -> None: cases = ( ( @@ -753,6 +785,29 @@ class TestOperationDaoReads(unittest.IsolatedAsyncioTestCase): self.db.execute.assert_awaited_once() + async def test_inbound_detail_empty_scope_is_false_and_skips_items(self) -> None: + self.assertIn( + 'warehouse_ids', + inspect.signature(OperationDao.get_inbound_by_id).parameters, + ) + parent_result = MagicMock() + parent_result.scalar_one_or_none.return_value = None + self.db.execute.return_value = parent_result + + actual = await OperationDao.get_inbound_by_id( + self.db, + inbound_id=11, + warehouse_ids=[], + ) + + self.assertIsNone(actual) + self.db.execute.assert_awaited_once() + parent_statement = self.db.execute.await_args.args[0] + mysql_sql = compile_sql(parent_statement, mysql.dialect()) + postgresql_sql = compile_sql(parent_statement, postgresql.dialect()) + self.assertIn('false = 1', mysql_sql) + self.assertIn('false', postgresql_sql) + async def test_simple_details_target_primary_keys(self) -> None: cases = ( (OperationDao.get_filling_by_id, BizFillingRecord, 'filling_id', 14),