diff --git a/shierfeng-fastapi-backend/module_business/controller/return_controller.py b/shierfeng-fastapi-backend/module_business/controller/return_controller.py new file mode 100644 index 0000000..df334b6 --- /dev/null +++ b/shierfeng-fastapi-backend/module_business/controller/return_controller.py @@ -0,0 +1,155 @@ +"""托盘两阶段退回接口。""" + +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, + PalletScanResponseModel, + ReturnConfirmModel, + ReturnModel, + ReturnPageQueryModel, + ReturnRequestModel, + ReturnScanQueryModel, +) +from module_business.service.return_service import ReturnService +from utils.response_util import ResponseUtil + +return_controller = APIRouterPro( + prefix='/business/return', + tags=['业务管理-托盘退回'], + dependencies=[PreAuthDependency()], +) + + +@return_controller.get( + '/list', + summary='获取托盘退回单分页列表接口', + description='按当前用户来源现场或目标仓库数据范围获取退回单', + response_model=PageResponseModel[ReturnModel], + dependencies=[UserInterfaceAuthDependency('business:return:list')], +) +async def get_business_return_list( + request: Request, + return_page_query: Annotated[ReturnPageQueryModel, Query()], + query_db: Annotated[AsyncSession, DBSessionDependency()], + current_user: Annotated[CurrentUserModel, CurrentUserDependency()], +) -> Response: + result = await ReturnService.get_return_list_services( + query_db, + return_page_query, + current_user, + is_page=True, + ) + return ResponseUtil.success(model_content=result) + + +@return_controller.get( + '/warehouse-options', + summary='获取退回目标仓库选项接口', + description='返回全部启用仓库作为客户发起退回的目标字典', + response_model=DataResponseModel[list[BusinessOptionModel]], + dependencies=[UserInterfaceAuthDependency('business:return:query')], +) +async def get_business_return_warehouse_options( + request: Request, + query_db: Annotated[AsyncSession, DBSessionDependency()], + current_user: Annotated[CurrentUserModel, CurrentUserDependency()], +) -> Response: + result = await ReturnService.get_warehouse_options_services(query_db, current_user) + return ResponseUtil.success(data=result) + + +@return_controller.get( + '/scan/{qr_token}', + summary='扫描退回托盘接口', + description='按申请现场或确认仓库数据范围重新查询托盘退回状态', + response_model=DataResponseModel[PalletScanResponseModel], + dependencies=[UserInterfaceAuthDependency('business:return:query')], +) +async def scan_business_return_pallet( + request: Request, + qr_token: Annotated[ + str, + Path(min_length=1, max_length=128, pattern=r'\S', description='二维码固定标识'), + ], + return_scan_query: Annotated[ReturnScanQueryModel, Query()], + query_db: Annotated[AsyncSession, DBSessionDependency()], + current_user: Annotated[CurrentUserModel, CurrentUserDependency()], +) -> Response: + result = await ReturnService.scan_return_pallet_services( + query_db, + qr_token, + return_scan_query, + current_user, + ) + return ResponseUtil.success(data=result) + + +@return_controller.post( + '/request', + summary='发起托盘退回申请接口', + description='原子冻结同一客户现场的全部退回托盘并保存库存快照', + response_model=ResponseBaseModel, + dependencies=[UserInterfaceAuthDependency('business:return:request')], +) +@Log(title='托盘退回', business_type=BusinessType.INSERT) +async def request_business_return( + request: Request, + request_command: ReturnRequestModel, + query_db: Annotated[AsyncSession, DBSessionDependency()], + current_user: Annotated[CurrentUserModel, CurrentUserDependency()], +) -> Response: + result = await ReturnService.request_return_services(query_db, request_command, current_user) + return ResponseUtil.success(msg=result.message) + + +@return_controller.get( + '/{return_id}', + summary='获取托盘退回单详情接口', + description='按来源现场或目标仓库数据范围获取指定退回单', + response_model=DataResponseModel[ReturnModel], + dependencies=[UserInterfaceAuthDependency('business:return:query')], +) +async def query_detail_business_return( + request: Request, + return_id: Annotated[int, Path(gt=0, description='退回单ID')], + query_db: Annotated[AsyncSession, DBSessionDependency()], + current_user: Annotated[CurrentUserModel, CurrentUserDependency()], +) -> Response: + result = await ReturnService.return_detail_services(query_db, return_id, current_user) + return ResponseUtil.success(data=result) + + +@return_controller.put( + '/{return_id}/confirm', + summary='确认托盘退回接口', + description='核对整单扫码集合和申请快照后原子移入目标仓库', + response_model=ResponseBaseModel, + dependencies=[UserInterfaceAuthDependency('business:return:confirm')], +) +@Log(title='托盘退回', business_type=BusinessType.UPDATE) +async def confirm_business_return( + request: Request, + return_id: Annotated[int, Path(gt=0, description='退回单ID')], + confirm_command: ReturnConfirmModel, + query_db: Annotated[AsyncSession, DBSessionDependency()], + current_user: Annotated[CurrentUserModel, CurrentUserDependency()], +) -> Response: + result = await ReturnService.confirm_return_services( + query_db, + return_id, + confirm_command, + current_user, + ) + return ResponseUtil.success(msg=result.message) diff --git a/shierfeng-fastapi-backend/module_business/dao/operation_dao.py b/shierfeng-fastapi-backend/module_business/dao/operation_dao.py index ef329b2..d3414fa 100644 --- a/shierfeng-fastapi-backend/module_business/dao/operation_dao.py +++ b/shierfeng-fastapi-backend/module_business/dao/operation_dao.py @@ -1,7 +1,7 @@ from datetime import datetime, time, timedelta from typing import Any -from sqlalchemy import ColumnElement, Select, delete, false, select, update +from sqlalchemy import ColumnElement, Select, delete, false, or_, select, update from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.orm.attributes import InstrumentedAttribute @@ -94,6 +94,27 @@ class OperationDao: conditions.append(BizShipmentOrder.target_site_id.in_(scoped_site_ids) if scoped_site_ids else false()) return conditions + @staticmethod + def _return_scope_conditions( + site_ids: list[int] | None, + warehouse_ids: list[int] | None, + ) -> list[ColumnElement[bool]]: + if site_ids is None and warehouse_ids is None: + return [] + + visibility_conditions: list[ColumnElement[bool]] = [] + if site_ids is not None: + scoped_site_ids = sorted(set(site_ids)) + visibility_conditions.append( + BizReturnOrder.source_site_id.in_(scoped_site_ids) if scoped_site_ids else false() + ) + if warehouse_ids is not None: + scoped_warehouse_ids = sorted(set(warehouse_ids)) + visibility_conditions.append( + BizReturnOrder.target_warehouse_id.in_(scoped_warehouse_ids) if scoped_warehouse_ids else false() + ) + return [or_(*visibility_conditions)] + @staticmethod async def _paginate( db: AsyncSession, @@ -300,13 +321,32 @@ class OperationDao: cls, db: AsyncSession, return_id: int, + site_ids: list[int] | None = None, + warehouse_ids: list[int] | None = None, ) -> BizReturnOrder | None: - return await cls._get_for_update( + return_order = await cls._get_for_update( db, BizReturnOrder, BizReturnOrder.return_id, return_id, + *cls._return_scope_conditions(site_ids, warehouse_ids), ) + if return_order is None: + return None + + # 父单锁定后使用 locking read 获取当前明细，避免 MySQL RR 沿用旧快照。 + item_result = await db.execute( + select(BizReturnItem) + .where(BizReturnItem.return_id == return_id) + .order_by(BizReturnItem.pallet_id, BizReturnItem.return_item_id) + .with_for_update() + .execution_options(populate_existing=True) + ) + return_order.items = sorted( + item_result.scalars().all(), + key=lambda item: (item.pallet_id, item.return_item_id), + ) + return return_order @classmethod async def get_active_filling_for_update( @@ -387,12 +427,15 @@ class OperationDao: cls, db: AsyncSession, idempotency_key: str, + site_ids: list[int] | None = None, + warehouse_ids: list[int] | None = None, ) -> BizReturnOrder | None: return await cls._get_by_key( db, BizReturnOrder, BizReturnOrder.request_idempotency_key, idempotency_key, + *cls._return_scope_conditions(site_ids, warehouse_ids), ) @classmethod @@ -400,12 +443,15 @@ class OperationDao: cls, db: AsyncSession, idempotency_key: str, + site_ids: list[int] | None = None, + warehouse_ids: list[int] | None = None, ) -> BizReturnOrder | None: return await cls._get_by_key( db, BizReturnOrder, BizReturnOrder.confirm_idempotency_key, idempotency_key, + *cls._return_scope_conditions(site_ids, warehouse_ids), ) @classmethod @@ -718,8 +764,10 @@ class OperationDao: db: AsyncSession, query_object: ReturnPageQueryModel, is_page: bool = True, + site_ids: list[int] | None = None, + warehouse_ids: list[int] | None = None, ) -> PageModel | list[dict[str, Any]]: - conditions = [] + conditions = cls._return_scope_conditions(site_ids, warehouse_ids) if query_object.return_no: conditions.append(BizReturnOrder.return_no.like(f'%{query_object.return_no}%')) if query_object.source_site_id: @@ -743,6 +791,8 @@ class OperationDao: cls, db: AsyncSession, return_id: int, + site_ids: list[int] | None = None, + warehouse_ids: list[int] | None = None, ) -> BizReturnOrder | None: return await cls._get_order_detail( db, @@ -752,6 +802,7 @@ class OperationDao: BizReturnItem, BizReturnItem.return_id, BizReturnItem.return_item_id, + *cls._return_scope_conditions(site_ids, warehouse_ids), ) @classmethod diff --git a/shierfeng-fastapi-backend/module_business/service/return_service.py b/shierfeng-fastapi-backend/module_business/service/return_service.py new file mode 100644 index 0000000..16cff8a --- /dev/null +++ b/shierfeng-fastapi-backend/module_business/service/return_service.py @@ -0,0 +1,1041 @@ +"""托盘两阶段退回服务。""" + +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 PalletEventType, PalletFlowState, ReturnStatus +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, + BizReturnItem, + BizReturnOrder, +) +from module_business.entity.vo.master_vo import WarehousePageQueryModel +from module_business.entity.vo.operation_vo import ( + BusinessOptionModel, + PalletScanResponseModel, + ReturnConfirmModel, + ReturnModel, + ReturnPageQueryModel, + ReturnRequestModel, + ReturnScanQueryModel, +) +from module_business.service.access_service import AccessService + + +class ReturnService: + """客户现场申请、仓库确认的两阶段退回服务层。""" + + _REQUESTABLE_STATES = frozenset( + { + PalletFlowState.AT_SITE_UNUSED.value, + PalletFlowState.AT_SITE_PARTIAL.value, + PalletFlowState.AT_SITE_EMPTY.value, + } + ) + + @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_return_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 _site_scope( + cls, + query_db: AsyncSession, + current_user: CurrentUserModel, + ) -> list[int] | None: + if cls._is_admin(current_user): + return None + user_id, _ = cls._current_identity(current_user) + return sorted(set(await AccessDao.list_user_site_ids(query_db, user_id))) + + @classmethod + async def _warehouse_scope( + cls, + query_db: AsyncSession, + current_user: CurrentUserModel, + ) -> list[int] | None: + if cls._is_admin(current_user): + return None + user_id, _ = cls._current_identity(current_user) + return sorted(set(await AccessDao.list_user_warehouse_ids(query_db, user_id))) + + @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) + site_ids = sorted(set(await AccessDao.list_user_site_ids(query_db, user_id))) + warehouse_ids = sorted(set(await AccessDao.list_user_warehouse_ids(query_db, user_id))) + return site_ids, warehouse_ids + + @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 _build_request_hash(page_object: ReturnRequestModel) -> str: + payload = json.dumps( + { + 'source_site_id': page_object.source_site_id, + 'target_warehouse_id': page_object.target_warehouse_id, + 'pallet_ids': sorted(page_object.pallet_ids), + }, + sort_keys=True, + separators=(',', ':'), + ).encode() + return hashlib.sha256(payload).hexdigest() + + @staticmethod + def _build_confirm_request_hash(return_id: int, page_object: ReturnConfirmModel) -> str: + payload = json.dumps( + { + 'return_id': return_id, + 'scanned_pallet_ids': sorted(page_object.scanned_pallet_ids), + }, + sort_keys=True, + separators=(',', ':'), + ).encode() + return hashlib.sha256(payload).hexdigest() + + @staticmethod + def _success(message: str, return_id: int) -> CrudResponseModel: + return CrudResponseModel(is_success=True, message=message, result=return_id) + + @staticmethod + def _is_exact_request_result( + return_order: BizReturnOrder, + page_object: ReturnRequestModel, + request_hash: str, + *, + require_items: bool = False, + ) -> bool: + if not ( + return_order.request_idempotency_key == page_object.idempotency_key + and return_order.request_hash == request_hash + and return_order.source_site_id == page_object.source_site_id + and return_order.target_warehouse_id == page_object.target_warehouse_id + and return_order.status in {ReturnStatus.REQUESTED.value, ReturnStatus.CONFIRMED.value} + ): + return False + if not require_items: + return True + items = getattr(return_order, 'items', None) + return items is not None and sorted(item.pallet_id for item in items) == sorted(page_object.pallet_ids) + + @staticmethod + def _is_exact_confirm_result( + return_order: BizReturnOrder, + return_id: int, + page_object: ReturnConfirmModel, + request_hash: str, + *, + require_items: bool = False, + ) -> bool: + if not ( + return_order.return_id == return_id + and return_order.status == ReturnStatus.CONFIRMED.value + and return_order.confirm_idempotency_key == page_object.idempotency_key + and return_order.confirm_request_hash == request_hash + ): + return False + if not require_items: + return True + items = getattr(return_order, 'items', None) + return items is not None and sorted(item.pallet_id for item in items) == sorted(page_object.scanned_pallet_ids) + + @classmethod + async def get_return_list_services( + cls, + query_db: AsyncSession, + query_object: ReturnPageQueryModel, + current_user: CurrentUserModel, + is_page: bool = True, + ) -> PageModel | list[ReturnModel]: + site_ids, warehouse_ids = await cls._scopes(query_db, current_user) + if site_ids == [] and warehouse_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_return_list( + query_db, + query_object, + is_page, + site_ids=site_ids, + warehouse_ids=warehouse_ids, + ) + rows = result.rows if isinstance(result, PageModel) else result + models = [ReturnModel.model_validate(record) for record in rows] + if isinstance(result, PageModel): + return result.model_copy(update={'rows': models}) + return models + + @classmethod + async def return_detail_services( + cls, + query_db: AsyncSession, + return_id: int, + current_user: CurrentUserModel, + ) -> ReturnModel: + site_ids, warehouse_ids = await cls._scopes(query_db, current_user) + return_order = await OperationDao.get_return_by_id( + query_db, + return_id, + site_ids=site_ids, + warehouse_ids=warehouse_ids, + ) + if return_order is None: + cls._raise_return_not_accessible(current_user) + return ReturnModel.model_validate(return_order) + + @classmethod + async def get_warehouse_options_services( + cls, + query_db: AsyncSession, + current_user: CurrentUserModel, + ) -> list[BusinessOptionModel]: + cls._current_identity(current_user) + warehouses = await MasterDao.list_warehouse( + query_db, + WarehousePageQueryModel(status='0'), + False, + ) + options = [ + BusinessOptionModel( + id=cls._record_value(warehouse, 'warehouse_id'), + code=cls._record_value(warehouse, 'warehouse_code'), + name=cls._record_value(warehouse, 'warehouse_name'), + ) + for warehouse in warehouses + if cls._record_value(warehouse, 'status') == '0' + ] + return sorted(options, key=lambda option: option.id) + + @classmethod + async def _assert_pallet_site_access( + cls, + query_db: AsyncSession, + current_user: CurrentUserModel, + site_id: int, + ) -> None: + try: + await AccessService.assert_site_access(query_db, current_user, site_id) + except PermissionException: + cls._raise_pallet_not_accessible(current_user) + + @staticmethod + def _validate_request_inventory( + pallet: BizPallet, + inventory: BizPalletInventory, + source_site_id: int, + ) -> None: + if pallet.pallet_id != inventory.pallet_id: + raise ServiceException(message='托盘资产与库存不一致') + if pallet.asset_status != '0': + raise ServiceException(message='托盘资产已停用，不允许退回') + if inventory.site_id != source_site_id: + raise ServiceException(message='同一退回单托盘必须属于同一现场') + if inventory.warehouse_id is not None or inventory.flow_state not in ReturnService._REQUESTABLE_STATES: + raise ServiceException(message='托盘当前状态不允许发起退回') + if ( + type(inventory.remaining_packages) is not int + or not 0 <= inventory.remaining_packages <= pallet.capacity_packages + ): + raise ServiceException(message='托盘剩余包数无效') + if type(inventory.version) is not int or inventory.version < 0: + raise ServiceException(message='托盘库存版本无效') + if inventory.flow_state == PalletFlowState.AT_SITE_EMPTY.value: + if inventory.remaining_packages != 0 or inventory.active_filling_id is not None: + raise ServiceException(message='空托盘库存与活动装填快照不一致') + elif inventory.remaining_packages <= 0 or inventory.active_filling_id is None: + raise ServiceException(message='带料托盘库存与活动装填快照不一致') + + @staticmethod + def _validate_active_filling( + filling: BizFillingRecord, + inventory: BizPalletInventory, + ) -> None: + if ( + filling.filling_id != inventory.active_filling_id + or filling.pallet_id != inventory.pallet_id + or filling.active_slot != 'ACTIVE' + or filling.closed_at is not None + ): + raise ServiceException(message='托盘活动装填周期无效') + + @classmethod + async def _scan_request_pallet( + cls, + query_db: AsyncSession, + qr_token: str, + current_user: CurrentUserModel, + ) -> PalletScanResponseModel: + 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 or inventory.site_id is None: + cls._raise_pallet_not_accessible(current_user, '托盘库存不存在或不在客户现场') + await cls._assert_pallet_site_access(query_db, current_user, inventory.site_id) + cls._validate_request_inventory(pallet, inventory, inventory.site_id) + + filling = None + if inventory.active_filling_id is not None: + filling = await OperationDao.get_filling_by_id(query_db, inventory.active_filling_id) + if filling is None: + raise ServiceException(message='托盘活动装填周期不存在') + cls._validate_active_filling(filling, inventory) + return cls._scan_response(pallet, inventory, filling, ['request']) + + @classmethod + async def _scan_confirm_pallet( + cls, + query_db: AsyncSession, + qr_token: str, + return_id: int, + current_user: CurrentUserModel, + ) -> PalletScanResponseModel: + warehouse_ids = await cls._warehouse_scope(query_db, current_user) + confirm_site_ids = None if warehouse_ids is None else [] + return_order = await OperationDao.get_return_by_id( + query_db, + return_id, + site_ids=confirm_site_ids, + warehouse_ids=warehouse_ids, + ) + if return_order is None: + cls._raise_return_not_accessible(current_user) + if return_order.status != ReturnStatus.REQUESTED.value: + 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) + matching_items = [item for item in return_order.items if item.pallet_id == pallet.pallet_id] + if len(matching_items) != 1: + raise ServiceException(message='扫码托盘不属于该退回单') + inventory = await MasterDao.get_pallet_inventory(query_db, pallet.pallet_id) + if inventory is None: + cls._raise_pallet_not_accessible(current_user, '托盘库存不存在') + item = matching_items[0] + cls._validate_confirm_snapshot(return_order, item, pallet, inventory) + + filling = None + if inventory.active_filling_id is not None: + filling = await OperationDao.get_filling_by_id(query_db, inventory.active_filling_id) + if filling is None: + raise ServiceException(message='托盘活动装填周期不存在') + cls._validate_active_filling(filling, inventory) + return cls._scan_response(pallet, inventory, filling, ['confirm']) + + @staticmethod + def _scan_response( + pallet: BizPallet, + inventory: BizPalletInventory, + filling: BizFillingRecord | None, + allowed_actions: list[str], + ) -> PalletScanResponseModel: + return PalletScanResponseModel( + pallet_id=pallet.pallet_id, + pallet_code=pallet.pallet_code, + flow_state=PalletFlowState(inventory.flow_state), + warehouse_id=inventory.warehouse_id, + site_id=inventory.site_id, + current_line_id=inventory.current_line_id, + active_filling_id=inventory.active_filling_id, + strain_id=filling.strain_id if filling is not None else None, + strain_code=filling.strain_code_snapshot if filling is not None else None, + strain_name=filling.strain_name_snapshot if filling is not None else None, + batch_no=filling.batch_no if filling is not None else None, + package_spec=filling.package_spec if filling is not None else None, + capacity_packages=pallet.capacity_packages, + remaining_packages=inventory.remaining_packages, + allowed_actions=allowed_actions, + ) + + @classmethod + async def scan_return_pallet_services( + cls, + query_db: AsyncSession, + qr_token: str, + query_object: ReturnScanQueryModel, + current_user: CurrentUserModel, + ) -> PalletScanResponseModel: + normalized_qr = qr_token.strip() + if not normalized_qr: + raise ServiceException(message='托盘二维码不能为空') + if query_object.mode == 'request': + return await cls._scan_request_pallet(query_db, normalized_qr, current_user) + if query_object.return_id is None: + raise ServiceException(message='确认退回扫码时必须提供退回单ID') + return await cls._scan_confirm_pallet( + query_db, + normalized_qr, + query_object.return_id, + current_user, + ) + + @classmethod + async def _lock_request_pallets( + cls, + query_db: AsyncSession, + pallet_ids: list[int], + source_site_id: int, + ) -> tuple[list[BizPalletInventory], list[BizPallet], dict[int, 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) + inventory_by_id = {inventory.pallet_id: inventory for inventory in inventories} + pallet_by_id = {pallet.pallet_id: pallet for pallet in pallets} + if set(inventory_by_id) != set(ordered_ids) or set(pallet_by_id) != set(ordered_ids): + raise ServiceException(message='存在不存在或库存缺失的托盘') + + 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] + for pallet, inventory in zip(ordered_pallets, ordered_inventories, strict=True): + cls._validate_request_inventory(pallet, inventory, source_site_id) + + filling_by_id = {} + filling_ids = sorted( + { + inventory.active_filling_id + for inventory in ordered_inventories + if inventory.active_filling_id is not None + } + ) + for filling_id in filling_ids: + filling = await OperationDao.get_active_filling_for_update(query_db, filling_id) + if filling is None: + raise ServiceException(message='托盘活动装填周期不存在或已关闭') + filling_by_id[filling_id] = filling + for inventory in ordered_inventories: + if inventory.active_filling_id is not None: + cls._validate_active_filling(filling_by_id[inventory.active_filling_id], inventory) + return ordered_inventories, ordered_pallets, filling_by_id + + @staticmethod + def _build_return_items( + return_id: int, + inventories: list[BizPalletInventory], + user_name: str, + current_time: datetime, + ) -> list[BizReturnItem]: + return [ + BizReturnItem( + return_id=return_id, + pallet_id=inventory.pallet_id, + active_filling_id_snapshot=inventory.active_filling_id, + remaining_packages_snapshot=inventory.remaining_packages, + create_by=user_name, + create_time=current_time, + ) + for inventory in inventories + ] + + @classmethod + async def _lock_new_request_route( + cls, + query_db: AsyncSession, + page_object: ReturnRequestModel, + current_user: CurrentUserModel, + ) -> None: + site = await MasterDao.get_site(query_db, page_object.source_site_id, for_update=True) + await AccessService.assert_site_access( + query_db, + current_user, + page_object.source_site_id, + for_update=True, + ) + if site is None: + raise ServiceException(message='来源客户现场不存在') + if site.status != '0': + raise ServiceException(message='来源客户现场已停用') + + warehouse = await MasterDao.get_warehouse( + query_db, + page_object.target_warehouse_id, + for_update=True, + ) + if warehouse is None: + raise ServiceException(message='目标仓库不存在') + if warehouse.status != '0': + raise ServiceException(message='目标仓库已停用') + + @classmethod + async def _lock_and_reauthorize_request_site( + cls, + query_db: AsyncSession, + return_order: BizReturnOrder, + current_user: CurrentUserModel, + ) -> None: + site = await MasterDao.get_site(query_db, return_order.source_site_id, for_update=True) + try: + await AccessService.assert_site_access( + query_db, + current_user, + return_order.source_site_id, + for_update=True, + ) + except PermissionException: + cls._raise_return_not_accessible(current_user) + if site is None: + raise ServiceException(message='来源客户现场不存在') + + @classmethod + async def _recover_request_integrity_error( + cls, + query_db: AsyncSession, + page_object: ReturnRequestModel, + request_hash: str, + current_user: CurrentUserModel, + ) -> CrudResponseModel: + try: + site_ids = await cls._site_scope(query_db, current_user) + request_warehouse_ids = None if site_ids is None else [] + recovered = await OperationDao.get_return_by_request_idempotency_key( + query_db, + page_object.idempotency_key, + site_ids=site_ids, + warehouse_ids=request_warehouse_ids, + ) + if recovered is None or not cls._is_exact_request_result(recovered, page_object, request_hash): + raise ServiceException(message='申请退回失败，幂等冲突') + locked = await OperationDao.get_return_for_update( + query_db, + recovered.return_id, + site_ids=site_ids, + warehouse_ids=request_warehouse_ids, + ) + if locked is None: + cls._raise_return_not_accessible(current_user) + if not cls._is_exact_request_result( + locked, + page_object, + request_hash, + require_items=True, + ): + raise ServiceException(message='申请退回失败，幂等键已被其他请求使用') + await cls._lock_and_reauthorize_request_site(query_db, locked, current_user) + return cls._success('申请成功', locked.return_id) + finally: + await query_db.rollback() + + @classmethod + async def request_return_services( + cls, + query_db: AsyncSession, + page_object: ReturnRequestModel, + current_user: CurrentUserModel, + ) -> CrudResponseModel: + try: + user_id, user_name = cls._current_identity(current_user) + ordered_ids = cls._validate_pallet_ids(page_object.pallet_ids) + request_hash = cls._build_request_hash(page_object) + site_ids = await cls._site_scope(query_db, current_user) + if site_ids is not None and page_object.source_site_id not in site_ids: + raise PermissionException(data='', message='无权访问该客户现场') + await AccessService.assert_site_access( + query_db, + current_user, + page_object.source_site_id, + ) + request_warehouse_ids = None if site_ids is None else [] + existing = await OperationDao.get_return_by_request_idempotency_key( + query_db, + page_object.idempotency_key, + site_ids=site_ids, + warehouse_ids=request_warehouse_ids, + ) + if existing is not None and not cls._is_exact_request_result(existing, page_object, request_hash): + raise ServiceException(message='申请退回失败，幂等键已被其他请求使用') + + await query_db.rollback() + if existing is not None: + locked_existing = await OperationDao.get_return_for_update( + query_db, + existing.return_id, + site_ids=site_ids, + warehouse_ids=request_warehouse_ids, + ) + if locked_existing is None: + cls._raise_return_not_accessible(current_user) + if not cls._is_exact_request_result( + locked_existing, + page_object, + request_hash, + require_items=True, + ): + raise ServiceException(message='申请退回失败，幂等键已被其他请求使用') + await cls._lock_and_reauthorize_request_site(query_db, locked_existing, current_user) + return_id = locked_existing.return_id + await query_db.rollback() + return cls._success('申请成功', return_id) + + inventories, _, _ = await cls._lock_request_pallets( + query_db, + ordered_ids, + page_object.source_site_id, + ) + await cls._lock_new_request_route(query_db, page_object, current_user) + for inventory in inventories: + PalletStateMachine.ensure_transition( + PalletFlowState(inventory.flow_state), + PalletFlowState.RETURNING, + ) + + current_time = datetime.now() + return_order = BizReturnOrder( + return_no=f'RE-{secrets.token_hex(16)}', + source_site_id=page_object.source_site_id, + target_warehouse_id=page_object.target_warehouse_id, + status=ReturnStatus.REQUESTED.value, + request_idempotency_key=page_object.idempotency_key, + request_hash=request_hash, + requested_by=user_id, + requested_at=current_time, + create_by=user_name, + create_time=current_time, + update_by=user_name, + update_time=current_time, + ) + return_order = await OperationDao.add_return_order(query_db, return_order) + await OperationDao.add_return_items( + query_db, + cls._build_return_items(return_order.return_id, inventories, user_name, current_time), + ) + + 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.RETURNING.value + inventory.version += 1 + inventory.updated_at = current_time + event = BizPalletEvent( + pallet_id=inventory.pallet_id, + event_type=PalletEventType.RETURN_REQUESTED.value, + source_type='RETURN', + source_id=return_order.return_id, + before_state=before_state, + after_state=PalletFlowState.RETURNING.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=inventory.site_id, + after_line_id=inventory.current_line_id, + before_remaining_packages=before_remaining_packages, + after_remaining_packages=inventory.remaining_packages, + operator_id=user_id, + occurred_at=current_time, + idempotency_key=page_object.idempotency_key, + request_hash=request_hash, + ) + await InventoryDao.add_event(query_db, inventory, event) + + await query_db.commit() + return cls._success('申请成功', return_order.return_id) + except IntegrityError: + await query_db.rollback() + return await cls._recover_request_integrity_error( + query_db, + page_object, + cls._build_request_hash(page_object), + current_user, + ) + except Exception: + await query_db.rollback() + raise + + @staticmethod + def _validate_confirm_snapshot( + return_order: BizReturnOrder, + item: BizReturnItem, + pallet: BizPallet, + inventory: BizPalletInventory, + ) -> None: + if pallet.pallet_id != inventory.pallet_id or item.pallet_id != inventory.pallet_id: + raise ServiceException(message='退回单托盘资产与库存不一致') + if pallet.asset_status != '0': + raise ServiceException(message='托盘资产已停用，不允许确认退回') + if ( + inventory.flow_state != PalletFlowState.RETURNING.value + or inventory.site_id != return_order.source_site_id + or inventory.warehouse_id is not None + ): + raise ServiceException(message='托盘当前状态或位置不允许确认退回') + if ( + inventory.remaining_packages != item.remaining_packages_snapshot + or inventory.active_filling_id != item.active_filling_id_snapshot + ): + raise ServiceException(message='托盘库存与退回申请快照不一致') + if ( + type(inventory.remaining_packages) is not int + or not 0 <= inventory.remaining_packages <= pallet.capacity_packages + ): + raise ServiceException(message='托盘剩余包数无效') + if type(inventory.version) is not int or inventory.version < 0: + raise ServiceException(message='托盘库存版本无效') + if inventory.remaining_packages > 0 and inventory.active_filling_id is None: + raise ServiceException(message='带料托盘活动装填快照无效') + if inventory.remaining_packages == 0 and inventory.active_filling_id is not None: + raise ServiceException(message='空托盘活动装填快照无效') + + @classmethod + async def _lock_confirm_pallets( + cls, + query_db: AsyncSession, + return_order: BizReturnOrder, + items: list[BizReturnItem], + ) -> tuple[list[BizPalletInventory], dict[int, BizFillingRecord]]: + ordered_ids = sorted(item.pallet_id for item in items) + inventories = await InventoryDao.get_inventories_for_update(query_db, ordered_ids) + pallets = await InventoryDao.get_pallets_by_ids(query_db, ordered_ids) + inventory_by_id = {inventory.pallet_id: inventory for inventory in inventories} + pallet_by_id = {pallet.pallet_id: pallet for pallet in pallets} + item_by_id = {item.pallet_id: item for item in items} + if ( + set(inventory_by_id) != set(ordered_ids) + or set(pallet_by_id) != set(ordered_ids) + or set(item_by_id) != set(ordered_ids) + ): + raise ServiceException(message='退回单存在不存在或库存缺失的托盘') + + ordered_inventories = [inventory_by_id[pallet_id] for pallet_id in ordered_ids] + for pallet_id in ordered_ids: + cls._validate_confirm_snapshot( + return_order, + item_by_id[pallet_id], + pallet_by_id[pallet_id], + inventory_by_id[pallet_id], + ) + + filling_by_id = {} + filling_ids = sorted( + { + inventory.active_filling_id + for inventory in ordered_inventories + if inventory.active_filling_id is not None + } + ) + for filling_id in filling_ids: + filling = await OperationDao.get_active_filling_for_update(query_db, filling_id) + if filling is None: + raise ServiceException(message='托盘活动装填周期不存在或已关闭') + filling_by_id[filling_id] = filling + for inventory in ordered_inventories: + if inventory.active_filling_id is not None: + cls._validate_active_filling(filling_by_id[inventory.active_filling_id], inventory) + return ordered_inventories, filling_by_id + + @classmethod + async def _lock_and_reauthorize_confirm_warehouse( + cls, + query_db: AsyncSession, + return_order: BizReturnOrder, + current_user: CurrentUserModel, + *, + require_active: bool, + ) -> None: + warehouse = await MasterDao.get_warehouse( + query_db, + return_order.target_warehouse_id, + for_update=True, + ) + try: + await AccessService.assert_warehouse_access( + query_db, + current_user, + return_order.target_warehouse_id, + for_update=True, + ) + except PermissionException: + cls._raise_return_not_accessible(current_user) + if warehouse is None: + raise ServiceException(message='目标仓库不存在') + if require_active and warehouse.status != '0': + raise ServiceException(message='目标仓库已停用') + + @classmethod + async def _recover_confirm_integrity_error( + cls, + query_db: AsyncSession, + return_id: int, + page_object: ReturnConfirmModel, + request_hash: str, + current_user: CurrentUserModel, + ) -> CrudResponseModel: + try: + warehouse_ids = await cls._warehouse_scope(query_db, current_user) + confirm_site_ids = None if warehouse_ids is None else [] + recovered = await OperationDao.get_return_by_confirm_idempotency_key( + query_db, + page_object.idempotency_key, + site_ids=confirm_site_ids, + warehouse_ids=warehouse_ids, + ) + if recovered is None: + raise ServiceException(message='确认退回失败，幂等冲突') + if not cls._is_exact_confirm_result(recovered, return_id, page_object, request_hash): + raise ServiceException(message='确认退回失败，幂等键已被其他请求使用') + locked = await OperationDao.get_return_for_update( + query_db, + return_id, + site_ids=confirm_site_ids, + warehouse_ids=warehouse_ids, + ) + if locked is None: + cls._raise_return_not_accessible(current_user) + if not cls._is_exact_confirm_result( + locked, + return_id, + page_object, + request_hash, + require_items=True, + ): + raise ServiceException(message='确认退回失败，幂等键已被其他请求使用') + await cls._lock_and_reauthorize_confirm_warehouse( + query_db, + locked, + current_user, + require_active=False, + ) + return cls._success('确认成功', return_id) + finally: + await query_db.rollback() + + @classmethod + async def confirm_return_services( # noqa: PLR0912, PLR0915 + cls, + query_db: AsyncSession, + return_id: int, + page_object: ReturnConfirmModel, + current_user: CurrentUserModel, + ) -> CrudResponseModel: + try: + user_id, user_name = cls._current_identity(current_user) + scanned_ids = cls._validate_pallet_ids(page_object.scanned_pallet_ids) + request_hash = cls._build_confirm_request_hash(return_id, page_object) + warehouse_ids = await cls._warehouse_scope(query_db, current_user) + confirm_site_ids = None if warehouse_ids is None else [] + existing = await OperationDao.get_return_by_confirm_idempotency_key( + query_db, + page_object.idempotency_key, + site_ids=confirm_site_ids, + warehouse_ids=warehouse_ids, + ) + if existing is not None and not cls._is_exact_confirm_result( + existing, + return_id, + page_object, + request_hash, + ): + raise ServiceException(message='确认退回失败，幂等键已被其他请求使用') + + return_order = await OperationDao.get_return_by_id( + query_db, + return_id, + site_ids=confirm_site_ids, + warehouse_ids=warehouse_ids, + ) + if return_order is None: + cls._raise_return_not_accessible(current_user) + try: + await AccessService.assert_warehouse_access( + query_db, + current_user, + return_order.target_warehouse_id, + ) + except PermissionException: + cls._raise_return_not_accessible(current_user) + + await query_db.rollback() + locked = await OperationDao.get_return_for_update( + query_db, + return_id, + site_ids=confirm_site_ids, + warehouse_ids=warehouse_ids, + ) + if locked is None: + cls._raise_return_not_accessible(current_user) + if locked.status == ReturnStatus.CONFIRMED.value: + if cls._is_exact_confirm_result( + locked, + return_id, + page_object, + request_hash, + require_items=True, + ): + await cls._lock_and_reauthorize_confirm_warehouse( + query_db, + locked, + current_user, + require_active=False, + ) + await query_db.rollback() + return cls._success('确认成功', return_id) + raise ServiceException(message='退回单已确认，幂等信息不匹配') + if locked.status != ReturnStatus.REQUESTED.value: + raise ServiceException(message='当前退回单状态不允许确认') + if existing is not None: + raise ServiceException(message='确认退回失败，并发状态不一致') + if locked.confirm_idempotency_key is not None or locked.confirm_request_hash is not None: + raise ServiceException(message='退回单确认信息不完整') + + items = list(locked.items) + item_ids = [item.pallet_id for item in items] + if not item_ids or len(item_ids) != len(set(item_ids)): + raise ServiceException(message='退回单托盘明细为空或重复') + if scanned_ids != sorted(item_ids): + raise ServiceException(message='确认扫码托盘集合与退回单明细不一致') + + inventories, filling_by_id = await cls._lock_confirm_pallets(query_db, locked, items) + await cls._lock_and_reauthorize_confirm_warehouse( + query_db, + locked, + current_user, + require_active=True, + ) + for inventory in inventories: + target_state = ( + PalletFlowState.FILLED_IN_WAREHOUSE + if inventory.remaining_packages > 0 + else PalletFlowState.EMPTY_IN_WAREHOUSE + ) + PalletStateMachine.ensure_transition(PalletFlowState(inventory.flow_state), target_state) + + current_time = datetime.now() + locked.status = ReturnStatus.CONFIRMED.value + locked.confirm_idempotency_key = page_object.idempotency_key + locked.confirm_request_hash = request_hash + locked.confirmed_by = user_id + locked.confirmed_at = current_time + locked.update_by = user_name + locked.update_time = current_time + await OperationDao.update_return_order(query_db, locked) + + 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 + if inventory.remaining_packages > 0: + target_state = PalletFlowState.FILLED_IN_WAREHOUSE.value + filling = filling_by_id[inventory.active_filling_id] + filling.warehouse_id = locked.target_warehouse_id + filling.update_by = user_name + filling.update_time = current_time + await OperationDao.update_filling_record(query_db, filling) + else: + target_state = PalletFlowState.EMPTY_IN_WAREHOUSE.value + inventory.active_filling_id = None + + inventory.flow_state = target_state + inventory.warehouse_id = locked.target_warehouse_id + inventory.site_id = None + inventory.current_line_id = None + inventory.version += 1 + inventory.updated_at = current_time + event = BizPalletEvent( + pallet_id=inventory.pallet_id, + event_type=PalletEventType.RETURN_CONFIRMED.value, + source_type='RETURN', + source_id=return_id, + before_state=before_state, + after_state=target_state, + before_warehouse_id=before_warehouse_id, + before_site_id=before_site_id, + before_line_id=before_line_id, + after_warehouse_id=locked.target_warehouse_id, + after_site_id=None, + 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=page_object.idempotency_key, + request_hash=request_hash, + ) + await InventoryDao.add_event(query_db, inventory, event) + + await query_db.commit() + return cls._success('确认成功', return_id) + except IntegrityError: + await query_db.rollback() + return await cls._recover_confirm_integrity_error( + query_db, + return_id, + page_object, + cls._build_confirm_request_hash(return_id, page_object), + current_user, + ) + except Exception: + await query_db.rollback() + raise 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 2202139..744d35c 100644 --- a/shierfeng-fastapi-backend/tests/module_business/test_inventory_dao.py +++ b/shierfeng-fastapi-backend/tests/module_business/test_inventory_dao.py @@ -123,7 +123,7 @@ class TestDaoPublicContract(unittest.TestCase): expected = { 'get_inbound_for_update': ('db', 'inbound_id', 'warehouse_ids'), 'get_shipment_for_update': ('db', 'shipment_id', 'warehouse_ids', 'site_ids'), - 'get_return_for_update': ('db', 'return_id'), + 'get_return_for_update': ('db', 'return_id', 'site_ids', 'warehouse_ids'), 'get_active_filling_for_update': ('db', 'filling_id'), 'get_inbound_by_confirm_idempotency_key': ('db', 'idempotency_key'), 'get_shipment_by_confirm_idempotency_key': ( @@ -134,8 +134,18 @@ class TestDaoPublicContract(unittest.TestCase): ), 'get_filling_by_idempotency_key': ('db', 'idempotency_key', 'warehouse_ids'), 'get_usage_by_idempotency_key': ('db', 'idempotency_key', 'site_ids'), - 'get_return_by_request_idempotency_key': ('db', 'idempotency_key'), - 'get_return_by_confirm_idempotency_key': ('db', 'idempotency_key'), + 'get_return_by_request_idempotency_key': ( + 'db', + 'idempotency_key', + 'site_ids', + 'warehouse_ids', + ), + 'get_return_by_confirm_idempotency_key': ( + 'db', + 'idempotency_key', + 'site_ids', + 'warehouse_ids', + ), 'get_inbound_by_id': ('db', 'inbound_id', 'warehouse_ids'), 'add_inbound': ('db', 'inbound'), 'update_inbound': ('db', 'inbound'), @@ -151,7 +161,7 @@ class TestDaoPublicContract(unittest.TestCase): 'replace_shipment_items': ('db', 'shipment_id', 'items'), 'get_usage_by_id': ('db', 'usage_id', 'site_ids'), 'add_usage_record': ('db', 'usage'), - 'get_return_by_id': ('db', 'return_id'), + 'get_return_by_id': ('db', 'return_id', 'site_ids', 'warehouse_ids'), 'add_return_order': ('db', 'return_order'), 'update_return_order': ('db', 'return_order'), 'add_return_items': ('db', 'items'), @@ -228,8 +238,22 @@ class TestDaoPublicContract(unittest.TestCase): self.assertNotIn('for_update', inspect.signature(OperationDao._get_by_key).parameters) return_list_signature = inspect.signature(OperationDao.get_return_list) - self.assertEqual(('db', 'query_object', 'is_page'), tuple(return_list_signature.parameters)) + self.assertEqual( + ('db', 'query_object', 'is_page', 'site_ids', 'warehouse_ids'), + tuple(return_list_signature.parameters), + ) self.assertIs(True, return_list_signature.parameters['is_page'].default) + for method_name in ( + 'get_return_for_update', + 'get_return_by_request_idempotency_key', + 'get_return_by_confirm_idempotency_key', + 'get_return_by_id', + 'get_return_list', + ): + with self.subTest(method=method_name): + signature = inspect.signature(getattr(OperationDao, method_name)) + self.assertIsNone(signature.parameters['site_ids'].default) + self.assertIsNone(signature.parameters['warehouse_ids'].default) def test_dao_source_never_controls_transaction_boundary(self) -> None: for module in (inventory_dao_module, operation_dao_module): @@ -423,13 +447,6 @@ class TestOperationDaoReads(unittest.IsolatedAsyncioTestCase): async def test_for_update_queries_target_primary_keys_and_lock_rows(self) -> None: cases = ( - ( - OperationDao.get_return_for_update, - BizReturnOrder, - 'return_id', - 13, - (), - ), ( OperationDao.get_active_filling_for_update, BizFillingRecord, diff --git a/shierfeng-fastapi-backend/tests/module_business/test_return_service.py b/shierfeng-fastapi-backend/tests/module_business/test_return_service.py new file mode 100644 index 0000000..31bae1b --- /dev/null +++ b/shierfeng-fastapi-backend/tests/module_business/test_return_service.py @@ -0,0 +1,1082 @@ +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.dialects import mysql, postgresql +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.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, ReturnStatus +from module_business.entity.do.master_do import BizCustomerSite, BizPallet, BizWarehouse +from module_business.entity.do.operation_do import ( + BizFillingRecord, + BizPalletEvent, + BizPalletInventory, + BizReturnItem, + BizReturnOrder, +) +from module_business.entity.vo.operation_vo import ( + BusinessOptionModel, + PalletScanResponseModel, + ReturnConfirmModel, + ReturnModel, + ReturnPageQueryModel, + ReturnRequestModel, + ReturnScanQueryModel, +) +from module_business.service.access_service import AccessService + +RETURN_SERVICE_SPEC = importlib.util.find_spec('module_business.service.return_service') +RETURN_CONTROLLER_SPEC = importlib.util.find_spec('module_business.controller.return_controller') + +if RETURN_SERVICE_SPEC is not None: + from module_business.service import return_service as return_service_module + from module_business.service.return_service import ReturnService +else: + return_service_module = None + ReturnService = None + +if RETURN_CONTROLLER_SPEC is not None: + from module_business.controller import return_controller as return_controller_module +else: + return_controller_module = None + + +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 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, + *, + site_id: int | None = 9, + warehouse_id: int | None = None, + line_id: int | None = 4, + remaining: int = 8, + filling_id: int | None = -1, + state: str = PalletFlowState.AT_SITE_PARTIAL.value, + version: int = 3, +) -> BizPalletInventory: + active_filling_id = (100 + pallet_id if remaining > 0 else None) if filling_id == -1 else filling_id + return BizPalletInventory( + pallet_id=pallet_id, + warehouse_id=warehouse_id, + site_id=site_id, + current_line_id=line_id, + active_filling_id=active_filling_id, + remaining_packages=remaining, + flow_state=state, + version=version, + ) + + +def make_filling(pallet_id: int, *, warehouse_id: int = 3) -> 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=datetime(2026, 7, 1), + ) + + +def make_return_order( + *, + return_id: int = 13, + status: str = ReturnStatus.REQUESTED.value, + snapshots: tuple[tuple[int, int | None, int], ...] = ((5, None, 0), (2, 102, 8)), + request_key: str = 'request-key', + request_hash: str = 'r' * 64, + confirm_key: str | None = None, + confirm_hash: str | None = None, +) -> BizReturnOrder: + order = BizReturnOrder( + return_id=return_id, + return_no=f'RE-{return_id}', + source_site_id=9, + target_warehouse_id=7, + status=status, + request_idempotency_key=request_key, + request_hash=request_hash, + confirm_idempotency_key=confirm_key, + confirm_request_hash=confirm_hash, + requested_by=8, + requested_at=datetime(2026, 7, 25, 9), + ) + order.items = [ + BizReturnItem( + return_item_id=index, + return_id=return_id, + pallet_id=pallet_id, + active_filling_id_snapshot=filling_id, + remaining_packages_snapshot=remaining, + ) + for index, (pallet_id, filling_id, remaining) in enumerate(snapshots, start=1) + ] + return order + + +def integrity_error() -> IntegrityError: + return IntegrityError('statement', {}, RuntimeError('constraint')) + + +def compile_sql(statement: object, dialect: object) -> str: + compiled = statement.compile( + dialect=dialect, + compile_kwargs={'literal_binds': True}, + ) + return ' '.join(str(compiled).split()).replace('%%', '%') + + +class ReturnBootstrapRedTest(unittest.TestCase): + def test_return_service_module_exists(self) -> None: + self.assertIsNotNone(RETURN_SERVICE_SPEC, 'ReturnService 尚未实现') + + def test_return_controller_module_exists(self) -> None: + self.assertIsNotNone(RETURN_CONTROLLER_SPEC, 'return_controller 尚未实现') + + +class ReturnDaoScopeRedTest(unittest.IsolatedAsyncioTestCase): + def setUp(self) -> None: + self.db = MagicMock(spec=AsyncSession) + + def test_return_reads_expose_stage_scopes(self) -> None: + expected = { + OperationDao.get_return_for_update: ('db', 'return_id', 'site_ids', 'warehouse_ids'), + OperationDao.get_return_by_request_idempotency_key: ( + 'db', + 'idempotency_key', + 'site_ids', + 'warehouse_ids', + ), + OperationDao.get_return_by_confirm_idempotency_key: ( + 'db', + 'idempotency_key', + 'site_ids', + 'warehouse_ids', + ), + OperationDao.get_return_by_id: ('db', 'return_id', 'site_ids', 'warehouse_ids'), + } + for method, parameters in expected.items(): + with self.subTest(method=method.__name__): + self.assertEqual(parameters, tuple(inspect.signature(method).parameters)) + self.assertIsNone(inspect.signature(method).parameters['site_ids'].default) + self.assertIsNone(inspect.signature(method).parameters['warehouse_ids'].default) + + list_signature = inspect.signature(OperationDao.get_return_list) + self.assertEqual( + ('db', 'query_object', 'is_page', 'site_ids', 'warehouse_ids'), + tuple(list_signature.parameters), + ) + self.assertIs(True, list_signature.parameters['is_page'].default) + + async def test_return_for_update_locks_scoped_parent_then_fresh_items(self) -> None: + self.assertIn( + 'site_ids', + inspect.signature(OperationDao.get_return_for_update).parameters, + '退回单锁查询尚未支持阶段数据范围', + ) + order = SimpleNamespace(return_id=13) + items = [ + SimpleNamespace(return_item_id=4, return_id=13, pallet_id=5), + SimpleNamespace(return_item_id=2, return_id=13, pallet_id=2), + ] + parent_result = MagicMock() + parent_result.scalar_one_or_none.return_value = order + 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_return_for_update( + self.db, + 13, + site_ids=[9, 7, 9], + warehouse_ids=[], + ) + + self.assertIs(order, 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(BizReturnOrder.__table__)) + self.assertTrue(item_statement.get_final_froms()[0].compare(BizReturnItem.__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()): + parent_sql = compile_sql(parent_statement, dialect) + item_sql = compile_sql(item_statement, dialect) + self.assertIn('biz_return_order.return_id = 13', parent_sql) + self.assertIn('biz_return_order.source_site_id IN (7, 9)', parent_sql) + self.assertIn('FOR UPDATE', parent_sql) + self.assertIn('biz_return_item.return_id = 13', item_sql) + self.assertIn('ORDER BY biz_return_item.pallet_id, biz_return_item.return_item_id', item_sql) + self.assertIn('FOR UPDATE', item_sql) + + async def test_return_scope_uses_site_or_warehouse_visibility(self) -> None: + self.assertIn( + 'site_ids', + inspect.signature(OperationDao.get_return_by_id).parameters, + '退回单详情尚未支持阶段数据范围', + ) + result = MagicMock() + result.scalar_one_or_none.return_value = None + self.db.execute.return_value = result + + await OperationDao.get_return_by_id( + self.db, + 13, + site_ids=[9], + warehouse_ids=[7], + ) + + statement = self.db.execute.await_args.args[0] + sql = compile_sql(statement, postgresql.dialect()) + self.assertIn('biz_return_order.source_site_id IN (9)', sql) + self.assertIn('biz_return_order.target_warehouse_id IN (7)', sql) + self.assertIn(' OR ', sql) + + +@unittest.skipUnless(ReturnService is not None, 'ReturnService 尚未实现') +class ReturnServiceContractRedTest(unittest.TestCase): + def test_service_exposes_complete_return_workflow(self) -> None: + expected = { + 'get_return_list_services', + 'return_detail_services', + 'get_warehouse_options_services', + 'scan_return_pallet_services', + 'request_return_services', + 'confirm_return_services', + } + self.assertEqual(expected, {name for name in expected if hasattr(ReturnService, name)}) + + def test_request_and_confirm_signatures_are_exact(self) -> None: + request_signature = inspect.signature(ReturnService.request_return_services) + confirm_signature = inspect.signature(ReturnService.confirm_return_services) + self.assertEqual(('query_db', 'page_object', 'current_user'), tuple(request_signature.parameters)) + self.assertEqual( + ('query_db', 'return_id', 'page_object', 'current_user'), + tuple(confirm_signature.parameters), + ) + request_hints = get_type_hints(ReturnService.request_return_services) + confirm_hints = get_type_hints(ReturnService.confirm_return_services) + self.assertIs(AsyncSession, request_hints['query_db']) + self.assertIs(ReturnRequestModel, request_hints['page_object']) + self.assertIs(CurrentUserModel, request_hints['current_user']) + self.assertIs(CrudResponseModel, request_hints['return']) + self.assertIs(AsyncSession, confirm_hints['query_db']) + self.assertIs(int, confirm_hints['return_id']) + self.assertIs(ReturnConfirmModel, confirm_hints['page_object']) + self.assertIs(CurrentUserModel, confirm_hints['current_user']) + self.assertIs(CrudResponseModel, confirm_hints['return']) + + def test_request_and_confirm_hashes_are_canonical_and_exclude_keys(self) -> None: + request = ReturnRequestModel( + sourceSiteId=9, + targetWarehouseId=7, + palletIds=[5, 2], + idempotencyKey='request-key-a', + ) + confirm = ReturnConfirmModel(scannedPalletIds=[5, 2], idempotencyKey='confirm-key-a') + expected_request = hashlib.sha256( + json.dumps( + {'pallet_ids': [2, 5], 'source_site_id': 9, 'target_warehouse_id': 7}, + sort_keys=True, + separators=(',', ':'), + ).encode() + ).hexdigest() + expected_confirm = hashlib.sha256( + json.dumps( + {'return_id': 13, 'scanned_pallet_ids': [2, 5]}, + sort_keys=True, + separators=(',', ':'), + ).encode() + ).hexdigest() + + self.assertEqual(expected_request, ReturnService._build_request_hash(request)) + self.assertEqual(expected_confirm, ReturnService._build_confirm_request_hash(13, confirm)) + self.assertEqual( + expected_request, + ReturnService._build_request_hash(request.model_copy(update={'idempotency_key': 'request-key-b'})), + ) + self.assertEqual( + expected_confirm, + ReturnService._build_confirm_request_hash( + 13, + confirm.model_copy(update={'idempotency_key': 'confirm-key-b'}), + ), + ) + + +@unittest.skipUnless(return_controller_module is not None, 'return_controller 尚未实现') +class ReturnControllerContractRedTest(unittest.TestCase): + @staticmethod + def _routes() -> dict[tuple[str, str], APIRoute]: + return { + (route.path, next(iter(route.methods))): route + for route in return_controller_module.return_controller.routes + if isinstance(route, APIRoute) + } + + def test_openapi_has_exact_seven_operations_aliases_and_models(self) -> None: + app = FastAPI() + app.include_router(return_controller_module.return_controller) + schema = app.openapi() + expected = { + ('/business/return/list', 'get'), + ('/business/return/warehouse-options', 'get'), + ('/business/return/scan/{qr_token}', 'get'), + ('/business/return/{return_id}', 'get'), + ('/business/return/request', 'post'), + ('/business/return/{return_id}/confirm', 'put'), + } + self.assertEqual(expected, {(path, method) for path, methods in schema['paths'].items() for method in methods}) + parameters = schema['paths']['/business/return/scan/{qr_token}']['get']['parameters'] + self.assertEqual({'qr_token', 'mode', 'returnId'}, {parameter['name'] for parameter in parameters}) + routes = self._routes() + self.assertEqual(PageResponseModel[ReturnModel], routes[('/business/return/list', 'GET')].response_model) + self.assertEqual( + DataResponseModel[list[BusinessOptionModel]], + routes[('/business/return/warehouse-options', 'GET')].response_model, + ) + self.assertEqual( + DataResponseModel[PalletScanResponseModel], + routes[('/business/return/scan/{qr_token}', 'GET')].response_model, + ) + self.assertEqual( + DataResponseModel[ReturnModel], + routes[('/business/return/{return_id}', 'GET')].response_model, + ) + self.assertEqual(ResponseBaseModel, routes[('/business/return/request', 'POST')].response_model) + self.assertEqual(ResponseBaseModel, routes[('/business/return/{return_id}/confirm', 'PUT')].response_model) + + def test_static_routes_permissions_current_user_and_write_logs(self) -> None: + routes = self._routes() + paths = [route.path for route in return_controller_module.return_controller.routes] + detail_index = paths.index('/business/return/{return_id}') + for static_path in ( + '/business/return/list', + '/business/return/warehouse-options', + '/business/return/scan/{qr_token}', + '/business/return/request', + ): + self.assertLess(paths.index(static_path), detail_index) + + expected = { + ('/business/return/list', 'GET'): 'business:return:list', + ('/business/return/warehouse-options', 'GET'): 'business:return:query', + ('/business/return/scan/{qr_token}', 'GET'): 'business:return:query', + ('/business/return/{return_id}', 'GET'): 'business:return:query', + ('/business/return/request', 'POST'): 'business:return:request', + ('/business/return/{return_id}/confirm', 'PUT'): 'business:return:confirm', + } + actual = {} + for key, route in 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'}: + self.assertTrue(hasattr(route.endpoint, '__wrapped__'), key) + actual[key] = permissions[0].perm + self.assertEqual(expected, actual) + + source = inspect.getsource(return_controller_module) + for forbidden in ( + 'business:return:add', + 'business:return:edit', + 'business:return:remove', + 'business:return:export', + ): + self.assertNotIn(forbidden, source) + + +@unittest.skipUnless(ReturnService is not None, 'ReturnService 尚未实现') +class ReturnReadAndScanRedTest(unittest.IsolatedAsyncioTestCase): + def setUp(self) -> None: + self.db = MagicMock(spec=AsyncSession) + self.user = make_user() + + async def test_list_uses_site_or_warehouse_visibility_and_empty_mapping_is_empty(self) -> None: + page = PageModel(rows=[], pageNum=1, pageSize=10, total=0, hasNext=False) + query = ReturnPageQueryModel() + with ( + patch.object(AccessDao, 'list_user_site_ids', new=AsyncMock(return_value=[9, 9])), + patch.object(AccessDao, 'list_user_warehouse_ids', new=AsyncMock(return_value=[7, 7])), + patch.object(OperationDao, 'get_return_list', new=AsyncMock(return_value=page)) as get_list, + ): + actual = await ReturnService.get_return_list_services(self.db, query, self.user) + + self.assertEqual([], actual.rows) + get_list.assert_awaited_once_with( + self.db, + query, + True, + site_ids=[9], + warehouse_ids=[7], + ) + + with ( + patch.object(AccessDao, 'list_user_site_ids', new=AsyncMock(return_value=[])), + patch.object(AccessDao, 'list_user_warehouse_ids', new=AsyncMock(return_value=[])), + patch.object(OperationDao, 'get_return_list', new=AsyncMock()) as empty_get_list, + ): + empty = await ReturnService.get_return_list_services(self.db, query, self.user) + self.assertEqual([], empty.rows) + empty_get_list.assert_not_awaited() + + async def test_warehouse_options_returns_all_enabled_without_warehouse_mapping(self) -> None: + warehouses = [ + BizWarehouse(warehouse_id=8, warehouse_code='W-8', warehouse_name='八号仓', status='1'), + BizWarehouse(warehouse_id=7, warehouse_code='W-7', warehouse_name='七号仓', status='0'), + BizWarehouse(warehouse_id=3, warehouse_code='W-3', warehouse_name='三号仓', status='0'), + ] + with ( + patch.object(MasterDao, 'list_warehouse', new=AsyncMock(return_value=warehouses)), + patch.object(AccessDao, 'list_user_warehouse_ids', new=AsyncMock()) as mappings, + ): + actual = await ReturnService.get_warehouse_options_services(self.db, self.user) + + self.assertEqual([3, 7], [option.id for option in actual]) + mappings.assert_not_awaited() + + async def test_request_scan_uses_current_site_scope_and_backend_inventory(self) -> None: + pallet = make_pallet(2) + inventory = make_inventory(2) + filling = make_filling(2) + with ( + patch.object(InventoryDao, 'get_pallet_by_qr_token', new=AsyncMock(return_value=pallet)) as get_pallet, + patch.object(MasterDao, 'get_pallet_inventory', new=AsyncMock(return_value=inventory)) as get_inventory, + patch.object(AccessService, 'assert_site_access', new=AsyncMock()) as site_access, + patch.object(OperationDao, 'get_filling_by_id', new=AsyncMock(return_value=filling)), + ): + actual = await ReturnService.scan_return_pallet_services( + self.db, + ' qr-2 ', + ReturnScanQueryModel(mode='request'), + self.user, + ) + + get_pallet.assert_awaited_once_with(self.db, 'qr-2') + get_inventory.assert_awaited_once_with(self.db, 2) + site_access.assert_awaited_once_with(self.db, self.user, 9) + self.assertEqual(PalletFlowState.AT_SITE_PARTIAL, actual.flow_state) + self.assertEqual(8, actual.remaining_packages) + self.assertEqual(['request'], actual.allowed_actions) + + async def test_confirm_scan_is_scoped_to_target_warehouse_and_exact_item_snapshot(self) -> None: + order = make_return_order(snapshots=((2, 102, 8),)) + pallet = make_pallet(2) + inventory = make_inventory(2, state=PalletFlowState.RETURNING.value) + filling = make_filling(2) + with ( + patch.object(AccessDao, 'list_user_warehouse_ids', new=AsyncMock(return_value=[7])), + patch.object(OperationDao, 'get_return_by_id', new=AsyncMock(return_value=order)) as get_order, + 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)), + ): + actual = await ReturnService.scan_return_pallet_services( + self.db, + 'qr-2', + ReturnScanQueryModel(mode='confirm', returnId=13), + self.user, + ) + + get_order.assert_awaited_once_with(self.db, 13, site_ids=[], warehouse_ids=[7]) + self.assertEqual(PalletFlowState.RETURNING, actual.flow_state) + self.assertEqual(['confirm'], actual.allowed_actions) + + async def test_confirm_scan_rejects_changed_snapshot_and_hides_unscoped_order(self) -> None: + order = make_return_order(snapshots=((2, 102, 8),)) + changed = make_inventory(2, remaining=7, state=PalletFlowState.RETURNING.value) + query = ReturnScanQueryModel(mode='confirm', returnId=13) + with ( + patch.object(AccessDao, 'list_user_warehouse_ids', new=AsyncMock(return_value=[7])), + patch.object(OperationDao, 'get_return_by_id', new=AsyncMock(return_value=order)), + patch.object(InventoryDao, 'get_pallet_by_qr_token', new=AsyncMock(return_value=make_pallet(2))), + patch.object(MasterDao, 'get_pallet_inventory', new=AsyncMock(return_value=changed)), + self.assertRaises(ServiceException) as changed_error, + ): + await ReturnService.scan_return_pallet_services(self.db, 'qr-2', query, self.user) + self.assertIn('快照', changed_error.exception.message) + + with ( + patch.object(AccessDao, 'list_user_warehouse_ids', new=AsyncMock(return_value=[])), + patch.object(OperationDao, 'get_return_by_id', new=AsyncMock(return_value=None)), + self.assertRaises(PermissionException) as hidden_error, + ): + await ReturnService.scan_return_pallet_services(self.db, 'qr-2', query, self.user) + self.assertEqual('无权访问该退回单', hidden_error.exception.message) + + +@unittest.skipUnless(ReturnService is not None, 'ReturnService 尚未实现') +class ReturnRequestTransactionRedTest(unittest.IsolatedAsyncioTestCase): + def setUp(self) -> None: + self.db = MagicMock(spec=AsyncSession) + self.user = make_user() + self.fixed_time = datetime(2026, 7, 25, 10, 30) + self.page = ReturnRequestModel( + sourceSiteId=9, + targetWarehouseId=7, + palletIds=[5, 2], + idempotencyKey='request-key', + ) + + async def test_request_freezes_all_pallets_saves_snapshots_and_appends_events(self) -> None: + inventories = [ + make_inventory(2), + make_inventory(5, remaining=0, state=PalletFlowState.AT_SITE_EMPTY.value), + ] + pallets = [make_pallet(2), make_pallet(5)] + filling = make_filling(2) + captured: dict[str, Any] = {} + events: list[BizPalletEvent] = [] + + async def add_order(_db: AsyncSession, order: BizReturnOrder) -> BizReturnOrder: + order.return_id = 13 + captured['order'] = order + return order + + async def add_items(_db: AsyncSession, items: list[BizReturnItem]) -> list[BizReturnItem]: + captured['items'] = items + return items + + async def add_event( + _db: AsyncSession, + _inventory: BizPalletInventory, + event: BizPalletEvent, + ) -> BizPalletEvent: + events.append(event) + return event + + with ( + patch.object(AccessDao, 'list_user_site_ids', new=AsyncMock(return_value=[9, 9])), + patch.object( + OperationDao, + 'get_return_by_request_idempotency_key', + new=AsyncMock(return_value=None), + ), + patch.object(InventoryDao, 'get_inventories_for_update', new=AsyncMock(return_value=inventories)) as locks, + patch.object(InventoryDao, 'get_pallets_by_ids', new=AsyncMock(return_value=pallets)), + patch.object( + OperationDao, + 'get_active_filling_for_update', + new=AsyncMock(return_value=filling), + ), + patch.object(AccessService, 'assert_site_access', new=AsyncMock()) as site_access, + patch.object( + AccessService, + 'assert_warehouse_access', + new=AsyncMock(), + ) as 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', + ) + ), + ), + patch.object( + MasterDao, + 'get_warehouse', + new=AsyncMock( + return_value=BizWarehouse( + warehouse_id=7, + warehouse_code='W-7', + warehouse_name='七号仓', + status='0', + ) + ), + ), + patch.object(OperationDao, 'add_return_order', new=AsyncMock(side_effect=add_order)), + patch.object(OperationDao, 'add_return_items', new=AsyncMock(side_effect=add_items)), + patch.object(InventoryDao, 'add_event', new=AsyncMock(side_effect=add_event)), + patch.object(return_service_module, 'datetime') as service_datetime, + ): + service_datetime.now.return_value = self.fixed_time + result = await ReturnService.request_return_services(self.db, self.page, self.user) + + self.assertTrue(result.is_success) + self.assertEqual(13, result.result) + locks.assert_awaited_once_with(self.db, [2, 5]) + site_access.assert_awaited_with(self.db, self.user, 9, for_update=True) + warehouse_access.assert_not_awaited() + self.assertEqual(ReturnStatus.REQUESTED.value, captured['order'].status) + self.assertEqual( + [(2, 102, 8), (5, None, 0)], + [ + (item.pallet_id, item.active_filling_id_snapshot, item.remaining_packages_snapshot) + for item in captured['items'] + ], + ) + self.assertEqual([PalletFlowState.RETURNING.value] * 2, [item.flow_state for item in inventories]) + self.assertEqual([4, 4], [item.current_line_id for item in inventories]) + self.assertEqual([4, 4], [item.version for item in inventories]) + self.assertEqual([PalletEventType.RETURN_REQUESTED.value] * 2, [event.event_type for event in events]) + self.assertEqual([9, 9], [event.after_site_id for event in events]) + self.assertEqual([4, 4], [event.after_line_id for event in events]) + self.db.commit.assert_awaited_once() + + async def test_request_rejects_mixed_site_and_rolls_back_whole_order(self) -> None: + inventories = [make_inventory(2), make_inventory(5, site_id=10)] + with ( + patch.object(AccessDao, 'list_user_site_ids', new=AsyncMock(return_value=[9])), + patch.object( + OperationDao, + 'get_return_by_request_idempotency_key', + new=AsyncMock(return_value=None), + ), + patch.object(AccessService, 'assert_site_access', new=AsyncMock()), + 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=[make_filling(2), make_filling(5)]), + ), + patch.object(OperationDao, 'add_return_order', new=AsyncMock()) as add_order, + self.assertRaises(ServiceException) as raised, + ): + await ReturnService.request_return_services(self.db, self.page, self.user) + + self.assertIn('同一现场', raised.exception.message) + add_order.assert_not_awaited() + self.db.commit.assert_not_awaited() + self.db.rollback.assert_awaited() + + async def test_request_same_key_same_hash_returns_original_and_different_hash_conflicts(self) -> None: + request_hash = ReturnService._build_request_hash(self.page) + exact = make_return_order(request_hash=request_hash) + with ( + patch.object(AccessDao, 'list_user_site_ids', new=AsyncMock(return_value=[9])), + patch.object( + OperationDao, + 'get_return_by_request_idempotency_key', + new=AsyncMock(return_value=exact), + ), + patch.object(OperationDao, 'get_return_for_update', new=AsyncMock(return_value=exact)), + patch.object(AccessService, 'assert_site_access', new=AsyncMock()), + patch.object(MasterDao, 'get_site', new=AsyncMock(return_value=SimpleNamespace(status='0'))), + patch.object(InventoryDao, 'get_inventories_for_update', new=AsyncMock()) as locks, + ): + result = await ReturnService.request_return_services(self.db, self.page, self.user) + self.assertEqual(13, result.result) + locks.assert_not_awaited() + self.db.commit.assert_not_awaited() + + conflict = make_return_order(request_hash='different') + conflict_db = MagicMock(spec=AsyncSession) + with ( + patch.object(AccessDao, 'list_user_site_ids', new=AsyncMock(return_value=[9])), + patch.object( + OperationDao, + 'get_return_by_request_idempotency_key', + new=AsyncMock(return_value=conflict), + ), + patch.object(AccessService, 'assert_site_access', new=AsyncMock()), + self.assertRaises(ServiceException) as raised, + ): + await ReturnService.request_return_services(conflict_db, self.page, self.user) + self.assertIn('幂等键已被其他请求使用', raised.exception.message) + conflict_db.commit.assert_not_awaited() + conflict_db.rollback.assert_awaited() + + async def test_request_checks_site_mapping_before_locking_pallets(self) -> None: + with ( + patch.object(AccessDao, 'list_user_site_ids', new=AsyncMock(return_value=[9])), + patch.object( + OperationDao, + 'get_return_by_request_idempotency_key', + new=AsyncMock(return_value=None), + ), + patch.object( + AccessService, + 'assert_site_access', + new=AsyncMock(side_effect=PermissionException(data='', message='无权访问该客户现场')), + ), + patch.object( + InventoryDao, + 'get_inventories_for_update', + new=AsyncMock(return_value=[]), + ) as pallet_locks, + patch.object(InventoryDao, 'get_pallets_by_ids', new=AsyncMock(return_value=[])), + ): + try: + await ReturnService.request_return_services(self.db, self.page, self.user) + except Exception as exc: + raised = exc + else: + raised = None + + self.assertIsInstance(raised, PermissionException) + self.assertEqual('无权访问该客户现场', raised.message) + pallet_locks.assert_not_awaited() + self.db.rollback.assert_awaited() + + async def test_admin_request_idempotency_scope_is_unrestricted(self) -> None: + admin = make_user(1) + request_hash = ReturnService._build_request_hash(self.page) + exact = make_return_order(request_hash=request_hash) + get_by_key = AsyncMock(return_value=exact) + with ( + patch.object(OperationDao, 'get_return_by_request_idempotency_key', new=get_by_key), + patch.object(OperationDao, 'get_return_for_update', new=AsyncMock(return_value=exact)), + patch.object(AccessService, 'assert_site_access', new=AsyncMock()), + patch.object(MasterDao, 'get_site', new=AsyncMock(return_value=SimpleNamespace(status='0'))), + ): + result = await ReturnService.request_return_services(self.db, self.page, admin) + + self.assertEqual(13, result.result) + get_by_key.assert_awaited_once_with( + self.db, + 'request-key', + site_ids=None, + warehouse_ids=None, + ) + + +@unittest.skipUnless(ReturnService is not None, 'ReturnService 尚未实现') +class ReturnConfirmTransactionRedTest(unittest.IsolatedAsyncioTestCase): + def setUp(self) -> None: + self.db = MagicMock(spec=AsyncSession) + self.user = make_user() + self.fixed_time = datetime(2026, 7, 25, 11, 45) + self.page = ReturnConfirmModel(scannedPalletIds=[5, 2], idempotencyKey='confirm-key') + + async def test_confirm_moves_residual_and_empty_pallets_and_audits_order(self) -> None: + order = make_return_order() + inventories = [ + make_inventory(2, state=PalletFlowState.RETURNING.value), + make_inventory(5, remaining=0, state=PalletFlowState.RETURNING.value), + ] + pallets = [make_pallet(2), make_pallet(5)] + filling = make_filling(2, warehouse_id=3) + events: list[BizPalletEvent] = [] + + async def add_event( + _db: AsyncSession, + _inventory: BizPalletInventory, + event: BizPalletEvent, + ) -> BizPalletEvent: + events.append(event) + return event + + with ( + patch.object(AccessDao, 'list_user_warehouse_ids', new=AsyncMock(return_value=[7, 7])), + patch.object(OperationDao, 'get_return_by_id', new=AsyncMock(return_value=order)), + patch.object( + OperationDao, + 'get_return_by_confirm_idempotency_key', + new=AsyncMock(return_value=None), + ), + patch.object(OperationDao, 'get_return_for_update', new=AsyncMock(return_value=order)) as lock_order, + patch.object(InventoryDao, 'get_inventories_for_update', new=AsyncMock(return_value=inventories)) as locks, + patch.object(InventoryDao, 'get_pallets_by_ids', new=AsyncMock(return_value=pallets)), + patch.object( + OperationDao, + 'get_active_filling_for_update', + new=AsyncMock(return_value=filling), + ), + patch.object(AccessService, 'assert_warehouse_access', new=AsyncMock()) as warehouse_access, + patch.object( + MasterDao, + 'get_warehouse', + new=AsyncMock( + return_value=BizWarehouse( + warehouse_id=7, + warehouse_code='W-7', + warehouse_name='七号仓', + status='0', + ) + ), + ), + patch.object(OperationDao, 'update_return_order', new=AsyncMock()), + patch.object(OperationDao, 'update_filling_record', new=AsyncMock()) as update_filling, + patch.object(InventoryDao, 'add_event', new=AsyncMock(side_effect=add_event)), + patch.object(return_service_module, 'datetime') as service_datetime, + ): + service_datetime.now.return_value = self.fixed_time + result = await ReturnService.confirm_return_services(self.db, 13, self.page, self.user) + + self.assertTrue(result.is_success) + self.assertEqual(13, result.result) + lock_order.assert_awaited_once_with(self.db, 13, site_ids=[], warehouse_ids=[7]) + locks.assert_awaited_once_with(self.db, [2, 5]) + warehouse_access.assert_awaited_with(self.db, self.user, 7, for_update=True) + self.assertEqual(PalletFlowState.FILLED_IN_WAREHOUSE.value, inventories[0].flow_state) + self.assertEqual(102, inventories[0].active_filling_id) + self.assertEqual('ACTIVE', filling.active_slot) + self.assertEqual(7, filling.warehouse_id) + update_filling.assert_awaited_once_with(self.db, filling) + self.assertEqual(PalletFlowState.EMPTY_IN_WAREHOUSE.value, inventories[1].flow_state) + self.assertIsNone(inventories[1].active_filling_id) + for inventory in inventories: + self.assertEqual(7, inventory.warehouse_id) + self.assertIsNone(inventory.site_id) + self.assertIsNone(inventory.current_line_id) + self.assertEqual(4, inventory.version) + self.assertEqual(ReturnStatus.CONFIRMED.value, order.status) + self.assertEqual('confirm-key', order.confirm_idempotency_key) + self.assertEqual(7, order.confirmed_by) + self.assertEqual(self.fixed_time, order.confirmed_at) + self.assertEqual([PalletEventType.RETURN_CONFIRMED.value] * 2, [event.event_type for event in events]) + self.assertEqual([7, 7], [event.after_warehouse_id for event in events]) + self.assertEqual([None, None], [event.after_site_id for event in events]) + self.db.commit.assert_awaited_once() + + async def test_confirm_preflights_idempotency_key_before_return_id(self) -> None: + call_order: list[str] = [] + order = make_return_order() + + async def get_by_key(*_args: Any, **_kwargs: Any) -> None: + call_order.append('key') + + async def get_by_id(*_args: Any, **_kwargs: Any) -> BizReturnOrder: + call_order.append('return') + return order + + with ( + patch.object(AccessDao, 'list_user_warehouse_ids', new=AsyncMock(return_value=[7])), + patch.object(OperationDao, 'get_return_by_confirm_idempotency_key', new=AsyncMock(side_effect=get_by_key)), + patch.object(OperationDao, 'get_return_by_id', new=AsyncMock(side_effect=get_by_id)), + patch.object(AccessService, 'assert_warehouse_access', new=AsyncMock()), + patch.object(OperationDao, 'get_return_for_update', new=AsyncMock(return_value=order)), + patch.object(InventoryDao, 'get_inventories_for_update', new=AsyncMock(return_value=[])), + patch.object(InventoryDao, 'get_pallets_by_ids', new=AsyncMock(return_value=[])), + ): + try: + await ReturnService.confirm_return_services(self.db, 13, self.page, self.user) + except ServiceException: + pass + + self.assertEqual(['key', 'return'], call_order[:2]) + + async def test_confirm_requires_exact_scan_set_before_inventory_locks(self) -> None: + order = make_return_order() + for scanned_ids in ([2], [2, 5, 7], [2, 7]): + with self.subTest(scanned_ids=scanned_ids): + db = MagicMock(spec=AsyncSession) + page = ReturnConfirmModel(scannedPalletIds=scanned_ids, idempotencyKey='confirm-key') + with ( + patch.object(AccessDao, 'list_user_warehouse_ids', new=AsyncMock(return_value=[7])), + patch.object(OperationDao, 'get_return_by_id', new=AsyncMock(return_value=order)), + patch.object(AccessService, 'assert_warehouse_access', new=AsyncMock()), + patch.object( + OperationDao, + 'get_return_by_confirm_idempotency_key', + new=AsyncMock(return_value=None), + ), + patch.object(OperationDao, 'get_return_for_update', new=AsyncMock(return_value=order)), + patch.object(InventoryDao, 'get_inventories_for_update', new=AsyncMock()) as locks, + self.assertRaises(ServiceException) as raised, + ): + await ReturnService.confirm_return_services(db, 13, page, self.user) + self.assertIn('扫码', raised.exception.message) + locks.assert_not_awaited() + db.rollback.assert_awaited() + + async def test_confirm_rejects_each_changed_snapshot_and_rolls_back(self) -> None: + scenarios = ( + ('remaining', make_inventory(2, remaining=7, state=PalletFlowState.RETURNING.value)), + ('filling', make_inventory(2, filling_id=999, state=PalletFlowState.RETURNING.value)), + ('state', make_inventory(2, state=PalletFlowState.AT_SITE_PARTIAL.value)), + ) + order = make_return_order(snapshots=((2, 102, 8),)) + page = ReturnConfirmModel(scannedPalletIds=[2], idempotencyKey='confirm-key') + for scenario, inventory in scenarios: + with self.subTest(scenario=scenario): + db = MagicMock(spec=AsyncSession) + with ( + patch.object(AccessDao, 'list_user_warehouse_ids', new=AsyncMock(return_value=[7])), + patch.object(OperationDao, 'get_return_by_id', new=AsyncMock(return_value=order)), + patch.object(AccessService, 'assert_warehouse_access', new=AsyncMock()), + patch.object( + OperationDao, + 'get_return_by_confirm_idempotency_key', + new=AsyncMock(return_value=None), + ), + patch.object(OperationDao, 'get_return_for_update', new=AsyncMock(return_value=order)), + 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, 'update_return_order', new=AsyncMock()) as update_order, + self.assertRaises(ServiceException) as raised, + ): + await ReturnService.confirm_return_services(db, 13, page, self.user) + self.assertTrue('快照' in raised.exception.message or '状态' in raised.exception.message) + update_order.assert_not_awaited() + db.commit.assert_not_awaited() + db.rollback.assert_awaited() + + async def test_confirm_same_key_same_request_is_idempotent_and_different_request_conflicts(self) -> None: + confirm_hash = ReturnService._build_confirm_request_hash(13, self.page) + exact = make_return_order( + status=ReturnStatus.CONFIRMED.value, + confirm_key='confirm-key', + confirm_hash=confirm_hash, + ) + with ( + patch.object(AccessDao, 'list_user_warehouse_ids', new=AsyncMock(return_value=[7])), + patch.object(OperationDao, 'get_return_by_id', new=AsyncMock(return_value=exact)), + patch.object( + OperationDao, + 'get_return_by_confirm_idempotency_key', + new=AsyncMock(return_value=exact), + ), + patch.object(OperationDao, 'get_return_for_update', new=AsyncMock(return_value=exact)), + patch.object(AccessService, 'assert_warehouse_access', new=AsyncMock()), + patch.object(MasterDao, 'get_warehouse', new=AsyncMock(return_value=SimpleNamespace(status='0'))), + patch.object(InventoryDao, 'get_inventories_for_update', new=AsyncMock()) as locks, + ): + result = await ReturnService.confirm_return_services(self.db, 13, self.page, self.user) + self.assertEqual(13, result.result) + locks.assert_not_awaited() + self.db.commit.assert_not_awaited() + self.db.rollback.assert_awaited() + + conflict_page = ReturnConfirmModel(scannedPalletIds=[2], idempotencyKey='confirm-key') + conflict_db = MagicMock(spec=AsyncSession) + with ( + patch.object(AccessDao, 'list_user_warehouse_ids', new=AsyncMock(return_value=[7])), + patch.object(OperationDao, 'get_return_by_id', new=AsyncMock(return_value=exact)), + patch.object(AccessService, 'assert_warehouse_access', new=AsyncMock()), + patch.object( + OperationDao, + 'get_return_by_confirm_idempotency_key', + new=AsyncMock(return_value=exact), + ), + self.assertRaises(ServiceException) as raised, + ): + await ReturnService.confirm_return_services(conflict_db, 13, conflict_page, self.user) + self.assertIn('幂等键已被其他请求使用', raised.exception.message) + conflict_db.rollback.assert_awaited() + + async def test_admin_confirm_scope_is_unrestricted(self) -> None: + admin = make_user(1) + confirm_hash = ReturnService._build_confirm_request_hash(13, self.page) + exact = make_return_order( + status=ReturnStatus.CONFIRMED.value, + confirm_key='confirm-key', + confirm_hash=confirm_hash, + ) + get_order = AsyncMock(return_value=exact) + with ( + patch.object(OperationDao, 'get_return_by_id', new=get_order), + patch.object( + OperationDao, + 'get_return_by_confirm_idempotency_key', + new=AsyncMock(return_value=exact), + ), + patch.object(OperationDao, 'get_return_for_update', new=AsyncMock(return_value=exact)), + patch.object(AccessService, 'assert_warehouse_access', new=AsyncMock()), + patch.object(MasterDao, 'get_warehouse', new=AsyncMock(return_value=SimpleNamespace(status='0'))), + ): + result = await ReturnService.confirm_return_services(self.db, 13, self.page, admin) + + self.assertEqual(13, result.result) + get_order.assert_awaited_once_with( + self.db, + 13, + site_ids=None, + warehouse_ids=None, + ) + + async def test_confirm_rechecks_warehouse_mapping_before_business_locks(self) -> None: + order = make_return_order() + with ( + patch.object(AccessDao, 'list_user_warehouse_ids', new=AsyncMock(return_value=[7])), + patch.object(OperationDao, 'get_return_by_id', new=AsyncMock(return_value=order)), + patch.object( + AccessService, + 'assert_warehouse_access', + new=AsyncMock(side_effect=PermissionException(data='', message='无权访问该仓库')), + ), + patch.object( + OperationDao, + 'get_return_by_confirm_idempotency_key', + new=AsyncMock(return_value=None), + ), + patch.object(OperationDao, 'get_return_for_update', new=AsyncMock(return_value=order)) as order_lock, + patch.object( + InventoryDao, + 'get_inventories_for_update', + new=AsyncMock(return_value=[]), + ) as pallet_locks, + patch.object(InventoryDao, 'get_pallets_by_ids', new=AsyncMock(return_value=[])), + ): + try: + await ReturnService.confirm_return_services(self.db, 13, self.page, self.user) + except Exception as exc: + raised = exc + else: + raised = None + + self.assertIsInstance(raised, PermissionException) + self.assertEqual('无权访问该退回单', raised.message) + order_lock.assert_not_awaited() + pallet_locks.assert_not_awaited() + self.db.rollback.assert_awaited()