From 7317631a1363e6bf35478b9dd91cae3635b85c0c 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 19:11:52 +0800 Subject: [PATCH] =?UTF-8?q?=E5=8A=9F=E8=83=BD=EF=BC=9A=E5=AE=9E=E7=8E=B0?= =?UTF-8?q?=E5=A4=9A=E6=89=98=E7=9B=98=E5=8F=91=E8=B4=A7=E5=87=BA=E5=BA=93?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../controller/shipment_controller.py | 213 +++ .../module_business/dao/access_dao.py | 24 +- .../module_business/dao/operation_dao.py | 47 +- .../module_business/service/access_service.py | 13 +- .../service/shipment_service.py | 840 +++++++++ .../module_business/test_access_service.py | 59 +- .../module_business/test_inventory_dao.py | 170 +- .../module_business/test_shipment_service.py | 1526 +++++++++++++++++ 8 files changed, 2867 insertions(+), 25 deletions(-) create mode 100644 shierfeng-fastapi-backend/module_business/controller/shipment_controller.py create mode 100644 shierfeng-fastapi-backend/module_business/service/shipment_service.py create mode 100644 shierfeng-fastapi-backend/tests/module_business/test_shipment_service.py diff --git a/shierfeng-fastapi-backend/module_business/controller/shipment_controller.py b/shierfeng-fastapi-backend/module_business/controller/shipment_controller.py new file mode 100644 index 0000000..5476d11 --- /dev/null +++ b/shierfeng-fastapi-backend/module_business/controller/shipment_controller.py @@ -0,0 +1,213 @@ +from typing import Annotated + +from fastapi import Path, Query, Request, Response +from sqlalchemy.ext.asyncio import AsyncSession + +from common.annotation.log_annotation import Log +from common.aspect.db_seesion import DBSessionDependency +from common.aspect.interface_auth import UserInterfaceAuthDependency +from common.aspect.pre_auth import CurrentUserDependency, PreAuthDependency +from common.enums import BusinessType +from common.router import APIRouterPro +from common.vo import DataResponseModel, PageResponseModel, ResponseBaseModel +from module_admin.entity.vo.user_vo import CurrentUserModel +from module_business.entity.vo.operation_vo import ( + BusinessOptionModel, + ConfirmCommandModel, + PalletScanResponseModel, + ShipmentCreateModel, + ShipmentModel, + ShipmentPageQueryModel, + ShipmentScanQueryModel, + ShipmentUpdateModel, +) +from module_business.service.shipment_service import ShipmentService +from utils.response_util import ResponseUtil + +shipment_controller = APIRouterPro( + prefix='/business/shipment', + tags=['业务管理-整托发货'], + dependencies=[PreAuthDependency()], +) + + +@shipment_controller.get( + '/list', + summary='获取整托发货单分页列表接口', + description='按当前用户仓库与现场双数据范围获取整托发货单', + response_model=PageResponseModel[ShipmentModel], + dependencies=[UserInterfaceAuthDependency('business:shipment:list')], +) +async def get_business_shipment_list( + request: Request, + shipment_page_query: Annotated[ShipmentPageQueryModel, Query()], + query_db: Annotated[AsyncSession, DBSessionDependency()], + current_user: Annotated[CurrentUserModel, CurrentUserDependency()], +) -> Response: + result = await ShipmentService.get_shipment_list_services( + query_db, + shipment_page_query, + current_user, + is_page=True, + ) + return ResponseUtil.success(model_content=result) + + +@shipment_controller.get( + '/warehouse-options', + summary='获取发货来源仓库选项接口', + description='仅返回当前用户有权访问的启用仓库', + response_model=DataResponseModel[list[BusinessOptionModel]], + dependencies=[UserInterfaceAuthDependency('business:shipment:query')], +) +async def get_business_shipment_warehouse_options( + request: Request, + query_db: Annotated[AsyncSession, DBSessionDependency()], + current_user: Annotated[CurrentUserModel, CurrentUserDependency()], +) -> Response: + result = await ShipmentService.get_warehouse_options_services(query_db, current_user) + return ResponseUtil.success(data=result) + + +@shipment_controller.get( + '/site-options', + summary='获取发货目标现场选项接口', + description='仅返回当前用户有权访问的启用客户现场', + response_model=DataResponseModel[list[BusinessOptionModel]], + dependencies=[UserInterfaceAuthDependency('business:shipment:query')], +) +async def get_business_shipment_site_options( + request: Request, + query_db: Annotated[AsyncSession, DBSessionDependency()], + current_user: Annotated[CurrentUserModel, CurrentUserDependency()], +) -> Response: + result = await ShipmentService.get_site_options_services(query_db, current_user) + return ResponseUtil.success(data=result) + + +@shipment_controller.get( + '/scan/{qr_token}', + summary='扫描整托发货托盘接口', + description='查询托盘是否可加入指定仓库至现场的整托发货单', + response_model=DataResponseModel[PalletScanResponseModel], + dependencies=[UserInterfaceAuthDependency('business:shipment:query')], +) +async def scan_business_shipment_pallet( + request: Request, + qr_token: Annotated[ + str, + Path(min_length=1, max_length=128, pattern=r'\S', description='二维码固定标识'), + ], + shipment_scan_query: Annotated[ShipmentScanQueryModel, Query()], + query_db: Annotated[AsyncSession, DBSessionDependency()], + current_user: Annotated[CurrentUserModel, CurrentUserDependency()], +) -> Response: + result = await ShipmentService.scan_shipment_pallet_services( + query_db, + qr_token, + shipment_scan_query, + current_user, + ) + return ResponseUtil.success(data=result) + + +@shipment_controller.get( + '/{shipment_id}', + summary='获取整托发货单详情接口', + description='按仓库与现场双数据范围获取指定整托发货单', + response_model=DataResponseModel[ShipmentModel], + dependencies=[UserInterfaceAuthDependency('business:shipment:query')], +) +async def query_detail_business_shipment( + request: Request, + shipment_id: Annotated[int, Path(gt=0, description='发货单ID')], + query_db: Annotated[AsyncSession, DBSessionDependency()], + current_user: Annotated[CurrentUserModel, CurrentUserDependency()], +) -> Response: + result = await ShipmentService.shipment_detail_services(query_db, shipment_id, current_user) + return ResponseUtil.success(data=result) + + +@shipment_controller.post( + '', + summary='新增整托发货草稿接口', + description='新增多托盘整托发货草稿', + response_model=ResponseBaseModel, + dependencies=[UserInterfaceAuthDependency('business:shipment:add')], +) +@Log(title='整托发货', business_type=BusinessType.INSERT) +async def add_business_shipment( + request: Request, + add_shipment: ShipmentCreateModel, + query_db: Annotated[AsyncSession, DBSessionDependency()], + current_user: Annotated[CurrentUserModel, CurrentUserDependency()], +) -> Response: + result = await ShipmentService.add_shipment_services(query_db, add_shipment, current_user) + return ResponseUtil.success(msg=result.message) + + +@shipment_controller.put( + '', + summary='编辑整托发货草稿接口', + description='编辑尚未确认的多托盘整托发货草稿', + response_model=ResponseBaseModel, + dependencies=[UserInterfaceAuthDependency('business:shipment:edit')], +) +@Log(title='整托发货', business_type=BusinessType.UPDATE) +async def edit_business_shipment( + request: Request, + edit_shipment: ShipmentUpdateModel, + query_db: Annotated[AsyncSession, DBSessionDependency()], + current_user: Annotated[CurrentUserModel, CurrentUserDependency()], +) -> Response: + result = await ShipmentService.edit_shipment_services(query_db, edit_shipment, current_user) + return ResponseUtil.success(msg=result.message) + + +@shipment_controller.delete( + '/{shipment_ids}', + summary='删除整托发货草稿接口', + description='原子删除一个或多个尚未确认的整托发货草稿', + response_model=ResponseBaseModel, + dependencies=[UserInterfaceAuthDependency('business:shipment:remove')], +) +@Log(title='整托发货', business_type=BusinessType.DELETE) +async def delete_business_shipment( + request: Request, + shipment_ids: Annotated[ + str, + Path(pattern=r'^[1-9]\d*(,[1-9]\d*)*$', description='需要删除的发货单ID'), + ], + query_db: Annotated[AsyncSession, DBSessionDependency()], + current_user: Annotated[CurrentUserModel, CurrentUserDependency()], +) -> Response: + result = await ShipmentService.delete_shipment_services( + query_db, + [int(shipment_id) for shipment_id in shipment_ids.split(',')], + current_user, + ) + return ResponseUtil.success(msg=result.message) + + +@shipment_controller.put( + '/{shipment_id}/confirm', + summary='确认整托发货单接口', + description='原子确认整张发货单并将全部托盘移至目标现场', + response_model=ResponseBaseModel, + dependencies=[UserInterfaceAuthDependency('business:shipment:confirm')], +) +@Log(title='整托发货', business_type=BusinessType.UPDATE) +async def confirm_business_shipment( + request: Request, + shipment_id: Annotated[int, Path(gt=0, description='发货单ID')], + confirm_command: ConfirmCommandModel, + query_db: Annotated[AsyncSession, DBSessionDependency()], + current_user: Annotated[CurrentUserModel, CurrentUserDependency()], +) -> Response: + result = await ShipmentService.confirm_shipment_services( + query_db, + shipment_id, + confirm_command, + current_user, + ) + return ResponseUtil.success(msg=result.message) diff --git a/shierfeng-fastapi-backend/module_business/dao/access_dao.py b/shierfeng-fastapi-backend/module_business/dao/access_dao.py index 0745432..d4a329b 100644 --- a/shierfeng-fastapi-backend/module_business/dao/access_dao.py +++ b/shierfeng-fastapi-backend/module_business/dao/access_dao.py @@ -28,16 +28,22 @@ class AccessDao: return mapped_user_id is not None @classmethod - async def user_has_site(cls, db: AsyncSession, user_id: int, site_id: int) -> bool: + async def user_has_site( + cls, + db: AsyncSession, + user_id: int, + site_id: int, + *, + for_update: bool = False, + ) -> bool: """判断用户是否关联指定客户现场。""" - mapped_user_id = ( - await db.execute( - select(BizUserSite.user_id).where( - BizUserSite.user_id == user_id, - BizUserSite.site_id == site_id, - ) - ) - ).scalar_one_or_none() + statement = select(BizUserSite.user_id).where( + BizUserSite.user_id == user_id, + BizUserSite.site_id == site_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/dao/operation_dao.py b/shierfeng-fastapi-backend/module_business/dao/operation_dao.py index 9df1773..fb699ae 100644 --- a/shierfeng-fastapi-backend/module_business/dao/operation_dao.py +++ b/shierfeng-fastapi-backend/module_business/dao/operation_dao.py @@ -69,6 +69,22 @@ class OperationDao: scoped_warehouse_ids = sorted(set(warehouse_ids)) return [BizFillingRecord.warehouse_id.in_(scoped_warehouse_ids) if scoped_warehouse_ids else false()] + @staticmethod + def _shipment_scope_conditions( + warehouse_ids: list[int] | None, + site_ids: list[int] | None, + ) -> list[ColumnElement[bool]]: + conditions = [] + if warehouse_ids is not None: + scoped_warehouse_ids = sorted(set(warehouse_ids)) + conditions.append( + BizShipmentOrder.source_warehouse_id.in_(scoped_warehouse_ids) if scoped_warehouse_ids else false() + ) + if site_ids is not None: + scoped_site_ids = sorted(set(site_ids)) + conditions.append(BizShipmentOrder.target_site_id.in_(scoped_site_ids) if scoped_site_ids else false()) + return conditions + @staticmethod async def _paginate( db: AsyncSession, @@ -243,13 +259,32 @@ class OperationDao: cls, db: AsyncSession, shipment_id: int, + warehouse_ids: list[int] | None = None, + site_ids: list[int] | None = None, ) -> BizShipmentOrder | None: - return await cls._get_for_update( + shipment = await cls._get_for_update( db, BizShipmentOrder, BizShipmentOrder.shipment_id, shipment_id, + *cls._shipment_scope_conditions(warehouse_ids, site_ids), ) + if shipment is None: + return None + + # 父单锁定后读取当前明细，避免 MySQL RR 沿用确认前的快照。 + item_result = await db.execute( + select(BizShipmentItem) + .where(BizShipmentItem.shipment_id == shipment_id) + .order_by(BizShipmentItem.pallet_id, BizShipmentItem.shipment_item_id) + .with_for_update() + .execution_options(populate_existing=True) + ) + shipment.items = sorted( + item_result.scalars().all(), + key=lambda item: (item.pallet_id, item.shipment_item_id), + ) + return shipment @classmethod async def get_return_for_update( @@ -297,12 +332,15 @@ class OperationDao: cls, db: AsyncSession, idempotency_key: str, + warehouse_ids: list[int] | None = None, + site_ids: list[int] | None = None, ) -> BizShipmentOrder | None: return await cls._get_by_key( db, BizShipmentOrder, BizShipmentOrder.confirm_idempotency_key, idempotency_key, + *cls._shipment_scope_conditions(warehouse_ids, site_ids), ) @classmethod @@ -520,8 +558,10 @@ class OperationDao: db: AsyncSession, query_object: ShipmentPageQueryModel, is_page: bool = True, + warehouse_ids: list[int] | None = None, + site_ids: list[int] | None = None, ) -> PageModel | list[dict[str, Any]]: - conditions = [] + conditions = cls._shipment_scope_conditions(warehouse_ids, site_ids) if query_object.shipment_no: conditions.append(BizShipmentOrder.shipment_no.like(f'%{query_object.shipment_no}%')) if query_object.source_warehouse_id: @@ -545,6 +585,8 @@ class OperationDao: cls, db: AsyncSession, shipment_id: int, + warehouse_ids: list[int] | None = None, + site_ids: list[int] | None = None, ) -> BizShipmentOrder | None: return await cls._get_order_detail( db, @@ -554,6 +596,7 @@ class OperationDao: BizShipmentItem, BizShipmentItem.shipment_id, BizShipmentItem.shipment_item_id, + *cls._shipment_scope_conditions(warehouse_ids, site_ids), ) @classmethod diff --git a/shierfeng-fastapi-backend/module_business/service/access_service.py b/shierfeng-fastapi-backend/module_business/service/access_service.py index fe9b0b8..f8b254a 100644 --- a/shierfeng-fastapi-backend/module_business/service/access_service.py +++ b/shierfeng-fastapi-backend/module_business/service/access_service.py @@ -38,9 +38,20 @@ class AccessService: db: AsyncSession, current_user: CurrentUserModel, site_id: int, + *, + for_update: bool = False, ) -> None: """校验当前用户是否有权访问指定客户现场。""" if current_user.user.admin: return - if not await AccessDao.user_has_site(db, current_user.user.user_id, site_id): + if for_update: + has_site = await AccessDao.user_has_site( + db, + current_user.user.user_id, + site_id, + for_update=True, + ) + else: + has_site = await AccessDao.user_has_site(db, current_user.user.user_id, site_id) + if not has_site: raise PermissionException(data='', message='无权访问该客户现场') diff --git a/shierfeng-fastapi-backend/module_business/service/shipment_service.py b/shierfeng-fastapi-backend/module_business/service/shipment_service.py new file mode 100644 index 0000000..69a7a87 --- /dev/null +++ b/shierfeng-fastapi-backend/module_business/service/shipment_service.py @@ -0,0 +1,840 @@ +import hashlib +import json +import secrets +from datetime import datetime +from typing import Any, NoReturn + +from sqlalchemy.exc import IntegrityError +from sqlalchemy.ext.asyncio import AsyncSession + +from common.vo import CrudResponseModel, PageModel +from exceptions.exception import PermissionException, ServiceException +from module_admin.entity.vo.user_vo import CurrentUserModel +from module_business.dao.access_dao import AccessDao +from module_business.dao.inventory_dao import InventoryDao +from module_business.dao.master_dao import MasterDao +from module_business.dao.operation_dao import OperationDao +from module_business.domain.enums import DocumentStatus, PalletEventType, PalletFlowState +from module_business.domain.pallet_state_machine import PalletStateMachine +from module_business.entity.do.master_do import BizPallet +from module_business.entity.do.operation_do import ( + BizFillingRecord, + BizPalletEvent, + BizPalletInventory, + BizShipmentItem, + BizShipmentOrder, +) +from module_business.entity.vo.master_vo import SitePageQueryModel, WarehousePageQueryModel +from module_business.entity.vo.operation_vo import ( + BusinessOptionModel, + ConfirmCommandModel, + PalletScanResponseModel, + ShipmentCreateModel, + ShipmentModel, + ShipmentPageQueryModel, + ShipmentScanQueryModel, + ShipmentUpdateModel, +) +from module_business.service.access_service import AccessService + + +class ShipmentService: + """多托盘整托发货服务层。""" + + @staticmethod + def _current_identity(current_user: CurrentUserModel) -> tuple[int, str]: + user = current_user.user + if user is None or user.user_id is None or user.user_id <= 0 or not user.user_name: + raise ServiceException(message='当前用户信息无效') + return user.user_id, user.user_name + + @staticmethod + def _is_admin(current_user: CurrentUserModel) -> bool: + return bool(current_user.user and current_user.user.admin) + + @staticmethod + def _record_value(record: Any, field: str) -> Any: + if not isinstance(record, dict): + return getattr(record, field) + if field in record: + return record[field] + words = field.split('_') + alias = words[0] + ''.join(word.capitalize() for word in words[1:]) + return record.get(alias) + + @classmethod + def _raise_shipment_not_accessible(cls, current_user: CurrentUserModel) -> NoReturn: + if cls._is_admin(current_user): + raise ServiceException(message='发货单不存在') + raise PermissionException(data='', message='无权访问该发货单') + + @classmethod + def _raise_pallet_not_accessible( + cls, + current_user: CurrentUserModel, + admin_message: str = '托盘不存在', + ) -> NoReturn: + if cls._is_admin(current_user): + raise ServiceException(message=admin_message) + raise PermissionException(data='', message='无权访问该托盘') + + @classmethod + async def _scopes( + cls, + query_db: AsyncSession, + current_user: CurrentUserModel, + ) -> tuple[list[int] | None, list[int] | None]: + if cls._is_admin(current_user): + return None, None + user_id, _ = cls._current_identity(current_user) + warehouse_ids = sorted(set(await AccessDao.list_user_warehouse_ids(query_db, user_id))) + site_ids = sorted(set(await AccessDao.list_user_site_ids(query_db, user_id))) + return warehouse_ids, site_ids + + @staticmethod + async def _validate_active_route( + query_db: AsyncSession, + source_warehouse_id: int, + target_site_id: int, + current_user: CurrentUserModel, + ) -> tuple[Any, Any]: + await AccessService.assert_warehouse_access(query_db, current_user, source_warehouse_id) + warehouse = await MasterDao.get_warehouse(query_db, source_warehouse_id) + if warehouse is None: + raise ServiceException(message='来源仓库不存在') + if warehouse.status != '0': + raise ServiceException(message='来源仓库已停用') + await AccessService.assert_site_access(query_db, current_user, target_site_id) + site = await MasterDao.get_site(query_db, target_site_id) + if site is None: + raise ServiceException(message='目标现场不存在') + if site.status != '0': + raise ServiceException(message='目标现场已停用') + return warehouse, site + + @staticmethod + def _validate_pallet_ids(pallet_ids: list[int]) -> list[int]: + if not pallet_ids or any(type(pallet_id) is not int or pallet_id <= 0 for pallet_id in pallet_ids): + raise ServiceException(message='托盘ID必须为正整数') + if len(pallet_ids) != len(set(pallet_ids)): + raise ServiceException(message='托盘ID不能重复') + return sorted(pallet_ids) + + @staticmethod + def _validate_locked_pallet( + pallet: BizPallet, + inventory: BizPalletInventory, + filling: BizFillingRecord, + source_warehouse_id: int, + ) -> None: + if pallet.pallet_id != inventory.pallet_id: + raise ServiceException(message='托盘资产与库存不一致') + if pallet.asset_status != '0': + raise ServiceException(message='托盘资产已停用，不允许发货') + if ( + inventory.flow_state != PalletFlowState.FILLED_IN_WAREHOUSE.value + or inventory.warehouse_id != source_warehouse_id + or inventory.site_id is not None + or inventory.current_line_id is not None + or inventory.active_filling_id is None + ): + raise ServiceException(message='托盘不符合整托发货条件') + if type(inventory.remaining_packages) is not int or inventory.remaining_packages <= 0: + raise ServiceException(message='托盘剩余包数必须为正整数') + if inventory.remaining_packages > pallet.capacity_packages: + raise ServiceException(message='托盘剩余包数不能超过容量') + if type(inventory.version) is not int or inventory.version < 0: + raise ServiceException(message='托盘库存版本无效') + if ( + filling.filling_id != inventory.active_filling_id + or filling.pallet_id != pallet.pallet_id + or filling.warehouse_id != source_warehouse_id + or filling.active_slot != 'ACTIVE' + ): + raise ServiceException(message='托盘活动装填周期无效') + + @classmethod + async def _lock_and_validate_pallets( + cls, + query_db: AsyncSession, + pallet_ids: list[int], + source_warehouse_id: int, + ) -> tuple[list[BizPalletInventory], list[BizPallet], list[BizFillingRecord]]: + ordered_ids = cls._validate_pallet_ids(pallet_ids) + inventories = await InventoryDao.get_inventories_for_update(query_db, ordered_ids) + pallets = await InventoryDao.get_pallets_by_ids(query_db, ordered_ids) + if ( + len(inventories) != len(ordered_ids) + or len(pallets) != len(ordered_ids) + or {inventory.pallet_id for inventory in inventories} != set(ordered_ids) + or {pallet.pallet_id for pallet in pallets} != set(ordered_ids) + ): + raise ServiceException(message='存在不存在或库存缺失的托盘') + + inventory_by_id = {inventory.pallet_id: inventory for inventory in inventories} + pallet_by_id = {pallet.pallet_id: pallet for pallet in pallets} + active_filling_ids = [] + for pallet_id in ordered_ids: + active_filling_id = inventory_by_id[pallet_id].active_filling_id + if active_filling_id is None: + raise ServiceException(message='托盘不存在活动装填周期') + active_filling_ids.append(active_filling_id) + + fillings = [] + for filling_id in sorted(set(active_filling_ids)): + filling = await OperationDao.get_active_filling_for_update(query_db, filling_id) + if filling is None: + raise ServiceException(message='托盘活动装填周期不存在或已关闭') + fillings.append(filling) + if len(fillings) != len(ordered_ids): + raise ServiceException(message='托盘活动装填周期不完整') + filling_by_id = {filling.filling_id: filling for filling in fillings} + + ordered_inventories = [inventory_by_id[pallet_id] for pallet_id in ordered_ids] + ordered_pallets = [pallet_by_id[pallet_id] for pallet_id in ordered_ids] + ordered_fillings = [filling_by_id[inventory.active_filling_id] for inventory in ordered_inventories] + for pallet, inventory, filling in zip( + ordered_pallets, + ordered_inventories, + ordered_fillings, + strict=True, + ): + cls._validate_locked_pallet(pallet, inventory, filling, source_warehouse_id) + return ordered_inventories, ordered_pallets, ordered_fillings + + @staticmethod + def _build_items( + shipment_id: int, + pallet_ids: list[int], + user_name: str, + current_time: datetime, + ) -> list[BizShipmentItem]: + return [ + BizShipmentItem( + shipment_id=shipment_id, + pallet_id=pallet_id, + active_filling_id=None, + package_qty=None, + create_by=user_name, + create_time=current_time, + ) + for pallet_id in sorted(pallet_ids) + ] + + @staticmethod + def _success(message: str) -> CrudResponseModel: + return CrudResponseModel(is_success=True, message=message) + + @staticmethod + def _build_confirm_request_hash(shipment_id: int) -> str: + payload = json.dumps( + {'shipment_id': shipment_id}, + sort_keys=True, + separators=(',', ':'), + ).encode() + return hashlib.sha256(payload).hexdigest() + + @classmethod + async def get_shipment_list_services( + cls, + query_db: AsyncSession, + query_object: ShipmentPageQueryModel, + current_user: CurrentUserModel, + is_page: bool = True, + ) -> PageModel | list[ShipmentModel]: + warehouse_ids, site_ids = await cls._scopes(query_db, current_user) + if ( + warehouse_ids is not None + and query_object.source_warehouse_id is not None + and query_object.source_warehouse_id not in warehouse_ids + ): + raise PermissionException(data='', message='无权访问该仓库') + if ( + site_ids is not None + and query_object.target_site_id is not None + and query_object.target_site_id not in site_ids + ): + raise PermissionException(data='', message='无权访问该客户现场') + if warehouse_ids == [] or site_ids == []: + if is_page: + return PageModel( + rows=[], + pageNum=query_object.page_num, + pageSize=query_object.page_size, + total=0, + hasNext=False, + ) + return [] + + result = await OperationDao.get_shipment_list( + query_db, + query_object, + is_page, + warehouse_ids=warehouse_ids, + site_ids=site_ids, + ) + rows = result.rows if isinstance(result, PageModel) else result + models = [ShipmentModel.model_validate(record) for record in rows] + if isinstance(result, PageModel): + return result.model_copy(update={'rows': models}) + return models + + @classmethod + async def shipment_detail_services( + cls, + query_db: AsyncSession, + shipment_id: int, + current_user: CurrentUserModel, + ) -> ShipmentModel: + warehouse_ids, site_ids = await cls._scopes(query_db, current_user) + shipment = await OperationDao.get_shipment_by_id( + query_db, + shipment_id, + warehouse_ids=warehouse_ids, + site_ids=site_ids, + ) + if shipment is None: + cls._raise_shipment_not_accessible(current_user) + return ShipmentModel.model_validate(shipment) + + @classmethod + async def get_warehouse_options_services( + cls, + query_db: AsyncSession, + current_user: CurrentUserModel, + ) -> list[BusinessOptionModel]: + warehouse_ids, _ = await cls._scopes(query_db, current_user) + allowed_ids = None if warehouse_ids is None else set(warehouse_ids) + warehouses = await MasterDao.list_warehouse( + query_db, + WarehousePageQueryModel(status='0'), + False, + ) + options = [] + for warehouse in warehouses: + warehouse_id = cls._record_value(warehouse, 'warehouse_id') + if cls._record_value(warehouse, 'status') != '0': + continue + if allowed_ids is not None and warehouse_id not in allowed_ids: + continue + options.append( + BusinessOptionModel( + id=warehouse_id, + code=cls._record_value(warehouse, 'warehouse_code'), + name=cls._record_value(warehouse, 'warehouse_name'), + ) + ) + return sorted(options, key=lambda option: option.id) + + @classmethod + async def get_site_options_services( + cls, + query_db: AsyncSession, + current_user: CurrentUserModel, + ) -> list[BusinessOptionModel]: + _, site_ids = await cls._scopes(query_db, current_user) + allowed_ids = None if site_ids is None else set(site_ids) + sites = await MasterDao.list_site( + query_db, + SitePageQueryModel(status='0'), + False, + ) + options = [] + for site in sites: + site_id = cls._record_value(site, 'site_id') + if cls._record_value(site, 'status') != '0': + continue + if allowed_ids is not None and site_id not in allowed_ids: + continue + options.append( + BusinessOptionModel( + id=site_id, + code=cls._record_value(site, 'site_code'), + name=cls._record_value(site, 'site_name'), + ) + ) + return sorted(options, key=lambda option: option.id) + + @staticmethod + def _validate_scan_pallet( + pallet: BizPallet, + inventory: BizPalletInventory, + filling: BizFillingRecord, + source_warehouse_id: int, + ) -> None: + if pallet.asset_status != '0': + raise ServiceException(message='托盘资产已停用，不允许发货') + if ( + inventory.flow_state != PalletFlowState.FILLED_IN_WAREHOUSE.value + or inventory.warehouse_id != source_warehouse_id + or inventory.site_id is not None + or inventory.current_line_id is not None + or inventory.active_filling_id is None + or type(inventory.remaining_packages) is not int + or inventory.remaining_packages <= 0 + or inventory.remaining_packages > pallet.capacity_packages + ): + raise ServiceException(message='托盘不符合整托发货条件') + if ( + filling.filling_id != inventory.active_filling_id + or filling.pallet_id != pallet.pallet_id + or filling.warehouse_id != source_warehouse_id + or filling.active_slot != 'ACTIVE' + ): + raise ServiceException(message='托盘活动装填周期无效') + + @classmethod + async def scan_shipment_pallet_services( + cls, + query_db: AsyncSession, + qr_token: str, + query_object: ShipmentScanQueryModel, + current_user: CurrentUserModel, + ) -> PalletScanResponseModel: + await AccessService.assert_warehouse_access( + query_db, + current_user, + query_object.source_warehouse_id, + ) + warehouse = await MasterDao.get_warehouse(query_db, query_object.source_warehouse_id) + if warehouse is None: + raise ServiceException(message='来源仓库不存在') + if warehouse.status != '0': + raise ServiceException(message='来源仓库已停用') + await AccessService.assert_site_access( + query_db, + current_user, + query_object.target_site_id, + ) + site = await MasterDao.get_site(query_db, query_object.target_site_id) + if site is None: + raise ServiceException(message='目标现场不存在') + if site.status != '0': + raise ServiceException(message='目标现场已停用') + + pallet = await InventoryDao.get_pallet_by_qr_token(query_db, qr_token) + if pallet is None: + cls._raise_pallet_not_accessible(current_user) + inventory = await MasterDao.get_pallet_inventory(query_db, pallet.pallet_id) + if inventory is None: + cls._raise_pallet_not_accessible(current_user, '托盘库存不存在') + if inventory.pallet_id != pallet.pallet_id or inventory.warehouse_id != query_object.source_warehouse_id: + cls._raise_pallet_not_accessible(current_user, '托盘不在来源仓库') + if inventory.site_id is not None or inventory.current_line_id is not None: + cls._raise_pallet_not_accessible(current_user, '托盘已在客户现场') + if inventory.active_filling_id is None: + raise ServiceException(message='托盘不存在活动装填周期') + filling = await OperationDao.get_filling_by_id( + query_db, + inventory.active_filling_id, + warehouse_ids=[query_object.source_warehouse_id], + ) + if filling is None: + raise ServiceException(message='托盘活动装填周期不存在') + cls._validate_scan_pallet( + pallet, + inventory, + filling, + query_object.source_warehouse_id, + ) + return PalletScanResponseModel( + pallet_id=pallet.pallet_id, + pallet_code=pallet.pallet_code, + flow_state=PalletFlowState.FILLED_IN_WAREHOUSE, + warehouse_id=inventory.warehouse_id, + site_id=None, + current_line_id=None, + active_filling_id=inventory.active_filling_id, + strain_id=filling.strain_id, + strain_code=filling.strain_code_snapshot, + strain_name=filling.strain_name_snapshot, + batch_no=filling.batch_no, + package_spec=filling.package_spec, + capacity_packages=pallet.capacity_packages, + remaining_packages=inventory.remaining_packages, + allowed_actions=['shipment'], + ) + + @classmethod + async def add_shipment_services( + cls, + query_db: AsyncSession, + page_object: ShipmentCreateModel, + current_user: CurrentUserModel, + ) -> CrudResponseModel: + try: + _, user_name = cls._current_identity(current_user) + await cls._validate_active_route( + query_db, + page_object.source_warehouse_id, + page_object.target_site_id, + current_user, + ) + ordered_ids = cls._validate_pallet_ids(page_object.pallet_ids) + await cls._lock_and_validate_pallets( + query_db, + ordered_ids, + page_object.source_warehouse_id, + ) + + current_time = datetime.now() + shipment = BizShipmentOrder( + shipment_no=f'SH-{secrets.token_hex(16)}', + source_warehouse_id=page_object.source_warehouse_id, + target_site_id=page_object.target_site_id, + status=DocumentStatus.DRAFT.value, + create_by=user_name, + create_time=current_time, + update_by=user_name, + update_time=current_time, + ) + shipment = await OperationDao.add_shipment(query_db, shipment) + await OperationDao.replace_shipment_items( + query_db, + shipment.shipment_id, + cls._build_items(shipment.shipment_id, ordered_ids, user_name, current_time), + ) + await query_db.commit() + return cls._success('新增成功') + except IntegrityError: + await query_db.rollback() + raise ServiceException(message='新增发货单失败，发货单号或明细已存在') from None + except Exception: + await query_db.rollback() + raise + + @classmethod + async def edit_shipment_services( + cls, + query_db: AsyncSession, + page_object: ShipmentUpdateModel, + current_user: CurrentUserModel, + ) -> CrudResponseModel: + try: + _, user_name = cls._current_identity(current_user) + warehouse_ids, site_ids = await cls._scopes(query_db, current_user) + shipment = await OperationDao.get_shipment_for_update( + query_db, + page_object.shipment_id, + warehouse_ids=warehouse_ids, + site_ids=site_ids, + ) + if shipment is None: + cls._raise_shipment_not_accessible(current_user) + if shipment.status != DocumentStatus.DRAFT.value: + raise ServiceException(message='已确认发货单不允许编辑') + if any(item.active_filling_id is not None or item.package_qty is not None for item in shipment.items): + raise ServiceException(message='发货草稿明细快照异常') + + await cls._validate_active_route( + query_db, + page_object.source_warehouse_id, + page_object.target_site_id, + current_user, + ) + ordered_ids = cls._validate_pallet_ids(page_object.pallet_ids) + await cls._lock_and_validate_pallets( + query_db, + ordered_ids, + page_object.source_warehouse_id, + ) + + current_time = datetime.now() + shipment.source_warehouse_id = page_object.source_warehouse_id + shipment.target_site_id = page_object.target_site_id + shipment.update_by = user_name + shipment.update_time = current_time + await OperationDao.update_shipment(query_db, shipment) + await OperationDao.replace_shipment_items( + query_db, + shipment.shipment_id, + cls._build_items(shipment.shipment_id, ordered_ids, user_name, current_time), + ) + await query_db.commit() + return cls._success('修改成功') + except IntegrityError: + await query_db.rollback() + raise ServiceException(message='修改发货单失败，单据或明细状态已变化') from None + except Exception: + await query_db.rollback() + raise + + @classmethod + async def delete_shipment_services( + cls, + query_db: AsyncSession, + shipment_ids: list[int], + current_user: CurrentUserModel, + ) -> CrudResponseModel: + try: + if not shipment_ids or any( + type(shipment_id) is not int or shipment_id <= 0 for shipment_id in shipment_ids + ): + raise ServiceException(message='发货单ID必须为正整数') + if len(shipment_ids) != len(set(shipment_ids)): + raise ServiceException(message='发货单ID不能重复') + + ordered_ids = sorted(shipment_ids) + warehouse_ids, site_ids = await cls._scopes(query_db, current_user) + for shipment_id in ordered_ids: + shipment = await OperationDao.get_shipment_for_update( + query_db, + shipment_id, + warehouse_ids=warehouse_ids, + site_ids=site_ids, + ) + if shipment is None: + cls._raise_shipment_not_accessible(current_user) + if shipment.status != DocumentStatus.DRAFT.value: + raise ServiceException(message='已确认发货单不允许删除') + + for shipment_id in ordered_ids: + await OperationDao.delete_shipment(query_db, shipment_id) + await query_db.commit() + return cls._success('删除成功') + except IntegrityError: + await query_db.rollback() + raise ServiceException(message='删除发货单失败，单据状态或关联数据已变化') from None + except Exception: + await query_db.rollback() + raise + + @staticmethod + def _is_exact_confirmed( + shipment: BizShipmentOrder, + shipment_id: int, + idempotency_key: str, + request_hash: str, + ) -> bool: + return ( + shipment.shipment_id == shipment_id + and shipment.status == DocumentStatus.CONFIRMED.value + and shipment.confirm_idempotency_key == idempotency_key + and shipment.confirm_request_hash == request_hash + ) + + @classmethod + async def _lock_and_reauthorize_route( + cls, + query_db: AsyncSession, + shipment: BizShipmentOrder, + current_user: CurrentUserModel, + ) -> None: + warehouse = await MasterDao.get_warehouse( + query_db, + shipment.source_warehouse_id, + for_update=True, + ) + 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( + cls, + query_db: AsyncSession, + shipment_id: int, + idempotency_key: str, + request_hash: str, + current_user: CurrentUserModel, + ) -> CrudResponseModel: + try: + warehouse_ids, site_ids = await cls._scopes(query_db, current_user) + recovered = await OperationDao.get_shipment_by_confirm_idempotency_key( + query_db, + idempotency_key, + warehouse_ids=warehouse_ids, + site_ids=site_ids, + ) + if recovered is None: + raise ServiceException(message='确认发货失败，幂等冲突') + if not cls._is_exact_confirmed( + recovered, + shipment_id, + idempotency_key, + request_hash, + ): + raise ServiceException(message='确认发货失败，幂等键已被其他请求使用') + await cls._validate_active_route( + query_db, + recovered.source_warehouse_id, + recovered.target_site_id, + current_user, + ) + return cls._success('确认成功') + finally: + await query_db.rollback() + + @classmethod + async def confirm_shipment_services( # noqa: PLR0912, PLR0915 + cls, + query_db: AsyncSession, + shipment_id: int, + page_object: ConfirmCommandModel, + current_user: CurrentUserModel, + ) -> CrudResponseModel: + try: + user_id, user_name = cls._current_identity(current_user) + warehouse_ids, site_ids = await cls._scopes(query_db, current_user) + shipment = await OperationDao.get_shipment_by_id( + query_db, + shipment_id, + warehouse_ids=warehouse_ids, + site_ids=site_ids, + ) + if shipment is None: + cls._raise_shipment_not_accessible(current_user) + + request_hash = cls._build_confirm_request_hash(shipment_id) + idempotency_key = page_object.idempotency_key + existing = await OperationDao.get_shipment_by_confirm_idempotency_key( + query_db, + idempotency_key, + warehouse_ids=warehouse_ids, + site_ids=site_ids, + ) + if existing is not None and not cls._is_exact_confirmed( + existing, + shipment_id, + idempotency_key, + request_hash, + ): + raise ServiceException(message='确认发货失败，幂等键已被其他请求使用') + + locked_shipment = await OperationDao.get_shipment_for_update( + query_db, + shipment_id, + warehouse_ids=warehouse_ids, + site_ids=site_ids, + ) + if locked_shipment is None: + cls._raise_shipment_not_accessible(current_user) + if locked_shipment.status == DocumentStatus.CONFIRMED.value: + if cls._is_exact_confirmed( + locked_shipment, + shipment_id, + idempotency_key, + request_hash, + ): + await cls._lock_and_reauthorize_route(query_db, locked_shipment, current_user) + await query_db.rollback() + return cls._success('确认成功') + raise ServiceException(message='发货单已确认，幂等信息不匹配') + if locked_shipment.status != DocumentStatus.DRAFT.value: + raise ServiceException(message='当前发货单状态不允许确认') + if existing is not None: + raise ServiceException(message='确认发货失败，并发状态不一致') + if locked_shipment.confirm_idempotency_key is not None or locked_shipment.confirm_request_hash is not None: + raise ServiceException(message='发货单确认信息不完整') + + items = list(locked_shipment.items) + pallet_ids = [item.pallet_id for item in items] + ordered_ids = cls._validate_pallet_ids(pallet_ids) + if any(item.active_filling_id is not None or item.package_qty is not None for item in items): + raise ServiceException(message='发货草稿明细快照异常') + inventories, _, _ = await cls._lock_and_validate_pallets( + query_db, + ordered_ids, + locked_shipment.source_warehouse_id, + ) + await cls._lock_and_reauthorize_route(query_db, locked_shipment, current_user) + + for inventory in inventories: + PalletStateMachine.ensure_transition( + PalletFlowState(inventory.flow_state), + PalletFlowState.AT_SITE_UNUSED, + ) + + inventory_by_id = {inventory.pallet_id: inventory for inventory in inventories} + current_time = datetime.now() + locked_shipment.status = DocumentStatus.CONFIRMED.value + locked_shipment.confirm_idempotency_key = idempotency_key + locked_shipment.confirm_request_hash = request_hash + locked_shipment.confirmed_by = user_id + locked_shipment.confirmed_at = current_time + locked_shipment.update_by = user_name + locked_shipment.update_time = current_time + for item in items: + inventory = inventory_by_id[item.pallet_id] + item.active_filling_id = inventory.active_filling_id + item.package_qty = inventory.remaining_packages + await OperationDao.update_shipment(query_db, locked_shipment) + + for inventory in inventories: + before_state = inventory.flow_state + before_warehouse_id = inventory.warehouse_id + before_site_id = inventory.site_id + before_line_id = inventory.current_line_id + before_remaining_packages = inventory.remaining_packages + + inventory.flow_state = PalletFlowState.AT_SITE_UNUSED.value + inventory.warehouse_id = None + inventory.site_id = locked_shipment.target_site_id + inventory.current_line_id = None + inventory.version += 1 + inventory.updated_at = current_time + event = BizPalletEvent( + pallet_id=inventory.pallet_id, + event_type=PalletEventType.SHIPMENT_CONFIRMED.value, + source_type='SHIPMENT', + source_id=shipment_id, + before_state=before_state, + after_state=PalletFlowState.AT_SITE_UNUSED.value, + before_warehouse_id=before_warehouse_id, + before_site_id=before_site_id, + before_line_id=before_line_id, + after_warehouse_id=None, + after_site_id=locked_shipment.target_site_id, + after_line_id=None, + before_remaining_packages=before_remaining_packages, + after_remaining_packages=inventory.remaining_packages, + operator_id=user_id, + occurred_at=current_time, + idempotency_key=idempotency_key, + request_hash=request_hash, + ) + await InventoryDao.add_event(query_db, inventory, event) + + await query_db.commit() + return cls._success('确认成功') + except IntegrityError: + await query_db.rollback() + return await cls._recover_confirm_integrity_error( + query_db, + shipment_id, + page_object.idempotency_key, + cls._build_confirm_request_hash(shipment_id), + current_user, + ) + except Exception: + await query_db.rollback() + raise 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 a354af0..a7faf72 100644 --- a/shierfeng-fastapi-backend/tests/module_business/test_access_service.py +++ b/shierfeng-fastapi-backend/tests/module_business/test_access_service.py @@ -3,6 +3,7 @@ import unittest from types import SimpleNamespace from unittest.mock import AsyncMock, MagicMock, patch +from sqlalchemy.dialects import mysql, postgresql from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.orm import InstrumentedAttribute from sqlalchemy.sql import Select, operators @@ -88,6 +89,33 @@ class AccessServiceTest(unittest.IsolatedAsyncioTestCase): user_has_site.assert_awaited_once_with(db, 8, 7) + @patch('module_business.service.access_service.AccessDao.user_has_site', new_callable=AsyncMock) + async def test_locking_site_access_is_forwarded(self, user_has_site: AsyncMock) -> None: + parameter = inspect.signature(AccessService.assert_site_access).parameters.get('for_update') + self.assertIsNotNone(parameter) + self.assertIs(inspect.Parameter.KEYWORD_ONLY, parameter.kind) + self.assertIs(False, parameter.default) + user_has_site.return_value = True + user = SimpleNamespace(user=SimpleNamespace(user_id=8, admin=False)) + db = AsyncMock(spec=AsyncSession) + + await AccessService.assert_site_access(db, user, 7, for_update=True) + + user_has_site.assert_awaited_once_with(db, 8, 7, for_update=True) + + @patch('module_business.service.access_service.AccessDao.user_has_site', new_callable=AsyncMock) + async def test_admin_bypasses_locking_site_mapping(self, user_has_site: AsyncMock) -> None: + user = SimpleNamespace(user=SimpleNamespace(user_id=1, admin=True)) + + await AccessService.assert_site_access( + AsyncMock(spec=AsyncSession), + user, + 7, + for_update=True, + ) + + user_has_site.assert_not_awaited() + @patch('module_business.service.access_service.AccessDao.user_has_site', new_callable=AsyncMock) async def test_unmapped_site_is_forbidden(self, user_has_site: AsyncMock) -> None: user_has_site.return_value = False @@ -176,8 +204,11 @@ class AccessDaoTest(unittest.IsolatedAsyncioTestCase): 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) + for dialect in (mysql.dialect(), postgresql.dialect()): + sql = ' '.join( + str(statement.compile(dialect=dialect, 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)): @@ -200,6 +231,30 @@ class AccessDaoTest(unittest.IsolatedAsyncioTestCase): ), ) + async def test_user_has_site_current_read_is_optional_keyword_only_lock(self) -> None: + parameter = inspect.signature(AccessDao.user_has_site).parameters.get('for_update') + self.assertIsNotNone(parameter) + self.assertIs(inspect.Parameter.KEYWORD_ONLY, parameter.kind) + self.assertIs(False, 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_site(db, 8, 7, for_update=True) + else: + await AccessDao.user_has_site(db, 8, 7) + + statement = db.execute.await_args.args[0] + for dialect in (mysql.dialect(), postgresql.dialect()): + sql = ' '.join( + str(statement.compile(dialect=dialect, compile_kwargs={'literal_binds': True})).split() + ) + self.assertIs(for_update, 'FOR UPDATE' in sql) + async def test_list_user_warehouse_ids_returns_scalar_ids(self) -> None: result = MagicMock() result.scalars.return_value.all.return_value = (11, 13) 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 5f1aa1a..d899d26 100644 --- a/shierfeng-fastapi-backend/tests/module_business/test_inventory_dao.py +++ b/shierfeng-fastapi-backend/tests/module_business/test_inventory_dao.py @@ -122,11 +122,16 @@ 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', 'warehouse_ids'), - 'get_shipment_for_update': ('db', 'shipment_id'), + 'get_shipment_for_update': ('db', 'shipment_id', 'warehouse_ids', 'site_ids'), 'get_return_for_update': ('db', 'return_id'), 'get_active_filling_for_update': ('db', 'filling_id'), 'get_inbound_by_confirm_idempotency_key': ('db', 'idempotency_key'), - 'get_shipment_by_confirm_idempotency_key': ('db', 'idempotency_key'), + 'get_shipment_by_confirm_idempotency_key': ( + 'db', + 'idempotency_key', + 'warehouse_ids', + 'site_ids', + ), 'get_filling_by_idempotency_key': ('db', 'idempotency_key', 'warehouse_ids'), 'get_usage_by_idempotency_key': ('db', 'idempotency_key'), 'get_return_by_request_idempotency_key': ('db', 'idempotency_key'), @@ -139,7 +144,7 @@ class TestDaoPublicContract(unittest.TestCase): 'get_filling_by_id': ('db', 'filling_id', 'warehouse_ids'), 'add_filling_record': ('db', 'filling'), 'update_filling_record': ('db', 'filling'), - 'get_shipment_by_id': ('db', 'shipment_id'), + 'get_shipment_by_id': ('db', 'shipment_id', 'warehouse_ids', 'site_ids'), 'add_shipment': ('db', 'shipment'), 'update_shipment': ('db', 'shipment'), 'delete_shipment': ('db', 'shipment_id'), @@ -177,6 +182,16 @@ class TestDaoPublicContract(unittest.TestCase): if warehouse_ids is not None: self.assertIsNone(warehouse_ids.default) + for method_name in ( + 'get_shipment_for_update', + 'get_shipment_by_confirm_idempotency_key', + 'get_shipment_by_id', + ): + with self.subTest(method=method_name): + signature = inspect.signature(getattr(OperationDao, method_name)) + self.assertIsNone(signature.parameters['warehouse_ids'].default) + self.assertIsNone(signature.parameters['site_ids'].default) + filling_list_signature = inspect.signature(OperationDao.get_filling_list) self.assertEqual( ('db', 'query_object', 'is_page', 'warehouse_ids'), @@ -185,7 +200,16 @@ class TestDaoPublicContract(unittest.TestCase): self.assertIs(True, filling_list_signature.parameters['is_page'].default) self.assertIsNone(filling_list_signature.parameters['warehouse_ids'].default) - for method_name in ('get_shipment_list', 'get_usage_list', 'get_return_list'): + shipment_list_signature = inspect.signature(OperationDao.get_shipment_list) + self.assertEqual( + ('db', 'query_object', 'is_page', 'warehouse_ids', 'site_ids'), + tuple(shipment_list_signature.parameters), + ) + self.assertIs(True, shipment_list_signature.parameters['is_page'].default) + self.assertIsNone(shipment_list_signature.parameters['warehouse_ids'].default) + self.assertIsNone(shipment_list_signature.parameters['site_ids'].default) + + for method_name in ('get_usage_list', 'get_return_list'): with self.subTest(method=method_name): signature = inspect.signature(getattr(OperationDao, method_name)) self.assertEqual(('db', 'query_object', 'is_page'), tuple(signature.parameters)) @@ -383,13 +407,6 @@ class TestOperationDaoReads(unittest.IsolatedAsyncioTestCase): async def test_for_update_queries_target_primary_keys_and_lock_rows(self) -> None: cases = ( - ( - OperationDao.get_shipment_for_update, - BizShipmentOrder, - 'shipment_id', - 12, - (), - ), ( OperationDao.get_return_for_update, BizReturnOrder, @@ -426,6 +443,69 @@ class TestOperationDaoReads(unittest.IsolatedAsyncioTestCase): for fragment in extra_fragments: self.assertIn(fragment, sql) + async def test_shipment_for_update_locks_scoped_parent_then_current_items(self) -> None: + shipment = SimpleNamespace(shipment_id=12) + items = [ + SimpleNamespace(shipment_item_id=1, shipment_id=12, pallet_id=5), + SimpleNamespace(shipment_item_id=3, shipment_id=12, pallet_id=2), + ] + parent_result = MagicMock() + parent_result.scalar_one_or_none.return_value = shipment + item_result = MagicMock() + item_result.scalars.return_value.all.return_value = items + self.db.execute.side_effect = [parent_result, item_result] + + actual = await OperationDao.get_shipment_for_update( + self.db, + 12, + warehouse_ids=[7, 3, 7], + site_ids=[11, 9, 11], + ) + + self.assertIs(shipment, actual) + self.assertEqual([2, 5], [item.pallet_id for item in actual.items]) + self.assertEqual(2, self.db.execute.await_count) + parent_statement = self.db.execute.await_args_list[0].args[0] + item_statement = self.db.execute.await_args_list[1].args[0] + self.assertTrue(parent_statement.get_final_froms()[0].compare(BizShipmentOrder.__table__)) + self.assertTrue(item_statement.get_final_froms()[0].compare(BizShipmentItem.__table__)) + self.assertIs(True, parent_statement.get_execution_options().get('populate_existing')) + self.assertIs(True, item_statement.get_execution_options().get('populate_existing')) + for dialect in (mysql.dialect(), postgresql.dialect()): + with self.subTest(dialect=dialect.name): + parent_sql = compile_sql(parent_statement, dialect) + item_sql = compile_sql(item_statement, dialect) + self.assertIn('biz_shipment_order.shipment_id = 12', parent_sql) + self.assertIn('biz_shipment_order.source_warehouse_id IN (3, 7)', parent_sql) + self.assertIn('biz_shipment_order.target_site_id IN (9, 11)', parent_sql) + self.assertIn('FOR UPDATE', parent_sql) + self.assertIn('biz_shipment_item.shipment_id = 12', item_sql) + self.assertIn( + 'ORDER BY biz_shipment_item.pallet_id, biz_shipment_item.shipment_item_id', + item_sql, + ) + self.assertIn('FOR UPDATE', item_sql) + + async def test_shipment_for_update_missing_scoped_parent_skips_item_lock(self) -> None: + parent_result = MagicMock() + parent_result.scalar_one_or_none.return_value = None + self.db.execute.return_value = parent_result + + actual = await OperationDao.get_shipment_for_update( + self.db, + 12, + warehouse_ids=[], + site_ids=[9], + ) + + 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_inbound_for_update_locks_parent_then_current_items(self) -> None: inbound = SimpleNamespace(inbound_id=11) items = [ @@ -716,6 +796,74 @@ class TestOperationDaoReads(unittest.IsolatedAsyncioTestCase): self.assertNotIn('warehouse_id IN', sql) self.assertNotIn('WHERE false', sql) + async def test_shipment_list_applies_both_scopes_and_empty_scope_is_false(self) -> None: + query = ShipmentPageQueryModel(page_num=2, page_size=7) + marker = object() + with patch.object(PageUtil, 'paginate', new=AsyncMock(return_value=marker)) as paginate: + actual = await OperationDao.get_shipment_list( + self.db, + query, + is_page=True, + warehouse_ids=[7, 3, 7], + site_ids=[11, 9, 11], + ) + + self.assertIs(marker, actual) + statement = paginate.await_args.args[1] + for dialect in (mysql.dialect(), postgresql.dialect()): + with self.subTest(dialect=dialect.name): + sql = compile_sql(statement, dialect) + self.assertIn('biz_shipment_order.source_warehouse_id IN (3, 7)', sql) + self.assertIn('biz_shipment_order.target_site_id IN (9, 11)', sql) + + with patch.object(PageUtil, 'paginate', new=AsyncMock(return_value=marker)) as empty_paginate: + await OperationDao.get_shipment_list( + self.db, + query, + warehouse_ids=[7], + site_ids=[], + ) + empty_statement = empty_paginate.await_args.args[1] + self.assertIn('false = 1', compile_sql(empty_statement, mysql.dialect())) + self.assertIn('false', compile_sql(empty_statement, postgresql.dialect())) + + async def test_shipment_detail_and_confirm_key_push_both_scopes_into_parent_sql(self) -> None: + parent_result = MagicMock() + parent_result.scalar_one_or_none.return_value = None + self.db.execute.return_value = parent_result + + self.assertIsNone( + await OperationDao.get_shipment_by_id( + self.db, + 12, + warehouse_ids=[7, 3], + site_ids=[11, 9], + ) + ) + self.db.execute.assert_awaited_once() + detail_statement = self.db.execute.await_args.args[0] + + self.db.reset_mock() + self.db.execute.return_value = parent_result + self.assertIsNone( + await OperationDao.get_shipment_by_confirm_idempotency_key( + self.db, + 'confirm-key', + warehouse_ids=[7, 3], + site_ids=[11, 9], + ) + ) + key_statement = self.db.execute.await_args.args[0] + for dialect in (mysql.dialect(), postgresql.dialect()): + with self.subTest(dialect=dialect.name): + detail_sql = compile_sql(detail_statement, dialect) + key_sql = compile_sql(key_statement, dialect) + for sql in (detail_sql, key_sql): + self.assertIn('biz_shipment_order.source_warehouse_id IN (3, 7)', sql) + self.assertIn('biz_shipment_order.target_site_id IN (9, 11)', sql) + self.assertIn("biz_shipment_order.confirm_idempotency_key = 'confirm-key'", key_sql) + self.assertNotIn('FOR UPDATE', key_sql) + def test_time_conditions_include_fractional_last_second_with_exclusive_next_day(self) -> None: lower, upper = OperationDao._time_conditions( BizUsageRecord.occurred_at, diff --git a/shierfeng-fastapi-backend/tests/module_business/test_shipment_service.py b/shierfeng-fastapi-backend/tests/module_business/test_shipment_service.py new file mode 100644 index 0000000..a4414ba --- /dev/null +++ b/shierfeng-fastapi-backend/tests/module_business/test_shipment_service.py @@ -0,0 +1,1526 @@ +import hashlib +import importlib.util +import inspect +import json +import unittest +from datetime import datetime +from types import SimpleNamespace +from typing import Any, get_type_hints +from unittest.mock import AsyncMock, MagicMock, patch + +from fastapi import FastAPI +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.vo import ( + CrudResponseModel, + DataResponseModel, + PageModel, + PageResponseModel, + ResponseBaseModel, +) +from exceptions.exception import PermissionException, ServiceException +from module_admin.entity.vo.user_vo import CurrentUserModel, UserInfoModel +from module_admin.service.login_service import LoginService +from module_business.controller import shipment_controller as shipment_controller_module +from module_business.dao.access_dao import AccessDao +from module_business.dao.inventory_dao import InventoryDao +from module_business.dao.master_dao import MasterDao +from module_business.dao.operation_dao import OperationDao +from module_business.domain.enums import PalletEventType, PalletFlowState +from module_business.domain.pallet_state_machine import PalletStateMachine +from module_business.entity.do.master_do import BizCustomerSite, BizPallet, BizWarehouse +from module_business.entity.do.operation_do import ( + BizFillingRecord, + BizPalletEvent, + BizPalletInventory, + BizShipmentItem, + BizShipmentOrder, +) +from module_business.entity.vo.operation_vo import ( + BusinessOptionModel, + ConfirmCommandModel, + PalletScanResponseModel, + ShipmentCreateModel, + ShipmentModel, + ShipmentPageQueryModel, + ShipmentScanQueryModel, + ShipmentUpdateModel, +) +from module_business.service import shipment_service as shipment_service_module +from module_business.service.access_service import AccessService +from module_business.service.shipment_service import ShipmentService + + +def make_user(user_id: int = 7) -> CurrentUserModel: + return CurrentUserModel( + permissions=[], + roles=[], + user=UserInfoModel(userId=user_id, userName='admin' if user_id == 1 else f'user-{user_id}'), + ) + + +def integrity_error() -> IntegrityError: + return IntegrityError('statement', {}, RuntimeError('constraint')) + + +def make_pallet(pallet_id: int, *, capacity: int = 20, asset_status: str = '0') -> BizPallet: + return BizPallet( + pallet_id=pallet_id, + pallet_code=f'P-{pallet_id:03d}', + qr_token=f'qr-{pallet_id}', + capacity_packages=capacity, + asset_status=asset_status, + ) + + +def make_inventory( + pallet_id: int, + *, + warehouse_id: int = 7, + site_id: int | None = None, + current_line_id: int | None = None, + active_filling_id: int | None = None, + remaining_packages: int = 8, + flow_state: str = 'FILLED_IN_WAREHOUSE', + version: int = 3, +) -> BizPalletInventory: + return BizPalletInventory( + pallet_id=pallet_id, + warehouse_id=warehouse_id, + site_id=site_id, + current_line_id=current_line_id, + active_filling_id=active_filling_id if active_filling_id is not None else 100 + pallet_id, + remaining_packages=remaining_packages, + flow_state=flow_state, + version=version, + ) + + +def make_filling(pallet_id: int, *, warehouse_id: int = 7) -> BizFillingRecord: + return BizFillingRecord( + filling_id=100 + pallet_id, + filling_no=f'FI-{pallet_id}', + pallet_id=pallet_id, + warehouse_id=warehouse_id, + strain_id=3, + strain_code_snapshot='S-03', + strain_name_snapshot='菌种三', + batch_no='B-202607', + package_spec='袋装', + initial_packages=20, + active_slot='ACTIVE', + idempotency_key=f'fill-{pallet_id}', + request_hash='f' * 64, + operator_id=7, + operated_at=SimpleNamespace(), + ) + + +def make_shipment( + shipment_id: int = 12, + *, + source_warehouse_id: int = 7, + target_site_id: int = 9, + pallet_ids: tuple[int, ...] = (5, 2), + status: str = 'DRAFT', + idempotency_key: str | None = None, + request_hash: str | None = None, +) -> BizShipmentOrder: + shipment = BizShipmentOrder( + shipment_id=shipment_id, + shipment_no=f'SH-{shipment_id}', + source_warehouse_id=source_warehouse_id, + target_site_id=target_site_id, + status=status, + confirm_idempotency_key=idempotency_key, + confirm_request_hash=request_hash, + ) + shipment.items = [ + BizShipmentItem( + shipment_item_id=index, + shipment_id=shipment_id, + pallet_id=pallet_id, + ) + for index, pallet_id in enumerate(pallet_ids, start=1) + ] + return shipment + + +class ShipmentServiceBootstrapTest(unittest.TestCase): + def test_shipment_service_module_exists(self) -> None: + spec = importlib.util.find_spec('module_business.service.shipment_service') + + self.assertIsNotNone(spec, 'ShipmentService 尚未实现') + + def test_confirm_signature_and_types_are_exact(self) -> None: + self.assertTrue( + hasattr(ShipmentService, 'confirm_shipment_services'), + 'confirm_shipment_services 尚未实现', + ) + signature = inspect.signature(ShipmentService.confirm_shipment_services) + self.assertEqual( + ('query_db', 'shipment_id', 'page_object', 'current_user'), + tuple(signature.parameters), + ) + hints = get_type_hints(ShipmentService.confirm_shipment_services) + self.assertIs(AsyncSession, hints['query_db']) + self.assertIs(int, hints['shipment_id']) + self.assertIs(ConfirmCommandModel, hints['page_object']) + self.assertIs(CurrentUserModel, hints['current_user']) + self.assertIs(CrudResponseModel, hints['return']) + + def test_confirm_hash_is_canonical_and_only_contains_shipment_id(self) -> None: + self.assertTrue( + hasattr(ShipmentService, '_build_confirm_request_hash'), + '_build_confirm_request_hash 尚未实现', + ) + expected = hashlib.sha256( + json.dumps( + {'shipment_id': 12}, + sort_keys=True, + separators=(',', ':'), + ).encode() + ).hexdigest() + + self.assertEqual(expected, ShipmentService._build_confirm_request_hash(12)) + self.assertNotEqual(expected, ShipmentService._build_confirm_request_hash(13)) + + def test_service_exposes_complete_shipment_workflow(self) -> None: + expected = { + 'get_shipment_list_services', + 'shipment_detail_services', + 'get_warehouse_options_services', + 'get_site_options_services', + 'scan_shipment_pallet_services', + 'add_shipment_services', + 'edit_shipment_services', + 'delete_shipment_services', + 'confirm_shipment_services', + } + + self.assertEqual( + expected, + {name for name in expected if hasattr(ShipmentService, name)}, + ) + + +class ShipmentInfrastructureContractTest(unittest.TestCase): + def test_shipment_dao_reads_accept_both_data_scopes(self) -> None: + expected = { + OperationDao.get_shipment_list: ( + 'db', + 'query_object', + 'is_page', + 'warehouse_ids', + 'site_ids', + ), + OperationDao.get_shipment_by_id: ( + 'db', + 'shipment_id', + 'warehouse_ids', + 'site_ids', + ), + OperationDao.get_shipment_for_update: ( + 'db', + 'shipment_id', + 'warehouse_ids', + 'site_ids', + ), + OperationDao.get_shipment_by_confirm_idempotency_key: ( + 'db', + 'idempotency_key', + 'warehouse_ids', + 'site_ids', + ), + } + + for method, parameter_names in expected.items(): + with self.subTest(method=method.__name__): + self.assertEqual(parameter_names, tuple(inspect.signature(method).parameters)) + + def test_site_access_supports_optional_current_read(self) -> None: + for method in (AccessDao.user_has_site, AccessService.assert_site_access): + with self.subTest(method=method.__qualname__): + parameter = inspect.signature(method).parameters.get('for_update') + self.assertIsNotNone(parameter) + self.assertIs(inspect.Parameter.KEYWORD_ONLY, parameter.kind) + self.assertIs(False, parameter.default) + + +class ShipmentControllerBootstrapTest(unittest.TestCase): + def test_shipment_controller_module_exists(self) -> None: + spec = importlib.util.find_spec('module_business.controller.shipment_controller') + + self.assertIsNotNone(spec, 'shipment_controller 尚未实现') + + +class ShipmentControllerContractTest(unittest.TestCase): + @staticmethod + def _routes() -> dict[tuple[str, str], APIRoute]: + return { + (route.path, next(iter(route.methods))): route + for route in shipment_controller_module.shipment_controller.routes + if isinstance(route, APIRoute) + } + + def test_openapi_has_nine_operations_aliases_and_response_models(self) -> None: + app = FastAPI() + app.include_router(shipment_controller_module.shipment_controller) + schema = app.openapi() + expected = { + ('/business/shipment/list', 'get'), + ('/business/shipment/warehouse-options', 'get'), + ('/business/shipment/site-options', 'get'), + ('/business/shipment/scan/{qr_token}', 'get'), + ('/business/shipment/{shipment_id}', 'get'), + ('/business/shipment', 'post'), + ('/business/shipment', 'put'), + ('/business/shipment/{shipment_ids}', 'delete'), + ('/business/shipment/{shipment_id}/confirm', 'put'), + } + self.assertEqual(expected, {(path, method) for path, methods in schema['paths'].items() for method in methods}) + scan_parameters = schema['paths']['/business/shipment/scan/{qr_token}']['get']['parameters'] + self.assertEqual( + {'qr_token', 'sourceWarehouseId', 'targetSiteId'}, + {parameter['name'] for parameter in scan_parameters}, + ) + + routes = self._routes() + self.assertEqual(PageResponseModel[ShipmentModel], routes[('/business/shipment/list', 'GET')].response_model) + for path in ('warehouse-options', 'site-options'): + self.assertEqual( + DataResponseModel[list[BusinessOptionModel]], + routes[(f'/business/shipment/{path}', 'GET')].response_model, + ) + self.assertEqual( + DataResponseModel[PalletScanResponseModel], + routes[('/business/shipment/scan/{qr_token}', 'GET')].response_model, + ) + self.assertEqual( + DataResponseModel[ShipmentModel], + routes[('/business/shipment/{shipment_id}', 'GET')].response_model, + ) + for key in ( + ('/business/shipment', 'POST'), + ('/business/shipment', 'PUT'), + ('/business/shipment/{shipment_ids}', 'DELETE'), + ('/business/shipment/{shipment_id}/confirm', 'PUT'), + ): + self.assertEqual(ResponseBaseModel, routes[key].response_model) + + def test_static_options_and_scan_precede_dynamic_detail(self) -> None: + paths = [route.path for route in shipment_controller_module.shipment_controller.routes] + detail_index = paths.index('/business/shipment/{shipment_id}') + for static_path in ( + '/business/shipment/warehouse-options', + '/business/shipment/site-options', + '/business/shipment/scan/{qr_token}', + ): + self.assertLess(paths.index(static_path), detail_index) + + def test_routes_have_exact_permissions_current_user_and_write_logs(self) -> None: + expected = { + ('/business/shipment/list', 'GET'): 'business:shipment:list', + ('/business/shipment/warehouse-options', 'GET'): 'business:shipment:query', + ('/business/shipment/site-options', 'GET'): 'business:shipment:query', + ('/business/shipment/scan/{qr_token}', 'GET'): 'business:shipment:query', + ('/business/shipment/{shipment_id}', 'GET'): 'business:shipment:query', + ('/business/shipment', 'POST'): 'business:shipment:add', + ('/business/shipment', 'PUT'): 'business:shipment:edit', + ('/business/shipment/{shipment_ids}', 'DELETE'): 'business:shipment:remove', + ('/business/shipment/{shipment_id}/confirm', 'PUT'): 'business:shipment:confirm', + } + actual = {} + for key, route in self._routes().items(): + permissions = [ + dependency.dependency + for dependency in route.dependencies + if isinstance(dependency.dependency, CheckUserInterfaceAuth) + ] + self.assertEqual(1, len(permissions), key) + self.assertTrue(any(isinstance(dependency.dependency, PreAuth) for dependency in route.dependencies)) + self.assertTrue( + any( + dependency.name == 'current_user' and dependency.call == LoginService.get_current_user + for dependency in route.dependant.dependencies + ), + key, + ) + if key[1] in {'POST', 'PUT', 'DELETE'}: + self.assertTrue(hasattr(route.endpoint, '__wrapped__'), key) + actual[key] = permissions[0].perm + + self.assertEqual(expected, actual) + self.assertEqual(6, len(set(actual.values()))) + + def test_controller_defines_no_scan_or_export_permission(self) -> None: + source = inspect.getsource(shipment_controller_module) + self.assertNotIn('business:shipment:scan', source) + self.assertNotIn('business:shipment:export', source) + + +class ShipmentReadServiceTest(unittest.IsolatedAsyncioTestCase): + def setUp(self) -> None: + self.db = MagicMock(spec=AsyncSession) + self.user = make_user() + + async def test_empty_either_scope_returns_empty_page_without_global_query(self) -> None: + query = ShipmentPageQueryModel(pageNum=2, pageSize=7) + get_list = 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=[])), + patch.object(OperationDao, 'get_shipment_list', new=get_list), + ): + try: + result = await ShipmentService.get_shipment_list_services( + self.db, + query, + self.user, + is_page=True, + ) + except NotImplementedError: + self.fail('双范围空集处理尚未实现') + + self.assertIsInstance(result, PageModel) + self.assertEqual([], result.rows) + self.assertEqual(0, result.total) + self.assertEqual(2, result.page_num) + self.assertEqual(7, result.page_size) + self.assertFalse(result.has_next) + get_list.assert_not_awaited() + + async def test_list_and_detail_push_both_scopes_before_reading(self) -> None: + shipment = make_shipment() + page = PageModel(rows=[shipment], pageNum=1, pageSize=10, total=1, hasNext=False) + get_list = AsyncMock(return_value=page) + get_detail = AsyncMock(return_value=shipment) + with ( + patch.object(AccessDao, 'list_user_warehouse_ids', new=AsyncMock(return_value=[7, 7])), + patch.object(AccessDao, 'list_user_site_ids', new=AsyncMock(return_value=[9, 9])), + patch.object(OperationDao, 'get_shipment_list', new=get_list), + patch.object(OperationDao, 'get_shipment_by_id', new=get_detail), + ): + listed = await ShipmentService.get_shipment_list_services( + self.db, + ShipmentPageQueryModel(), + self.user, + ) + try: + detail = await ShipmentService.shipment_detail_services(self.db, 12, self.user) + except NotImplementedError: + self.fail('双范围单据详情尚未实现') + + self.assertEqual(1, listed.total) + self.assertEqual(12, detail.shipment_id) + get_list.assert_awaited_once_with( + self.db, + unittest.mock.ANY, + True, + warehouse_ids=[7], + site_ids=[9], + ) + get_detail.assert_awaited_once_with( + self.db, + 12, + warehouse_ids=[7], + site_ids=[9], + ) + + async def test_options_only_return_enabled_assigned_warehouse_and_site(self) -> None: + warehouses = [ + BizWarehouse(warehouse_id=7, warehouse_code='W-7', warehouse_name='七号仓', status='0'), + BizWarehouse(warehouse_id=8, warehouse_code='W-8', warehouse_name='八号仓', status='0'), + BizWarehouse(warehouse_id=10, warehouse_code='W-10', warehouse_name='停用仓', status='1'), + ] + sites = [ + BizCustomerSite(site_id=9, customer_id=1, site_code='S-9', site_name='九号现场', status='0'), + BizCustomerSite(site_id=11, customer_id=1, site_code='S-11', site_name='十一号现场', status='0'), + BizCustomerSite(site_id=12, customer_id=1, site_code='S-12', site_name='停用现场', status='1'), + ] + 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(MasterDao, 'list_warehouse', new=AsyncMock(return_value=warehouses)), + patch.object(MasterDao, 'list_site', new=AsyncMock(return_value=sites)), + ): + try: + warehouse_options = await ShipmentService.get_warehouse_options_services(self.db, self.user) + site_options = await ShipmentService.get_site_options_services(self.db, self.user) + except NotImplementedError: + self.fail('发货仓库/现场选项尚未实现') + + self.assertEqual([(7, 'W-7', '七号仓')], [(item.id, item.code, item.name) for item in warehouse_options]) + self.assertEqual([(9, 'S-9', '九号现场')], [(item.id, item.code, item.name) for item in site_options]) + + async def test_scan_is_read_only_and_returns_backend_remaining_packages(self) -> None: + query = ShipmentScanQueryModel(sourceWarehouseId=7, targetSiteId=9) + pallet = make_pallet(2) + inventory = make_inventory(2, remaining_packages=8) + filling = make_filling(2) + lock_inventory = AsyncMock() + lock_pallets = AsyncMock() + lock_filling = AsyncMock() + with ( + patch.object(AccessService, 'assert_warehouse_access', new=AsyncMock()), + 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', + ) + ), + ), + 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(InventoryDao, 'get_pallet_by_qr_token', new=AsyncMock(return_value=pallet)), + patch.object(MasterDao, 'get_pallet_inventory', new=AsyncMock(return_value=inventory)), + patch.object(OperationDao, 'get_filling_by_id', new=AsyncMock(return_value=filling)), + patch.object(InventoryDao, 'get_inventories_for_update', new=lock_inventory), + patch.object(InventoryDao, 'get_pallets_by_ids', new=lock_pallets), + patch.object(OperationDao, 'get_active_filling_for_update', new=lock_filling), + ): + try: + result = await ShipmentService.scan_shipment_pallet_services( + self.db, + 'qr-2', + query, + self.user, + ) + except NotImplementedError: + self.fail('整托发货扫码尚未实现') + + self.assertEqual(2, result.pallet_id) + self.assertEqual(PalletFlowState.FILLED_IN_WAREHOUSE, result.flow_state) + self.assertEqual(8, result.remaining_packages) + self.assertEqual(['shipment'], result.allowed_actions) + lock_inventory.assert_not_awaited() + lock_pallets.assert_not_awaited() + lock_filling.assert_not_awaited() + self.db.commit.assert_not_awaited() + self.db.rollback.assert_not_awaited() + + async def test_explicit_filter_outside_either_scope_is_rejected_before_dao(self) -> None: + cases = ( + (ShipmentPageQueryModel(sourceWarehouseId=8), '无权访问该仓库'), + (ShipmentPageQueryModel(targetSiteId=10), '无权访问该客户现场'), + ) + for query, expected_message in cases: + with self.subTest(expected_message=expected_message): + get_list = 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_list', new=get_list), + self.assertRaises(PermissionException) as raised, + ): + await ShipmentService.get_shipment_list_services(self.db, query, self.user) + + self.assertEqual(expected_message, raised.exception.message) + get_list.assert_not_awaited() + + async def test_missing_detail_is_hidden_for_user_but_precise_for_admin(self) -> None: + get_detail = AsyncMock(return_value=None) + 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=get_detail), + self.assertRaises(PermissionException) as user_raised, + ): + await ShipmentService.shipment_detail_services(self.db, 404, self.user) + + self.assertEqual('无权访问该发货单', user_raised.exception.message) + get_detail.assert_awaited_once_with( + self.db, + 404, + warehouse_ids=[7], + site_ids=[9], + ) + + get_detail.reset_mock() + with ( + patch.object(OperationDao, 'get_shipment_by_id', new=get_detail), + self.assertRaises(ServiceException) as admin_raised, + ): + await ShipmentService.shipment_detail_services(self.db, 404, make_user(1)) + self.assertEqual('发货单不存在', admin_raised.exception.message) + get_detail.assert_awaited_once_with( + self.db, + 404, + warehouse_ids=None, + site_ids=None, + ) + + async def test_scan_hides_missing_qr_external_warehouse_and_site_location(self) -> None: + route_mocks = ( + patch.object(AccessService, 'assert_warehouse_access', new=AsyncMock()), + 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', + ) + ), + ), + patch.object( + MasterDao, + 'get_site', + new=AsyncMock( + return_value=BizCustomerSite( + site_id=9, + customer_id=1, + site_code='S-9', + site_name='九号现场', + status='0', + ) + ), + ), + ) + scenarios = ( + (None, None), + (make_pallet(2), make_inventory(2, warehouse_id=8)), + (make_pallet(2), make_inventory(2, site_id=11)), + ) + for pallet, inventory in scenarios: + with self.subTest(pallet=pallet, inventory=inventory): + with ( + route_mocks[0], + route_mocks[1], + route_mocks[2], + route_mocks[3], + patch.object(InventoryDao, 'get_pallet_by_qr_token', new=AsyncMock(return_value=pallet)), + patch.object(MasterDao, 'get_pallet_inventory', new=AsyncMock(return_value=inventory)), + self.assertRaises(PermissionException) as raised, + ): + await ShipmentService.scan_shipment_pallet_services( + self.db, + 'hidden-token', + ShipmentScanQueryModel(sourceWarehouseId=7, targetSiteId=9), + self.user, + ) + self.assertEqual('无权访问该托盘', raised.exception.message) + + +class ShipmentDraftServiceTest(unittest.IsolatedAsyncioTestCase): + def setUp(self) -> None: + self.db = MagicMock(spec=AsyncSession) + self.user = make_user() + self.fixed_time = datetime(2026, 7, 26, 16, 30) + + async def test_add_only_persists_pallet_ids_with_null_snapshots(self) -> None: # noqa: PLR0915 + page = ShipmentCreateModel(sourceWarehouseId=7, targetSiteId=9, palletIds=[5, 2]) + inventories = [make_inventory(2), make_inventory(5)] + pallets = [make_pallet(2), make_pallet(5)] + fillings = {102: make_filling(2), 105: make_filling(5)} + captured: dict[str, Any] = {} + actions: list[str] = [] + + async def warehouse_access(*_args: Any, **_kwargs: Any) -> None: + actions.append('warehouse-access') + + async def site_access(*_args: Any, **_kwargs: Any) -> None: + actions.append('site-access') + + async def get_warehouse(*_args: Any, **_kwargs: Any) -> BizWarehouse: + actions.append('warehouse-read') + return BizWarehouse( + warehouse_id=7, + warehouse_code='W-7', + warehouse_name='七号仓', + status='0', + ) + + async def get_site(*_args: Any, **_kwargs: Any) -> BizCustomerSite: + actions.append('site-read') + return BizCustomerSite( + site_id=9, + customer_id=1, + site_code='S-9', + site_name='九号现场', + status='0', + ) + + async def lock_inventories(*_args: Any) -> list[BizPalletInventory]: + actions.append('inventory-lock') + return inventories + + async def lock_pallets(*_args: Any) -> list[BizPallet]: + actions.append('pallet-lock') + return pallets + + async def lock_filling(_db: AsyncSession, filling_id: int) -> BizFillingRecord: + actions.append(f'filling-lock-{filling_id}') + return fillings[filling_id] + + async def add_order(_db: AsyncSession, shipment: BizShipmentOrder) -> BizShipmentOrder: + actions.append('add-order') + shipment.shipment_id = 12 + captured['shipment'] = shipment + return shipment + + async def replace_items( + _db: AsyncSession, + shipment_id: int, + items: list[BizShipmentItem], + ) -> list[BizShipmentItem]: + actions.append('replace-items') + captured['items'] = items + self.assertEqual(12, shipment_id) + return items + + async def commit() -> None: + actions.append('commit') + + self.db.commit.side_effect = commit + add_event = AsyncMock() + with ( + 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( + InventoryDao, + 'get_inventories_for_update', + new=AsyncMock(side_effect=lock_inventories), + ) as inventory_lock, + patch.object(InventoryDao, 'get_pallets_by_ids', new=AsyncMock(side_effect=lock_pallets)) as pallet_lock, + patch.object( + OperationDao, + 'get_active_filling_for_update', + new=AsyncMock(side_effect=lock_filling), + ), + patch.object(OperationDao, 'add_shipment', new=AsyncMock(side_effect=add_order)), + patch.object(OperationDao, 'replace_shipment_items', new=AsyncMock(side_effect=replace_items)), + patch.object(InventoryDao, 'add_event', new=add_event), + patch.object(shipment_service_module, 'datetime', create=True) as service_datetime, + patch.object(shipment_service_module, 'secrets', create=True) as service_secrets, + ): + service_datetime.now.return_value = self.fixed_time + service_secrets.token_hex.return_value = 'draft-token' + try: + result = await ShipmentService.add_shipment_services(self.db, page, self.user) + except NotImplementedError: + self.fail('发货草稿新增尚未实现') + + self.assertTrue(result.is_success) + self.assertEqual('DRAFT', captured['shipment'].status) + self.assertEqual('SH-draft-token', captured['shipment'].shipment_no) + self.assertEqual([2, 5], [item.pallet_id for item in captured['items']]) + self.assertTrue(all(item.active_filling_id is None for item in captured['items'])) + self.assertTrue(all(item.package_qty is None for item in captured['items'])) + inventory_lock.assert_awaited_once_with(self.db, [2, 5]) + pallet_lock.assert_awaited_once_with(self.db, [2, 5]) + self.assertEqual( + [ + 'warehouse-access', + 'warehouse-read', + 'site-access', + 'site-read', + 'inventory-lock', + 'pallet-lock', + 'filling-lock-102', + 'filling-lock-105', + 'add-order', + 'replace-items', + 'commit', + ], + actions, + ) + self.assertEqual(['FILLED_IN_WAREHOUSE', 'FILLED_IN_WAREHOUSE'], [item.flow_state for item in inventories]) + add_event.assert_not_awaited() + self.db.commit.assert_awaited_once() + self.db.rollback.assert_not_awaited() + + async def test_edit_revalidates_current_draft_and_keeps_snapshots_null(self) -> None: + page = ShipmentUpdateModel( + shipmentId=12, + sourceWarehouseId=8, + targetSiteId=10, + palletIds=[3], + ) + shipment = make_shipment() + inventory = make_inventory(3, warehouse_id=8) + pallet = make_pallet(3) + filling = make_filling(3, warehouse_id=8) + captured: dict[str, Any] = {} + + async def update_order(_db: AsyncSession, updated: BizShipmentOrder) -> BizShipmentOrder: + captured['shipment'] = updated + return updated + + async def replace_items( + _db: AsyncSession, + _shipment_id: int, + items: list[BizShipmentItem], + ) -> list[BizShipmentItem]: + captured['items'] = items + return items + + 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), + ) as lock_order, + patch.object(AccessService, 'assert_warehouse_access', new=AsyncMock()), + patch.object(AccessService, 'assert_site_access', new=AsyncMock()), + patch.object( + MasterDao, + 'get_warehouse', + new=AsyncMock( + return_value=BizWarehouse( + warehouse_id=8, + warehouse_code='W-8', + warehouse_name='八号仓', + status='0', + ) + ), + ), + patch.object( + MasterDao, + 'get_site', + new=AsyncMock( + return_value=BizCustomerSite( + site_id=10, + customer_id=1, + site_code='S-10', + site_name='十号现场', + status='0', + ) + ), + ), + patch.object(InventoryDao, 'get_inventories_for_update', new=AsyncMock(return_value=[inventory])), + patch.object(InventoryDao, 'get_pallets_by_ids', new=AsyncMock(return_value=[pallet])), + patch.object( + OperationDao, + 'get_active_filling_for_update', + new=AsyncMock(return_value=filling), + ), + patch.object(OperationDao, 'update_shipment', new=AsyncMock(side_effect=update_order)), + patch.object(OperationDao, 'replace_shipment_items', new=AsyncMock(side_effect=replace_items)), + patch.object(shipment_service_module, 'datetime') as service_datetime, + ): + service_datetime.now.return_value = self.fixed_time + try: + result = await ShipmentService.edit_shipment_services(self.db, page, self.user) + except NotImplementedError: + self.fail('发货草稿编辑尚未实现') + + self.assertTrue(result.is_success) + lock_order.assert_awaited_once_with( + self.db, + 12, + warehouse_ids=[7, 8], + site_ids=[9, 10], + ) + self.assertEqual(8, captured['shipment'].source_warehouse_id) + self.assertEqual(10, captured['shipment'].target_site_id) + self.assertEqual([3], [item.pallet_id for item in captured['items']]) + self.assertIsNone(captured['items'][0].active_filling_id) + self.assertIsNone(captured['items'][0].package_qty) + self.assertEqual('FILLED_IN_WAREHOUSE', inventory.flow_state) + self.db.commit.assert_awaited_once() + + async def test_batch_delete_preflights_all_drafts_before_any_delete(self) -> None: + documents = { + 11: make_shipment(11, status='DRAFT'), + 12: make_shipment(12, status='CONFIRMED'), + } + + async def lock_order( + _db: AsyncSession, + shipment_id: int, + **_kwargs: Any, + ) -> BizShipmentOrder: + return documents[shipment_id] + + delete = 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_for_update', new=AsyncMock(side_effect=lock_order)) as lock, + patch.object(OperationDao, 'delete_shipment', new=delete), + ): + try: + with self.assertRaises(ServiceException): + await ShipmentService.delete_shipment_services(self.db, [12, 11], self.user) + except NotImplementedError: + self.fail('发货草稿原子批量删除尚未实现') + + self.assertEqual([11, 12], [call.args[1] for call in lock.await_args_list]) + delete.assert_not_awaited() + self.db.commit.assert_not_awaited() + self.db.rollback.assert_awaited_once() + + +class ShipmentConfirmServiceTest(unittest.IsolatedAsyncioTestCase): + def setUp(self) -> None: + self.db = MagicMock(spec=AsyncSession) + self.user = make_user() + self.command = ConfirmCommandModel(idempotencyKey='confirm-key') + self.fixed_time = datetime(2026, 7, 26, 17, 20) + self.request_hash = ShipmentService._build_confirm_request_hash(12) + + async def test_confirm_snapshots_full_remaining_and_moves_every_pallet_atomically( # noqa: PLR0915 + self, + ) -> None: + shipment = make_shipment() + inventories = [ + make_inventory(2, remaining_packages=8, version=4), + make_inventory(5, remaining_packages=13, version=8), + ] + pallets = [make_pallet(2), make_pallet(5)] + fillings = {102: make_filling(2), 105: make_filling(5)} + actions: list[str] = [] + events: list[BizPalletEvent] = [] + + async def detail(*_args: Any, **_kwargs: Any) -> BizShipmentOrder: + actions.append('detail') + return shipment + + async def idempotency(*_args: Any, **_kwargs: Any) -> None: + actions.append('idempotency-read') + + async def order_lock(*_args: Any, **_kwargs: Any) -> BizShipmentOrder: + actions.append('order-lock') + return shipment + + async def inventory_lock(*_args: Any) -> list[BizPalletInventory]: + actions.append('inventory-lock') + return inventories + + async def pallet_lock(*_args: Any) -> list[BizPallet]: + actions.append('pallet-lock') + return pallets + + async def filling_lock(_db: AsyncSession, filling_id: int) -> BizFillingRecord: + actions.append(f'filling-lock-{filling_id}') + return fillings[filling_id] + + async def warehouse_lock(*_args: Any, **kwargs: Any) -> BizWarehouse: + self.assertTrue(kwargs.get('for_update')) + actions.append('warehouse-lock') + return BizWarehouse( + warehouse_id=7, + warehouse_code='W-7', + warehouse_name='七号仓', + status='0', + ) + + async def warehouse_mapping(*_args: Any, **kwargs: Any) -> None: + self.assertTrue(kwargs.get('for_update')) + actions.append('warehouse-mapping-lock') + + async def site_lock(*_args: Any, **kwargs: Any) -> BizCustomerSite: + self.assertTrue(kwargs.get('for_update')) + actions.append('site-lock') + return BizCustomerSite( + site_id=9, + customer_id=1, + site_code='S-9', + site_name='九号现场', + status='0', + ) + + async def site_mapping(*_args: Any, **kwargs: Any) -> None: + self.assertTrue(kwargs.get('for_update')) + actions.append('site-mapping-lock') + + def transition(current: PalletFlowState, target: PalletFlowState) -> None: + actions.append(f'transition-{current.value}-{target.value}') + + async def update(*_args: Any) -> BizShipmentOrder: + actions.append('update-order') + return shipment + + async def add_event( + _db: AsyncSession, + inventory: BizPalletInventory, + event: BizPalletEvent, + ) -> BizPalletEvent: + actions.append(f'event-{inventory.pallet_id}') + event.event_id = 200 + inventory.pallet_id + inventory.last_event_id = event.event_id + events.append(event) + return event + + async def commit() -> None: + actions.append('commit') + + self.db.commit.side_effect = commit + 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(side_effect=detail)) as detail_read, + patch.object( + OperationDao, + 'get_shipment_by_confirm_idempotency_key', + new=AsyncMock(side_effect=idempotency), + ), + patch.object(OperationDao, 'get_shipment_for_update', new=AsyncMock(side_effect=order_lock)), + patch.object( + InventoryDao, + 'get_inventories_for_update', + new=AsyncMock(side_effect=inventory_lock), + ), + patch.object(InventoryDao, 'get_pallets_by_ids', new=AsyncMock(side_effect=pallet_lock)), + patch.object( + OperationDao, + 'get_active_filling_for_update', + new=AsyncMock(side_effect=filling_lock), + ), + patch.object(MasterDao, 'get_warehouse', new=AsyncMock(side_effect=warehouse_lock)), + patch.object( + AccessService, + 'assert_warehouse_access', + new=AsyncMock(side_effect=warehouse_mapping), + ), + patch.object(MasterDao, 'get_site', new=AsyncMock(side_effect=site_lock)), + patch.object(AccessService, 'assert_site_access', new=AsyncMock(side_effect=site_mapping)), + patch.object(PalletStateMachine, 'ensure_transition', side_effect=transition), + patch.object(OperationDao, 'update_shipment', new=AsyncMock(side_effect=update)), + patch.object(InventoryDao, 'add_event', new=AsyncMock(side_effect=add_event)), + patch.object(shipment_service_module, 'datetime') as service_datetime, + ): + service_datetime.now.return_value = self.fixed_time + try: + result = await ShipmentService.confirm_shipment_services( + self.db, + 12, + self.command, + self.user, + ) + except NotImplementedError: + self.fail('多托盘整托确认尚未实现') + + self.assertTrue(result.is_success) + self.assertEqual( + [ + 'detail', + 'idempotency-read', + 'order-lock', + 'inventory-lock', + 'pallet-lock', + 'filling-lock-102', + 'filling-lock-105', + 'warehouse-lock', + 'warehouse-mapping-lock', + 'site-lock', + 'site-mapping-lock', + 'transition-FILLED_IN_WAREHOUSE-AT_SITE_UNUSED', + 'transition-FILLED_IN_WAREHOUSE-AT_SITE_UNUSED', + 'update-order', + 'event-2', + 'event-5', + 'commit', + ], + actions, + ) + detail_read.assert_awaited_once_with( + self.db, + 12, + warehouse_ids=[7], + site_ids=[9], + ) + self.assertEqual('CONFIRMED', shipment.status) + self.assertEqual('confirm-key', shipment.confirm_idempotency_key) + self.assertEqual(self.request_hash, shipment.confirm_request_hash) + self.assertEqual(7, shipment.confirmed_by) + self.assertEqual(self.fixed_time, shipment.confirmed_at) + snapshots = {item.pallet_id: (item.active_filling_id, item.package_qty) for item in shipment.items} + self.assertEqual({2: (102, 8), 5: (105, 13)}, snapshots) + for inventory, original_version, remaining in zip( + inventories, + (4, 8), + (8, 13), + strict=True, + ): + self.assertEqual('AT_SITE_UNUSED', inventory.flow_state) + self.assertIsNone(inventory.warehouse_id) + self.assertEqual(9, inventory.site_id) + self.assertIsNone(inventory.current_line_id) + self.assertEqual(100 + inventory.pallet_id, inventory.active_filling_id) + self.assertEqual(remaining, inventory.remaining_packages) + self.assertEqual(original_version + 1, inventory.version) + self.assertEqual(self.fixed_time, inventory.updated_at) + self.assertEqual([2, 5], [event.pallet_id for event in events]) + for event in events: + self.assertEqual(PalletEventType.SHIPMENT_CONFIRMED.value, event.event_type) + self.assertEqual('SHIPMENT', event.source_type) + self.assertEqual(12, event.source_id) + self.assertEqual('FILLED_IN_WAREHOUSE', event.before_state) + self.assertEqual('AT_SITE_UNUSED', event.after_state) + self.assertEqual(7, event.before_warehouse_id) + self.assertIsNone(event.before_site_id) + self.assertIsNone(event.after_warehouse_id) + self.assertEqual(9, event.after_site_id) + self.assertEqual(event.before_remaining_packages, event.after_remaining_packages) + self.assertEqual(7, event.operator_id) + self.assertEqual(self.fixed_time, event.occurred_at) + self.assertEqual('confirm-key', event.idempotency_key) + self.assertEqual(self.request_hash, event.request_hash) + self.db.commit.assert_awaited_once() + self.db.rollback.assert_not_awaited() + + async def test_rr_current_parent_recovers_exact_confirm_without_repeating_inventory_writes(self) -> None: + initial = make_shipment(status='DRAFT') + confirmed = make_shipment( + status='CONFIRMED', + idempotency_key='confirm-key', + request_hash=self.request_hash, + ) + 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=initial)), + patch.object( + OperationDao, + 'get_shipment_by_confirm_idempotency_key', + new=AsyncMock(return_value=None), + ), + 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='0', + ) + ), + ) 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='0', + ) + ), + ) 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( + self.db, + 12, + self.command, + self.user, + ) + + self.assertTrue(result.is_success) + warehouse_lock.assert_awaited_once_with(self.db, 7, for_update=True) + warehouse_access.assert_awaited_once_with(self.db, self.user, 7, for_update=True) + site_lock.assert_awaited_once_with(self.db, 9, for_update=True) + site_access.assert_awaited_once_with(self.db, self.user, 9, for_update=True) + 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,)) + inventory = make_inventory(2, remaining_packages=6) + lock_validate = AsyncMock(return_value=([inventory], [make_pallet(2)], [make_filling(2)])) + 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=initial)), + patch.object( + OperationDao, + 'get_shipment_by_confirm_idempotency_key', + new=AsyncMock(return_value=None), + ), + patch.object(OperationDao, 'get_shipment_for_update', new=AsyncMock(return_value=locked)), + patch.object(ShipmentService, '_lock_and_validate_pallets', new=lock_validate), + patch.object( + MasterDao, + 'get_warehouse', + new=AsyncMock( + return_value=BizWarehouse( + warehouse_id=7, + warehouse_code='W-7', + warehouse_name='七号仓', + status='0', + ) + ), + ), + 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(AccessService, 'assert_warehouse_access', new=AsyncMock()), + patch.object(AccessService, 'assert_site_access', new=AsyncMock()), + patch.object(PalletStateMachine, 'ensure_transition'), + patch.object(OperationDao, 'update_shipment', new=AsyncMock()), + patch.object(InventoryDao, 'add_event', new=AsyncMock()), + ): + result = await ShipmentService.confirm_shipment_services( + self.db, + 12, + self.command, + self.user, + ) + + self.assertTrue(result.is_success) + lock_validate.assert_awaited_once_with(self.db, [2], 7) + self.assertEqual((102, 6), (locked.items[0].active_filling_id, locked.items[0].package_qty)) + self.assertIsNone(initial.items[0].active_filling_id) + self.assertIsNone(initial.items[0].package_qty) + + async def test_same_key_on_other_shipment_is_rejected_before_parent_lock(self) -> None: + target = make_shipment() + other = make_shipment( + 13, + status='CONFIRMED', + idempotency_key='confirm-key', + request_hash=ShipmentService._build_confirm_request_hash(13), + ) + parent_lock = 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=target)), + patch.object( + OperationDao, + 'get_shipment_by_confirm_idempotency_key', + new=AsyncMock(return_value=other), + ), + patch.object(OperationDao, 'get_shipment_for_update', new=parent_lock), + self.assertRaises(ServiceException) as raised, + ): + await ShipmentService.confirm_shipment_services(self.db, 12, self.command, self.user) + + self.assertEqual('确认发货失败，幂等键已被其他请求使用', raised.exception.message) + parent_lock.assert_not_awaited() + self.db.rollback.assert_awaited_once() + + async def test_integrity_recovery_nonexact_result_is_stable_conflict_not_success(self) -> None: + initial = make_shipment() + nonexact = make_shipment( + 13, + status='CONFIRMED', + idempotency_key='confirm-key', + request_hash=ShipmentService._build_confirm_request_hash(13), + ) + 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, nonexact]), + ), + 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) + + self.assertEqual('确认发货失败，幂等键已被其他请求使用', raised.exception.message) + self.assertEqual(2, self.db.rollback.await_count) + self.db.commit.assert_not_awaited() + + 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) + update = AsyncMock() + add_event = AsyncMock() + mapping = 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=shipment)), + patch.object( + OperationDao, + 'get_shipment_by_confirm_idempotency_key', + new=AsyncMock(return_value=None), + ), + patch.object(OperationDao, 'get_shipment_for_update', new=AsyncMock(return_value=shipment)), + patch.object( + ShipmentService, + '_lock_and_validate_pallets', + new=AsyncMock(return_value=([inventory], [make_pallet(2)], [make_filling(2)])), + ), + patch.object( + MasterDao, + 'get_warehouse', + new=AsyncMock( + return_value=BizWarehouse( + warehouse_id=7, + warehouse_code='W-7', + warehouse_name='停用仓', + status='1', + ) + ), + ), + patch.object(AccessService, 'assert_warehouse_access', new=mapping), + patch.object(OperationDao, 'update_shipment', new=update), + patch.object(InventoryDao, 'add_event', new=add_event), + self.assertRaises(ServiceException) as raised, + ): + await ShipmentService.confirm_shipment_services(self.db, 12, self.command, self.user) + + self.assertEqual('来源仓库已停用', raised.exception.message) + self.assertEqual('DRAFT', shipment.status) + self.assertEqual('FILLED_IN_WAREHOUSE', inventory.flow_state) + self.assertEqual(4, inventory.version) + mapping.assert_not_awaited() + update.assert_not_awaited() + add_event.assert_not_awaited() + self.db.rollback.assert_awaited_once() + + async def test_integrity_error_recovers_only_fresh_scoped_exact_result(self) -> None: + initial = make_shipment(status='DRAFT') + recovered = make_shipment( + status='CONFIRMED', + idempotency_key='confirm-key', + request_hash=self.request_hash, + ) + warehouse_scope = AsyncMock(side_effect=[[7], [7]]) + site_scope = AsyncMock(side_effect=[[9], [9]]) + key_read = AsyncMock(side_effect=[None, recovered]) + with ( + patch.object(AccessDao, 'list_user_warehouse_ids', new=warehouse_scope), + 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(AccessService, 'assert_warehouse_access', new=AsyncMock()), + 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', + ) + ), + ), + patch.object( + MasterDao, + 'get_site', + new=AsyncMock( + return_value=BizCustomerSite( + site_id=9, + customer_id=1, + site_code='S-9', + site_name='九号现场', + status='0', + ) + ), + ), + ): + result = await ShipmentService.confirm_shipment_services( + self.db, + 12, + self.command, + self.user, + ) + + self.assertTrue(result.is_success) + self.assertEqual(2, warehouse_scope.await_count) + self.assertEqual(2, site_scope.await_count) + self.assertEqual(2, key_read.await_count) + for call in key_read.await_args_list: + self.assertEqual([7], call.kwargs['warehouse_ids']) + self.assertEqual([9], call.kwargs['site_ids']) + self.assertEqual(2, self.db.rollback.await_count) + self.db.commit.assert_not_awaited() + + async def test_revoked_mapping_after_locks_hides_document_and_mutates_nothing(self) -> None: + shipment = make_shipment(pallet_ids=(2,)) + inventory = make_inventory(2, version=4) + update = AsyncMock() + add_event = AsyncMock() + transition = MagicMock() + site_lock = 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=shipment)), + patch.object( + OperationDao, + 'get_shipment_by_confirm_idempotency_key', + new=AsyncMock(return_value=None), + ), + patch.object(OperationDao, 'get_shipment_for_update', new=AsyncMock(return_value=shipment)), + patch.object(InventoryDao, 'get_inventories_for_update', new=AsyncMock(return_value=[inventory])), + patch.object(InventoryDao, 'get_pallets_by_ids', new=AsyncMock(return_value=[make_pallet(2)])), + patch.object( + OperationDao, + 'get_active_filling_for_update', + new=AsyncMock(return_value=make_filling(2)), + ), + patch.object( + MasterDao, + 'get_warehouse', + new=AsyncMock( + return_value=BizWarehouse( + warehouse_id=7, + warehouse_code='W-7', + warehouse_name='七号仓', + status='0', + ) + ), + ), + patch.object( + AccessService, + 'assert_warehouse_access', + new=AsyncMock(side_effect=PermissionException(data='', message='无权访问该仓库')), + ), + patch.object(MasterDao, 'get_site', new=site_lock), + patch.object(PalletStateMachine, 'ensure_transition', new=transition), + 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) + self.assertEqual('DRAFT', shipment.status) + self.assertEqual('FILLED_IN_WAREHOUSE', inventory.flow_state) + self.assertEqual(7, inventory.warehouse_id) + self.assertIsNone(inventory.site_id) + self.assertEqual(4, inventory.version) + site_lock.assert_not_awaited() + transition.assert_not_called() + update.assert_not_awaited() + add_event.assert_not_awaited() + self.db.commit.assert_not_awaited() + self.db.rollback.assert_awaited_once() + + async def test_all_transitions_validate_before_first_mutation(self) -> None: + shipment = make_shipment() + inventories = [make_inventory(2, version=4), make_inventory(5, version=8)] + fillings = {102: make_filling(2), 105: make_filling(5)} + update = AsyncMock() + add_event = AsyncMock() + + async def get_filling(_db: AsyncSession, filling_id: int) -> BizFillingRecord: + return fillings[filling_id] + + transition = MagicMock(side_effect=[None, ValueError('second pallet invalid')]) + 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=shipment)), + patch.object( + OperationDao, + 'get_shipment_by_confirm_idempotency_key', + new=AsyncMock(return_value=None), + ), + patch.object(OperationDao, 'get_shipment_for_update', new=AsyncMock(return_value=shipment)), + patch.object(InventoryDao, 'get_inventories_for_update', new=AsyncMock(return_value=inventories)), + patch.object( + InventoryDao, + 'get_pallets_by_ids', + new=AsyncMock(return_value=[make_pallet(2), make_pallet(5)]), + ), + patch.object( + OperationDao, + 'get_active_filling_for_update', + new=AsyncMock(side_effect=get_filling), + ), + patch.object( + MasterDao, + 'get_warehouse', + new=AsyncMock( + return_value=BizWarehouse( + warehouse_id=7, + warehouse_code='W-7', + warehouse_name='七号仓', + status='0', + ) + ), + ), + 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(AccessService, 'assert_warehouse_access', new=AsyncMock()), + patch.object(AccessService, 'assert_site_access', new=AsyncMock()), + patch.object(PalletStateMachine, 'ensure_transition', new=transition), + patch.object(OperationDao, 'update_shipment', new=update), + patch.object(InventoryDao, 'add_event', new=add_event), + self.assertRaisesRegex(ValueError, 'second pallet invalid'), + ): + await ShipmentService.confirm_shipment_services(self.db, 12, self.command, self.user) + + self.assertEqual('DRAFT', shipment.status) + self.assertTrue(all(item.active_filling_id is None and item.package_qty is None for item in shipment.items)) + self.assertEqual(['FILLED_IN_WAREHOUSE', 'FILLED_IN_WAREHOUSE'], [item.flow_state for item in inventories]) + self.assertEqual([4, 8], [item.version for item in inventories]) + update.assert_not_awaited() + add_event.assert_not_awaited() + self.db.commit.assert_not_awaited() + self.db.rollback.assert_awaited_once() + + +if __name__ == '__main__': + unittest.main()