From 6f22cf028283a4b83dc2735207551bfaf9882baf 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 15:19:25 +0800 Subject: [PATCH] =?UTF-8?q?=E5=8A=9F=E8=83=BD=EF=BC=9A=E5=AE=9E=E7=8E=B0?= =?UTF-8?q?=E7=A9=BA=E6=89=98=E7=9B=98=E9=A6=96=E6=AC=A1=E5=85=A5=E5=BA=93?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../controller/inbound_controller.py | 197 +++ .../module_business/dao/operation_dao.py | 26 +- .../service/inbound_service.py | 579 ++++++++ .../module_business/test_inbound_service.py | 1210 +++++++++++++++++ .../module_business/test_inventory_dao.py | 104 +- 5 files changed, 2106 insertions(+), 10 deletions(-) create mode 100644 shierfeng-fastapi-backend/module_business/controller/inbound_controller.py create mode 100644 shierfeng-fastapi-backend/module_business/service/inbound_service.py create mode 100644 shierfeng-fastapi-backend/tests/module_business/test_inbound_service.py diff --git a/shierfeng-fastapi-backend/module_business/controller/inbound_controller.py b/shierfeng-fastapi-backend/module_business/controller/inbound_controller.py new file mode 100644 index 0000000..1c2e14d --- /dev/null +++ b/shierfeng-fastapi-backend/module_business/controller/inbound_controller.py @@ -0,0 +1,197 @@ +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, + InboundCreateModel, + InboundModel, + InboundPageQueryModel, + InboundScanQueryModel, + InboundUpdateModel, + PalletScanResponseModel, +) +from module_business.service.inbound_service import InboundService +from utils.response_util import ResponseUtil + +inbound_controller = APIRouterPro( + prefix='/business/inbound', + tags=['业务管理-仓库入库'], + dependencies=[PreAuthDependency()], +) + + +@inbound_controller.get( + '/list', + summary='获取首次入库单分页列表接口', + description='按当前用户仓库数据范围获取首次入库单分页列表', + response_model=PageResponseModel[InboundModel], + dependencies=[UserInterfaceAuthDependency('business:inbound:list')], +) +async def get_business_inbound_list( + request: Request, + inbound_page_query: Annotated[InboundPageQueryModel, Query()], + query_db: Annotated[AsyncSession, DBSessionDependency()], + current_user: Annotated[CurrentUserModel, CurrentUserDependency()], +) -> Response: + result = await InboundService.get_inbound_list_services( + query_db, + inbound_page_query, + current_user, + is_page=True, + ) + return ResponseUtil.success(model_content=result) + + +@inbound_controller.get( + '/warehouse-options', + summary='获取首次入库仓库选项接口', + description='仅返回当前用户有权访问的启用仓库', + response_model=DataResponseModel[list[BusinessOptionModel]], + dependencies=[UserInterfaceAuthDependency('business:inbound:query')], +) +async def get_business_inbound_warehouse_options( + request: Request, + query_db: Annotated[AsyncSession, DBSessionDependency()], + current_user: Annotated[CurrentUserModel, CurrentUserDependency()], +) -> Response: + result = await InboundService.get_warehouse_options_services(query_db, current_user) + return ResponseUtil.success(data=result) + + +@inbound_controller.get( + '/scan/{qr_token}', + summary='扫描首次入库托盘接口', + description='查询托盘是否可加入所选仓库的首次入库单', + response_model=DataResponseModel[PalletScanResponseModel], + dependencies=[UserInterfaceAuthDependency('business:inbound:query')], +) +async def scan_business_inbound_pallet( + request: Request, + qr_token: Annotated[ + str, + Path(min_length=1, max_length=128, pattern=r'\S', description='二维码固定标识'), + ], + inbound_scan_query: Annotated[InboundScanQueryModel, Query()], + query_db: Annotated[AsyncSession, DBSessionDependency()], + current_user: Annotated[CurrentUserModel, CurrentUserDependency()], +) -> Response: + result = await InboundService.scan_inbound_pallet_services( + query_db, + qr_token, + inbound_scan_query, + current_user, + ) + return ResponseUtil.success(data=result) + + +@inbound_controller.get( + '/{inbound_id}', + summary='获取首次入库单详情接口', + description='按仓库数据范围获取指定首次入库单详情', + response_model=DataResponseModel[InboundModel], + dependencies=[UserInterfaceAuthDependency('business:inbound:query')], +) +async def query_detail_business_inbound( + request: Request, + inbound_id: Annotated[int, Path(gt=0, description='入库单ID')], + query_db: Annotated[AsyncSession, DBSessionDependency()], + current_user: Annotated[CurrentUserModel, CurrentUserDependency()], +) -> Response: + result = await InboundService.inbound_detail_services(query_db, inbound_id, current_user) + return ResponseUtil.success(data=result) + + +@inbound_controller.post( + '', + summary='新增首次入库草稿接口', + description='新增空托盘首次入库草稿', + response_model=ResponseBaseModel, + dependencies=[UserInterfaceAuthDependency('business:inbound:add')], +) +@Log(title='仓库入库', business_type=BusinessType.INSERT) +async def add_business_inbound( + request: Request, + add_inbound: InboundCreateModel, + query_db: Annotated[AsyncSession, DBSessionDependency()], + current_user: Annotated[CurrentUserModel, CurrentUserDependency()], +) -> Response: + result = await InboundService.add_inbound_services(query_db, add_inbound, current_user) + return ResponseUtil.success(msg=result.message) + + +@inbound_controller.put( + '', + summary='编辑首次入库草稿接口', + description='编辑尚未确认的首次入库草稿', + response_model=ResponseBaseModel, + dependencies=[UserInterfaceAuthDependency('business:inbound:edit')], +) +@Log(title='仓库入库', business_type=BusinessType.UPDATE) +async def edit_business_inbound( + request: Request, + edit_inbound: InboundUpdateModel, + query_db: Annotated[AsyncSession, DBSessionDependency()], + current_user: Annotated[CurrentUserModel, CurrentUserDependency()], +) -> Response: + result = await InboundService.edit_inbound_services(query_db, edit_inbound, current_user) + return ResponseUtil.success(msg=result.message) + + +@inbound_controller.delete( + '/{inbound_ids}', + summary='删除首次入库草稿接口', + description='原子删除一个或多个尚未确认的首次入库草稿', + response_model=ResponseBaseModel, + dependencies=[UserInterfaceAuthDependency('business:inbound:remove')], +) +@Log(title='仓库入库', business_type=BusinessType.DELETE) +async def delete_business_inbound( + request: Request, + inbound_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 InboundService.delete_inbound_services( + query_db, + [int(inbound_id) for inbound_id in inbound_ids.split(',')], + current_user, + ) + return ResponseUtil.success(msg=result.message) + + +@inbound_controller.put( + '/{inbound_id}/confirm', + summary='确认首次入库单接口', + description='原子确认整张首次入库单并追加托盘流转事件', + response_model=ResponseBaseModel, + dependencies=[UserInterfaceAuthDependency('business:inbound:confirm')], +) +@Log(title='仓库入库', business_type=BusinessType.UPDATE) +async def confirm_business_inbound( + request: Request, + inbound_id: Annotated[int, Path(gt=0, description='入库单ID')], + confirm_command: ConfirmCommandModel, + query_db: Annotated[AsyncSession, DBSessionDependency()], + current_user: Annotated[CurrentUserModel, CurrentUserDependency()], +) -> Response: + result = await InboundService.confirm_inbound_services( + query_db, + inbound_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 5a28b20..a844f6e 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, select, update +from sqlalchemy import ColumnElement, Select, delete, false, select, update from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.orm.attributes import InstrumentedAttribute @@ -186,12 +186,28 @@ class OperationDao: db: AsyncSession, inbound_id: int, ) -> BizInboundOrder | None: - return await cls._get_for_update( + inbound = await cls._get_for_update( db, BizInboundOrder, BizInboundOrder.inbound_id, inbound_id, ) + if inbound is None: + return None + + # 父单锁定后用 locking read 获取最新明细，避免 MySQL RR 继续读取确认前的旧快照。 + item_result = await db.execute( + select(BizInboundItem) + .where(BizInboundItem.inbound_id == inbound_id) + .order_by(BizInboundItem.pallet_id, BizInboundItem.inbound_item_id) + .with_for_update() + .execution_options(populate_existing=True) + ) + inbound.items = sorted( + item_result.scalars().all(), + key=lambda item: (item.pallet_id, item.inbound_item_id), + ) + return inbound @classmethod async def get_shipment_for_update( @@ -318,8 +334,14 @@ class OperationDao: db: AsyncSession, query_object: InboundPageQueryModel, is_page: bool = True, + warehouse_ids: list[int] | None = None, ) -> PageModel | list[dict[str, Any]]: conditions = [] + if warehouse_ids is not None: + scoped_warehouse_ids = sorted(set(warehouse_ids)) + conditions.append( + BizInboundOrder.warehouse_id.in_(scoped_warehouse_ids) if scoped_warehouse_ids else false() + ) if query_object.inbound_no: conditions.append(BizInboundOrder.inbound_no.like(f'%{query_object.inbound_no}%')) if query_object.warehouse_id: diff --git a/shierfeng-fastapi-backend/module_business/service/inbound_service.py b/shierfeng-fastapi-backend/module_business/service/inbound_service.py new file mode 100644 index 0000000..0435bd0 --- /dev/null +++ b/shierfeng-fastapi-backend/module_business/service/inbound_service.py @@ -0,0 +1,579 @@ +import hashlib +import json +import secrets +from datetime import datetime +from typing import Any + +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 ( + BizInboundItem, + BizInboundOrder, + BizPalletEvent, + BizPalletInventory, +) +from module_business.entity.vo.master_vo import WarehousePageQueryModel +from module_business.entity.vo.operation_vo import ( + BusinessOptionModel, + ConfirmCommandModel, + InboundCreateModel, + InboundModel, + InboundPageQueryModel, + InboundScanQueryModel, + InboundUpdateModel, + PalletScanResponseModel, +) +from module_business.service.access_service import AccessService + + +class InboundService: + """空托盘首次入库服务层。""" + + @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) + + @staticmethod + def _build_confirm_request_hash(inbound_id: int) -> str: + payload = json.dumps( + {'inbound_id': inbound_id}, + sort_keys=True, + separators=(',', ':'), + ).encode() + return hashlib.sha256(payload).hexdigest() + + @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))) + + @staticmethod + async def _validate_active_warehouse( + query_db: AsyncSession, + warehouse_id: int, + current_user: CurrentUserModel, + ) -> Any: + warehouse = await MasterDao.get_warehouse(query_db, warehouse_id) + if warehouse is None: + raise ServiceException(message='目标仓库不存在') + if warehouse.status != '0': + raise ServiceException(message='目标仓库已停用') + await AccessService.assert_warehouse_access(query_db, current_user, warehouse_id) + return warehouse + + @staticmethod + def _validate_pallet_pair(pallet: BizPallet, inventory: BizPalletInventory) -> None: + if pallet.asset_status != '0': + raise ServiceException(message='托盘资产已停用，不允许首次入库') + if ( + inventory.flow_state != PalletFlowState.UNSTOCKED.value + or inventory.warehouse_id is not None + or inventory.site_id is not None + or inventory.current_line_id is not None + or inventory.active_filling_id is not None + or inventory.remaining_packages != 0 + ): + raise ServiceException(message='托盘不符合首次入库条件') + if not isinstance(inventory.version, int) or inventory.version < 0: + raise ServiceException(message='托盘库存版本无效') + + @classmethod + async def _lock_and_validate_pallets( + cls, + query_db: AsyncSession, + pallet_ids: list[int], + ) -> tuple[list[BizPalletInventory], list[BizPallet]]: + ordered_ids = sorted(pallet_ids) + inventories = await InventoryDao.get_inventories_for_update(query_db, ordered_ids) + pallets = await InventoryDao.get_pallets_by_ids(query_db, ordered_ids) + expected_ids = set(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) != expected_ids or set(pallet_by_id) != expected_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_pallet_pair(pallet, inventory) + return ordered_inventories, ordered_pallets + + @staticmethod + def _build_items( + inbound_id: int, + pallet_ids: list[int], + user_name: str, + current_time: datetime, + ) -> list[BizInboundItem]: + return [ + BizInboundItem( + inbound_id=inbound_id, + pallet_id=pallet_id, + 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) + + @classmethod + async def get_inbound_list_services( + cls, + query_db: AsyncSession, + query_object: InboundPageQueryModel, + current_user: CurrentUserModel, + is_page: bool = True, + ) -> PageModel | list[InboundModel]: + warehouse_ids = await cls._warehouse_scope(query_db, current_user) + if ( + warehouse_ids is not None + and query_object.warehouse_id is not None + and query_object.warehouse_id not in warehouse_ids + ): + raise PermissionException(data='', message='无权访问该仓库') + + result = await OperationDao.get_inbound_list( + query_db, + query_object, + is_page, + warehouse_ids=warehouse_ids, + ) + rows = result.rows if isinstance(result, PageModel) else result + models = [InboundModel.model_validate(record) for record in rows] + if isinstance(result, PageModel): + return result.model_copy(update={'rows': models}) + return models + + @classmethod + async def inbound_detail_services( + cls, + query_db: AsyncSession, + inbound_id: int, + current_user: CurrentUserModel, + ) -> InboundModel: + inbound = await OperationDao.get_inbound_by_id(query_db, inbound_id) + if inbound is None: + raise ServiceException(message='入库单不存在') + await AccessService.assert_warehouse_access(query_db, current_user, inbound.warehouse_id) + return InboundModel.model_validate(inbound) + + @classmethod + async def get_warehouse_options_services( + cls, + query_db: AsyncSession, + current_user: CurrentUserModel, + ) -> list[BusinessOptionModel]: + warehouse_ids = await cls._warehouse_scope(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 scan_inbound_pallet_services( + cls, + query_db: AsyncSession, + qr_token: str, + query_object: InboundScanQueryModel, + current_user: CurrentUserModel, + ) -> PalletScanResponseModel: + await AccessService.assert_warehouse_access( + query_db, + current_user, + query_object.warehouse_id, + ) + warehouse = await MasterDao.get_warehouse(query_db, query_object.warehouse_id) + if warehouse is None: + raise ServiceException(message='目标仓库不存在') + if warehouse.status != '0': + raise ServiceException(message='目标仓库已停用') + + pallet = await InventoryDao.get_pallet_by_qr_token(query_db, qr_token) + if pallet is None: + raise ServiceException(message='托盘不存在') + inventory = await MasterDao.get_pallet_inventory(query_db, pallet.pallet_id) + if inventory is None: + raise ServiceException(message='托盘库存不存在') + cls._validate_pallet_pair(pallet, inventory) + return PalletScanResponseModel( + pallet_id=pallet.pallet_id, + pallet_code=pallet.pallet_code, + flow_state=PalletFlowState.UNSTOCKED, + warehouse_id=None, + site_id=None, + current_line_id=None, + active_filling_id=None, + strain_id=None, + strain_code=None, + strain_name=None, + batch_no=None, + package_spec=None, + capacity_packages=pallet.capacity_packages, + remaining_packages=0, + allowed_actions=['inbound'], + ) + + @classmethod + async def add_inbound_services( + cls, + query_db: AsyncSession, + page_object: InboundCreateModel, + current_user: CurrentUserModel, + ) -> CrudResponseModel: + try: + _, user_name = cls._current_identity(current_user) + await cls._validate_active_warehouse( + query_db, + page_object.warehouse_id, + current_user, + ) + ordered_ids = sorted(page_object.pallet_ids) + await cls._lock_and_validate_pallets(query_db, ordered_ids) + + current_time = datetime.now() + inbound = BizInboundOrder( + inbound_no=f'IN-{secrets.token_hex(16)}', + warehouse_id=page_object.warehouse_id, + status=DocumentStatus.DRAFT.value, + create_by=user_name, + create_time=current_time, + update_by=user_name, + update_time=current_time, + ) + inbound = await OperationDao.add_inbound(query_db, inbound) + await OperationDao.replace_inbound_items( + query_db, + inbound.inbound_id, + cls._build_items(inbound.inbound_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_inbound_services( + cls, + query_db: AsyncSession, + page_object: InboundUpdateModel, + current_user: CurrentUserModel, + ) -> CrudResponseModel: + try: + _, user_name = cls._current_identity(current_user) + inbound = await OperationDao.get_inbound_for_update(query_db, page_object.inbound_id) + if inbound is None: + raise ServiceException(message='入库单不存在') + await AccessService.assert_warehouse_access( + query_db, + current_user, + inbound.warehouse_id, + ) + if inbound.status != DocumentStatus.DRAFT.value: + raise ServiceException(message='已确认入库单不允许编辑') + await cls._validate_active_warehouse( + query_db, + page_object.warehouse_id, + current_user, + ) + ordered_ids = sorted(page_object.pallet_ids) + await cls._lock_and_validate_pallets(query_db, ordered_ids) + + current_time = datetime.now() + inbound.warehouse_id = page_object.warehouse_id + inbound.update_by = user_name + inbound.update_time = current_time + await OperationDao.update_inbound(query_db, inbound) + await OperationDao.replace_inbound_items( + query_db, + inbound.inbound_id, + cls._build_items(inbound.inbound_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_inbound_services( + cls, + query_db: AsyncSession, + inbound_ids: list[int], + current_user: CurrentUserModel, + ) -> CrudResponseModel: + try: + if not inbound_ids or any(inbound_id <= 0 for inbound_id in inbound_ids): + raise ServiceException(message='入库单ID必须为正整数') + if len(inbound_ids) != len(set(inbound_ids)): + raise ServiceException(message='入库单ID不能重复') + + ordered_ids = sorted(inbound_ids) + for inbound_id in ordered_ids: + inbound = await OperationDao.get_inbound_for_update(query_db, inbound_id) + if inbound is None: + raise ServiceException(message='入库单不存在') + await AccessService.assert_warehouse_access( + query_db, + current_user, + inbound.warehouse_id, + ) + if inbound.status != DocumentStatus.DRAFT.value: + raise ServiceException(message='已确认入库单不允许删除') + + for inbound_id in ordered_ids: + await OperationDao.delete_inbound(query_db, inbound_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( + inbound: BizInboundOrder, + inbound_id: int, + idempotency_key: str, + request_hash: str, + ) -> bool: + return ( + inbound.inbound_id == inbound_id + and inbound.status == DocumentStatus.CONFIRMED.value + and inbound.confirm_idempotency_key == idempotency_key + and inbound.confirm_request_hash == request_hash + ) + + @classmethod + async def _recover_confirm_integrity_error( + cls, + query_db: AsyncSession, + inbound_id: int, + idempotency_key: str, + request_hash: str, + current_user: CurrentUserModel, + ) -> CrudResponseModel: + try: + recovered = await OperationDao.get_inbound_by_confirm_idempotency_key( + query_db, + idempotency_key, + ) + if recovered is None: + raise ServiceException(message='确认入库失败，幂等冲突') + await AccessService.assert_warehouse_access( + query_db, + current_user, + recovered.warehouse_id, + ) + if not cls._is_exact_confirmed( + recovered, + inbound_id, + idempotency_key, + request_hash, + ): + raise ServiceException(message='确认入库失败，幂等键已被其他请求使用') + return cls._success('确认成功') + finally: + await query_db.rollback() + + @classmethod + async def confirm_inbound_services( # noqa: PLR0912, PLR0915 + cls, + query_db: AsyncSession, + inbound_id: int, + page_object: ConfirmCommandModel, + current_user: CurrentUserModel, + ) -> CrudResponseModel: + try: + user_id, user_name = cls._current_identity(current_user) + inbound = await OperationDao.get_inbound_by_id(query_db, inbound_id) + if inbound is None: + raise ServiceException(message='入库单不存在') + await AccessService.assert_warehouse_access( + query_db, + current_user, + inbound.warehouse_id, + ) + + request_hash = cls._build_confirm_request_hash(inbound_id) + idempotency_key = page_object.idempotency_key + existing = await OperationDao.get_inbound_by_confirm_idempotency_key( + query_db, + idempotency_key, + ) + if existing is not None: + await AccessService.assert_warehouse_access( + query_db, + current_user, + existing.warehouse_id, + ) + if not cls._is_exact_confirmed( + existing, + inbound_id, + idempotency_key, + request_hash, + ): + raise ServiceException(message='确认入库失败，幂等键已被其他请求使用') + + locked_inbound = await OperationDao.get_inbound_for_update(query_db, inbound_id) + if locked_inbound is None: + raise ServiceException(message='入库单不存在') + await AccessService.assert_warehouse_access( + query_db, + current_user, + locked_inbound.warehouse_id, + ) + if locked_inbound.status == DocumentStatus.CONFIRMED.value: + if cls._is_exact_confirmed( + locked_inbound, + inbound_id, + idempotency_key, + request_hash, + ): + await query_db.rollback() + return cls._success('确认成功') + raise ServiceException(message='入库单已确认，幂等信息不匹配') + if locked_inbound.status != DocumentStatus.DRAFT.value: + raise ServiceException(message='当前入库单状态不允许确认') + if existing is not None: + raise ServiceException(message='确认入库失败，并发状态不一致') + if locked_inbound.confirm_idempotency_key is not None or locked_inbound.confirm_request_hash is not None: + raise ServiceException(message='入库单确认信息不完整') + + pallet_ids = [item.pallet_id for item in locked_inbound.items] + if not pallet_ids: + raise ServiceException(message='入库单至少需要一个托盘') + if len(pallet_ids) != len(set(pallet_ids)): + raise ServiceException(message='入库单托盘不能重复') + ordered_ids = sorted(pallet_ids) + inventories, _ = await cls._lock_and_validate_pallets(query_db, ordered_ids) + + for inventory in inventories: + PalletStateMachine.ensure_transition( + PalletFlowState(inventory.flow_state), + PalletFlowState.EMPTY_IN_WAREHOUSE, + ) + + current_time = datetime.now() + locked_inbound.status = DocumentStatus.CONFIRMED.value + locked_inbound.confirm_idempotency_key = idempotency_key + locked_inbound.confirm_request_hash = request_hash + locked_inbound.confirmed_by = user_id + locked_inbound.confirmed_at = current_time + locked_inbound.update_by = user_name + locked_inbound.update_time = current_time + await OperationDao.update_inbound(query_db, locked_inbound) + + 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.EMPTY_IN_WAREHOUSE.value + inventory.warehouse_id = locked_inbound.warehouse_id + inventory.site_id = None + inventory.current_line_id = None + inventory.active_filling_id = None + inventory.remaining_packages = 0 + inventory.version += 1 + inventory.updated_at = current_time + event = BizPalletEvent( + pallet_id=inventory.pallet_id, + event_type=PalletEventType.INBOUND_CONFIRMED.value, + source_type='INBOUND', + source_id=inbound_id, + before_state=before_state, + after_state=PalletFlowState.EMPTY_IN_WAREHOUSE.value, + before_warehouse_id=before_warehouse_id, + before_site_id=before_site_id, + before_line_id=before_line_id, + after_warehouse_id=locked_inbound.warehouse_id, + after_site_id=None, + after_line_id=None, + before_remaining_packages=before_remaining_packages, + after_remaining_packages=0, + 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, + inbound_id, + page_object.idempotency_key, + cls._build_confirm_request_hash(inbound_id), + current_user, + ) + except Exception: + await query_db.rollback() + raise diff --git a/shierfeng-fastapi-backend/tests/module_business/test_inbound_service.py b/shierfeng-fastapi-backend/tests/module_business/test_inbound_service.py new file mode 100644 index 0000000..c27f863 --- /dev/null +++ b/shierfeng-fastapi-backend/tests/module_business/test_inbound_service.py @@ -0,0 +1,1210 @@ +import hashlib +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 inbound_controller as inbound_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 BizPallet +from module_business.entity.do.operation_do import ( + BizInboundItem, + BizInboundOrder, + BizPalletEvent, + BizPalletInventory, +) +from module_business.entity.vo.operation_vo import ( + BusinessOptionModel, + ConfirmCommandModel, + InboundCreateModel, + InboundModel, + InboundPageQueryModel, + InboundScanQueryModel, + InboundUpdateModel, + PalletScanResponseModel, +) +from module_business.service import inbound_service as inbound_service_module +from module_business.service.access_service import AccessService +from module_business.service.inbound_service import InboundService + + +def integrity_error() -> IntegrityError: + return IntegrityError('statement', {}, RuntimeError('constraint')) + + +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, *, 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=20, + asset_status=asset_status, + ) + + +def make_inventory( + pallet_id: int, + *, + flow_state: str = 'UNSTOCKED', + warehouse_id: int | None = None, + site_id: int | None = None, + current_line_id: int | None = None, + active_filling_id: int | None = None, + remaining_packages: int = 0, + version: int = 3, +) -> BizPalletInventory: + return BizPalletInventory( + pallet_id=pallet_id, + flow_state=flow_state, + warehouse_id=warehouse_id, + site_id=site_id, + current_line_id=current_line_id, + active_filling_id=active_filling_id, + remaining_packages=remaining_packages, + version=version, + ) + + +def make_inbound( + inbound_id: int = 11, + *, + warehouse_id: int = 7, + pallet_ids: tuple[int, ...] = (5, 2), + status: str = 'DRAFT', + idempotency_key: str | None = None, + request_hash: str | None = None, +) -> BizInboundOrder: + inbound = BizInboundOrder( + inbound_id=inbound_id, + inbound_no=f'IN-{inbound_id}', + warehouse_id=warehouse_id, + status=status, + confirm_idempotency_key=idempotency_key, + confirm_request_hash=request_hash, + ) + inbound.items = [ + BizInboundItem( + inbound_item_id=index, + inbound_id=inbound_id, + pallet_id=pallet_id, + ) + for index, pallet_id in enumerate(pallet_ids, start=1) + ] + return inbound + + +class TestInboundPublicContract(unittest.TestCase): + def test_service_and_controller_are_available(self) -> None: + self.assertIsNotNone(InboundService) + self.assertIsNotNone(inbound_controller_module.inbound_controller) + + def test_confirm_signature_and_types_are_exact(self) -> None: + signature = inspect.signature(InboundService.confirm_inbound_services) + self.assertEqual( + ('query_db', 'inbound_id', 'page_object', 'current_user'), + tuple(signature.parameters), + ) + hints = get_type_hints(InboundService.confirm_inbound_services) + self.assertIs(AsyncSession, hints['query_db']) + self.assertIs(int, hints['inbound_id']) + self.assertIs(ConfirmCommandModel, hints['page_object']) + self.assertIs(CurrentUserModel, hints['current_user']) + self.assertIs(CrudResponseModel, hints['return']) + + def test_confirm_hash_is_compact_sorted_sha256_and_excludes_key(self) -> None: + expected_payload = json.dumps( + {'inbound_id': 11}, + sort_keys=True, + separators=(',', ':'), + ).encode() + expected = hashlib.sha256(expected_payload).hexdigest() + + self.assertEqual(expected, InboundService._build_confirm_request_hash(11)) + self.assertNotEqual(expected, InboundService._build_confirm_request_hash(12)) + + +class TestInboundDraftService(unittest.IsolatedAsyncioTestCase): + def setUp(self) -> None: + self.db = MagicMock(spec=AsyncSession) + self.user = make_user() + self.fixed_time = datetime(2026, 7, 26, 10, 30) + + async def test_add_validates_and_locks_pallets_before_writing_draft(self) -> None: + page = InboundCreateModel(warehouseId=7, palletIds=[5, 2]) + inventories = [make_inventory(2), make_inventory(5)] + pallets = [make_pallet(2), make_pallet(5)] + actions: list[str] = [] + captured: dict[str, Any] = {} + + async def access(*_args: Any) -> None: + actions.append('access') + + 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 add_inbound(_db: AsyncSession, inbound: BizInboundOrder) -> BizInboundOrder: + actions.append('add-order') + inbound.inbound_id = 11 + captured['order'] = inbound + return inbound + + async def replace_items( + _db: AsyncSession, + inbound_id: int, + items: list[BizInboundItem], + ) -> list[BizInboundItem]: + actions.append('replace-items') + captured['inbound_id'] = inbound_id + captured['items'] = items + return items + + async def commit() -> None: + actions.append('commit') + + self.db.commit.side_effect = commit + with ( + patch.object( + MasterDao, + 'get_warehouse', + new=AsyncMock(return_value=SimpleNamespace(warehouse_id=7, status='0')), + ), + patch.object(AccessService, 'assert_warehouse_access', new=AsyncMock(side_effect=access)), + 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, 'add_inbound', new=AsyncMock(side_effect=add_inbound)), + patch.object(OperationDao, 'replace_inbound_items', new=AsyncMock(side_effect=replace_items)), + patch.object(inbound_service_module.secrets, 'token_hex', return_value='fixed-random') as token_hex, + patch.object(inbound_service_module, 'datetime') as service_datetime, + ): + service_datetime.now.return_value = self.fixed_time + result = await InboundService.add_inbound_services(self.db, page, self.user) + + self.assertEqual( + ['access', 'inventory-lock', 'pallet-lock', 'add-order', 'replace-items', 'commit'], + actions, + ) + inventory_lock.assert_awaited_once_with(self.db, [2, 5]) + pallet_lock.assert_awaited_once_with(self.db, [2, 5]) + token_hex.assert_called_once_with(16) + order = captured['order'] + self.assertEqual('IN-fixed-random', order.inbound_no) + self.assertLessEqual(len(order.inbound_no), 64) + self.assertEqual('DRAFT', order.status) + self.assertEqual('user-7', order.create_by) + self.assertEqual('user-7', order.update_by) + self.assertEqual(self.fixed_time, order.create_time) + self.assertEqual(self.fixed_time, order.update_time) + self.assertEqual(11, captured['inbound_id']) + self.assertEqual([2, 5], [item.pallet_id for item in captured['items']]) + self.assertTrue(all(item.create_by == 'user-7' for item in captured['items'])) + self.assertTrue(all(item.create_time == self.fixed_time for item in captured['items'])) + self.assertTrue(result.is_success) + self.db.rollback.assert_not_awaited() + + async def test_add_rejects_invalid_pallet_before_any_order_or_event_write(self) -> None: + page = InboundCreateModel(warehouseId=7, palletIds=[2, 5]) + invalid_inventory = make_inventory(5, warehouse_id=7) + add_order = AsyncMock() + replace_items = AsyncMock() + add_event = AsyncMock() + with ( + patch.object( + MasterDao, + 'get_warehouse', + new=AsyncMock(return_value=SimpleNamespace(warehouse_id=7, status='0')), + ), + patch.object(AccessService, 'assert_warehouse_access', new=AsyncMock()), + patch.object( + InventoryDao, + 'get_inventories_for_update', + new=AsyncMock(return_value=[make_inventory(2), invalid_inventory]), + ), + patch.object( + InventoryDao, + 'get_pallets_by_ids', + new=AsyncMock(return_value=[make_pallet(2), make_pallet(5)]), + ), + patch.object(OperationDao, 'add_inbound', new=add_order), + patch.object(OperationDao, 'replace_inbound_items', new=replace_items), + patch.object(InventoryDao, 'add_event', new=add_event), + self.assertRaises(ServiceException), + ): + await InboundService.add_inbound_services(self.db, page, self.user) + + add_order.assert_not_awaited() + replace_items.assert_not_awaited() + add_event.assert_not_awaited() + self.db.commit.assert_not_awaited() + self.db.rollback.assert_awaited_once() + + async def test_add_requires_active_authorized_warehouse(self) -> None: + page = InboundCreateModel(warehouseId=7, palletIds=[2]) + lock = AsyncMock() + cases = ( + (None, ServiceException), + (SimpleNamespace(warehouse_id=7, status='1'), ServiceException), + ) + for warehouse, exception_type in cases: + with self.subTest(warehouse=warehouse): + self.db.reset_mock() + lock.reset_mock() + with ( + patch.object(MasterDao, 'get_warehouse', new=AsyncMock(return_value=warehouse)), + patch.object(AccessService, 'assert_warehouse_access', new=AsyncMock()), + patch.object(InventoryDao, 'get_inventories_for_update', new=lock), + self.assertRaises(exception_type), + ): + await InboundService.add_inbound_services(self.db, page, self.user) + lock.assert_not_awaited() + self.db.rollback.assert_awaited_once() + + self.db.reset_mock() + lock.reset_mock() + with ( + patch.object( + MasterDao, + 'get_warehouse', + new=AsyncMock(return_value=SimpleNamespace(warehouse_id=7, status='0')), + ), + patch.object( + AccessService, + 'assert_warehouse_access', + new=AsyncMock(side_effect=PermissionException(data='', message='无权访问该仓库')), + ), + patch.object(InventoryDao, 'get_inventories_for_update', new=lock), + self.assertRaises(PermissionException), + ): + await InboundService.add_inbound_services(self.db, page, self.user) + lock.assert_not_awaited() + self.db.rollback.assert_awaited_once() + + async def test_edit_checks_original_and_target_access_then_replaces_items(self) -> None: + page = InboundUpdateModel(inboundId=11, warehouseId=8, palletIds=[5, 2]) + inbound = make_inbound(warehouse_id=7, pallet_ids=(3,)) + access = AsyncMock() + update = AsyncMock() + replace = AsyncMock() + with ( + patch.object(OperationDao, 'get_inbound_for_update', new=AsyncMock(return_value=inbound)), + patch.object(AccessService, 'assert_warehouse_access', new=access), + patch.object( + MasterDao, + 'get_warehouse', + new=AsyncMock(return_value=SimpleNamespace(warehouse_id=8, status='0')), + ), + patch.object( + InventoryDao, + 'get_inventories_for_update', + new=AsyncMock(return_value=[make_inventory(2), make_inventory(5)]), + ) as inventory_lock, + patch.object( + InventoryDao, + 'get_pallets_by_ids', + new=AsyncMock(return_value=[make_pallet(2), make_pallet(5)]), + ) as pallet_lock, + patch.object(OperationDao, 'update_inbound', new=update), + patch.object(OperationDao, 'replace_inbound_items', new=replace), + patch.object(inbound_service_module, 'datetime') as service_datetime, + ): + service_datetime.now.return_value = self.fixed_time + result = await InboundService.edit_inbound_services(self.db, page, self.user) + + self.assertEqual( + [unittest.mock.call(self.db, self.user, 7), unittest.mock.call(self.db, self.user, 8)], + access.await_args_list, + ) + inventory_lock.assert_awaited_once_with(self.db, [2, 5]) + pallet_lock.assert_awaited_once_with(self.db, [2, 5]) + self.assertEqual(8, inbound.warehouse_id) + self.assertEqual('user-7', inbound.update_by) + self.assertEqual(self.fixed_time, inbound.update_time) + update.assert_awaited_once_with(self.db, inbound) + replace.assert_awaited_once() + self.assertEqual([2, 5], [item.pallet_id for item in replace.await_args.args[2]]) + self.db.commit.assert_awaited_once() + self.assertTrue(result.is_success) + + async def test_edit_rejects_confirmed_document_without_pallet_or_order_write(self) -> None: + page = InboundUpdateModel(inboundId=11, warehouseId=8, palletIds=[2]) + update = AsyncMock() + lock_pallets = AsyncMock() + with ( + patch.object( + OperationDao, + 'get_inbound_for_update', + new=AsyncMock(return_value=make_inbound(status='CONFIRMED')), + ), + patch.object(AccessService, 'assert_warehouse_access', new=AsyncMock()), + patch.object(MasterDao, 'get_warehouse', new=AsyncMock()), + patch.object(InventoryDao, 'get_inventories_for_update', new=lock_pallets), + patch.object(OperationDao, 'update_inbound', new=update), + self.assertRaises(ServiceException), + ): + await InboundService.edit_inbound_services(self.db, page, self.user) + + lock_pallets.assert_not_awaited() + update.assert_not_awaited() + self.db.rollback.assert_awaited_once() + + async def test_batch_delete_locks_and_preflights_all_before_ordered_delete(self) -> None: + actions: list[str] = [] + + async def lock(_db: AsyncSession, inbound_id: int) -> BizInboundOrder: + actions.append(f'lock-{inbound_id}') + return make_inbound(inbound_id=inbound_id, warehouse_id=inbound_id) + + async def access(_db: AsyncSession, _user: CurrentUserModel, warehouse_id: int) -> None: + actions.append(f'access-{warehouse_id}') + + async def delete(_db: AsyncSession, inbound_id: int) -> None: + actions.append(f'delete-{inbound_id}') + + async def commit() -> None: + actions.append('commit') + + self.db.commit.side_effect = commit + with ( + patch.object(OperationDao, 'get_inbound_for_update', new=AsyncMock(side_effect=lock)), + patch.object(AccessService, 'assert_warehouse_access', new=AsyncMock(side_effect=access)), + patch.object(OperationDao, 'delete_inbound', new=AsyncMock(side_effect=delete)), + ): + result = await InboundService.delete_inbound_services(self.db, [5, 2], self.user) + + self.assertEqual( + ['lock-2', 'access-2', 'lock-5', 'access-5', 'delete-2', 'delete-5', 'commit'], + actions, + ) + self.assertTrue(result.is_success) + self.db.rollback.assert_not_awaited() + + async def test_batch_delete_is_atomic_when_later_document_is_confirmed(self) -> None: + delete = AsyncMock() + records = { + 2: make_inbound(inbound_id=2, warehouse_id=2), + 5: make_inbound(inbound_id=5, warehouse_id=5, status='CONFIRMED'), + } + with ( + patch.object( + OperationDao, + 'get_inbound_for_update', + new=AsyncMock(side_effect=lambda _db, inbound_id: records[inbound_id]), + ), + patch.object(AccessService, 'assert_warehouse_access', new=AsyncMock()), + patch.object(OperationDao, 'delete_inbound', new=delete), + self.assertRaises(ServiceException), + ): + await InboundService.delete_inbound_services(self.db, [5, 2], self.user) + + delete.assert_not_awaited() + self.db.commit.assert_not_awaited() + self.db.rollback.assert_awaited_once() + + async def test_all_draft_mutations_convert_integrity_errors(self) -> None: + add_page = InboundCreateModel(warehouseId=7, palletIds=[2]) + with ( + patch.object( + MasterDao, + 'get_warehouse', + new=AsyncMock(return_value=SimpleNamespace(warehouse_id=7, status='0')), + ), + patch.object(AccessService, 'assert_warehouse_access', new=AsyncMock()), + patch.object( + InventoryDao, + 'get_inventories_for_update', + new=AsyncMock(return_value=[make_inventory(2)]), + ), + patch.object(InventoryDao, 'get_pallets_by_ids', new=AsyncMock(return_value=[make_pallet(2)])), + patch.object(OperationDao, 'add_inbound', new=AsyncMock(side_effect=integrity_error())), + self.assertRaises(ServiceException), + ): + await InboundService.add_inbound_services(self.db, add_page, self.user) + self.db.rollback.assert_awaited_once() + + self.db.reset_mock() + with ( + patch.object( + OperationDao, + 'get_inbound_for_update', + new=AsyncMock(return_value=make_inbound(pallet_ids=(2,))), + ), + patch.object(AccessService, 'assert_warehouse_access', new=AsyncMock()), + patch.object(OperationDao, 'delete_inbound', new=AsyncMock(side_effect=integrity_error())), + self.assertRaises(ServiceException), + ): + await InboundService.delete_inbound_services(self.db, [11], self.user) + self.db.rollback.assert_awaited_once() + + +class TestInboundReadService(unittest.IsolatedAsyncioTestCase): + def setUp(self) -> None: + self.db = MagicMock(spec=AsyncSession) + self.user = make_user() + + async def test_list_passes_normal_user_scope_into_paginated_dao(self) -> None: + query = InboundPageQueryModel(pageNum=2, pageSize=5) + page = PageModel[Any]( + rows=[make_inbound()], + pageNum=2, + pageSize=5, + total=1, + hasNext=False, + ) + get_list = AsyncMock(return_value=page) + with ( + patch.object(AccessDao, 'list_user_warehouse_ids', new=AsyncMock(return_value=[7, 3, 7])) as scope, + patch.object(OperationDao, 'get_inbound_list', new=get_list), + ): + actual = await InboundService.get_inbound_list_services(self.db, query, self.user, is_page=True) + + scope.assert_awaited_once_with(self.db, 7) + get_list.assert_awaited_once_with(self.db, query, True, warehouse_ids=[3, 7]) + self.assertIsInstance(actual, PageModel) + self.assertIsInstance(actual.rows[0], InboundModel) + + async def test_admin_list_uses_none_scope_and_does_not_query_assignments(self) -> None: + query = InboundPageQueryModel() + page = PageModel[Any](rows=[], pageNum=1, pageSize=10, total=0, hasNext=False) + scope = AsyncMock() + get_list = AsyncMock(return_value=page) + with ( + patch.object(AccessDao, 'list_user_warehouse_ids', new=scope), + patch.object(OperationDao, 'get_inbound_list', new=get_list), + ): + await InboundService.get_inbound_list_services(self.db, query, make_user(1), is_page=True) + + scope.assert_not_awaited() + get_list.assert_awaited_once_with(self.db, query, True, warehouse_ids=None) + + async def test_explicit_unauthorized_warehouse_is_403_before_query(self) -> None: + query = InboundPageQueryModel(warehouseId=9) + get_list = AsyncMock() + with ( + patch.object(AccessDao, 'list_user_warehouse_ids', new=AsyncMock(return_value=[3, 7])), + patch.object(OperationDao, 'get_inbound_list', new=get_list), + self.assertRaises(PermissionException), + ): + await InboundService.get_inbound_list_services(self.db, query, self.user, is_page=True) + get_list.assert_not_awaited() + + async def test_user_without_scope_gets_empty_paginated_result_without_total_leak(self) -> None: + query = InboundPageQueryModel(pageNum=4, pageSize=6) + empty_page = PageModel[Any](rows=[], pageNum=4, pageSize=6, total=0, hasNext=False) + get_list = AsyncMock(return_value=empty_page) + with ( + patch.object(AccessDao, 'list_user_warehouse_ids', new=AsyncMock(return_value=[])), + patch.object(OperationDao, 'get_inbound_list', new=get_list), + ): + actual = await InboundService.get_inbound_list_services(self.db, query, self.user, is_page=True) + + get_list.assert_awaited_once_with(self.db, query, True, warehouse_ids=[]) + self.assertEqual([], actual.rows) + self.assertEqual(0, actual.total) + self.assertFalse(actual.has_next) + + async def test_detail_loads_then_checks_warehouse_scope(self) -> None: + inbound = make_inbound() + access = AsyncMock() + with ( + patch.object(OperationDao, 'get_inbound_by_id', new=AsyncMock(return_value=inbound)), + patch.object(AccessService, 'assert_warehouse_access', new=access), + ): + actual = await InboundService.inbound_detail_services(self.db, 11, self.user) + + access.assert_awaited_once_with(self.db, self.user, 7) + self.assertIsInstance(actual, InboundModel) + + async def test_missing_detail_fails_stably_without_access_probe(self) -> None: + access = AsyncMock() + with ( + patch.object(OperationDao, 'get_inbound_by_id', new=AsyncMock(return_value=None)), + patch.object(AccessService, 'assert_warehouse_access', new=access), + self.assertRaises(ServiceException) as raised, + ): + await InboundService.inbound_detail_services(self.db, 404, self.user) + self.assertEqual('入库单不存在', raised.exception.message) + access.assert_not_awaited() + + async def test_warehouse_options_filter_disabled_and_unauthorized_records(self) -> None: + warehouses = [ + SimpleNamespace(warehouse_id=2, warehouse_code='W2', warehouse_name='二仓', status='0'), + SimpleNamespace(warehouse_id=3, warehouse_code='W3', warehouse_name='三仓', status='1'), + SimpleNamespace(warehouse_id=5, warehouse_code='W5', warehouse_name='五仓', status='0'), + ] + with ( + patch.object(MasterDao, 'list_warehouse', new=AsyncMock(return_value=warehouses)) as list_warehouse, + patch.object(AccessDao, 'list_user_warehouse_ids', new=AsyncMock(return_value=[2, 3])), + ): + actual = await InboundService.get_warehouse_options_services(self.db, self.user) + + query = list_warehouse.await_args.args[1] + self.assertEqual('0', query.status) + self.assertIs(False, list_warehouse.await_args.args[2]) + self.assertEqual([BusinessOptionModel(id=2, code='W2', name='二仓')], actual) + self.assertEqual({'id', 'code', 'name'}, set(BusinessOptionModel.model_fields)) + + async def test_admin_warehouse_options_include_all_active_records(self) -> None: + warehouses = [ + SimpleNamespace(warehouse_id=2, warehouse_code='W2', warehouse_name='二仓', status='0'), + SimpleNamespace(warehouse_id=5, warehouse_code='W5', warehouse_name='五仓', status='0'), + ] + scope = AsyncMock() + with ( + patch.object(MasterDao, 'list_warehouse', new=AsyncMock(return_value=warehouses)), + patch.object(AccessDao, 'list_user_warehouse_ids', new=scope), + ): + actual = await InboundService.get_warehouse_options_services(self.db, make_user(1)) + + scope.assert_not_awaited() + self.assertEqual([2, 5], [option.id for option in actual]) + + async def test_warehouse_options_accept_real_page_util_camel_case_rows(self) -> None: + warehouses = [ + { + 'warehouseId': 2, + 'warehouseCode': 'W2', + 'warehouseName': '二仓', + 'status': '0', + } + ] + with ( + patch.object(MasterDao, 'list_warehouse', new=AsyncMock(return_value=warehouses)), + patch.object(AccessDao, 'list_user_warehouse_ids', new=AsyncMock(return_value=[2])), + ): + actual = await InboundService.get_warehouse_options_services(self.db, self.user) + + self.assertEqual([BusinessOptionModel(id=2, code='W2', name='二仓')], actual) + + +class TestInboundScanService(unittest.IsolatedAsyncioTestCase): + def setUp(self) -> None: + self.db = MagicMock(spec=AsyncSession) + self.user = make_user() + self.query = InboundScanQueryModel(warehouseId=7) + + async def test_scan_returns_only_unstocked_inbound_action_without_lock_or_write(self) -> None: + pallet = make_pallet(2) + inventory = make_inventory(2) + with ( + patch.object(AccessService, 'assert_warehouse_access', new=AsyncMock()) as access, + patch.object( + MasterDao, + 'get_warehouse', + new=AsyncMock(return_value=SimpleNamespace(warehouse_id=7, status='0')), + ), + 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(InventoryDao, 'get_inventory_for_update', new=AsyncMock()) as locking_read, + patch.object(InventoryDao, 'add_event', new=AsyncMock()) as add_event, + ): + actual = await InboundService.scan_inbound_pallet_services( + self.db, + 'persisted-token', + self.query, + self.user, + ) + + access.assert_awaited_once_with(self.db, self.user, 7) + get_pallet.assert_awaited_once_with(self.db, 'persisted-token') + get_inventory.assert_awaited_once_with(self.db, 2) + locking_read.assert_not_awaited() + add_event.assert_not_awaited() + self.db.commit.assert_not_awaited() + self.assertIsInstance(actual, PalletScanResponseModel) + self.assertEqual(PalletFlowState.UNSTOCKED, actual.flow_state) + self.assertEqual(['inbound'], actual.allowed_actions) + self.assertIsNone(actual.strain_id) + self.assertIsNone(actual.batch_no) + + async def test_scan_rejects_every_ineligible_asset_or_inventory_state(self) -> None: + cases = ( + ('disabled-asset', make_pallet(2, asset_status='1'), make_inventory(2)), + ('wrong-state', make_pallet(2), make_inventory(2, flow_state='EMPTY_IN_WAREHOUSE')), + ('warehouse-set', make_pallet(2), make_inventory(2, warehouse_id=7)), + ('site-set', make_pallet(2), make_inventory(2, site_id=8)), + ('line-set', make_pallet(2), make_inventory(2, current_line_id=9)), + ('filling-set', make_pallet(2), make_inventory(2, active_filling_id=10)), + ('remaining', make_pallet(2), make_inventory(2, remaining_packages=1)), + ) + for label, pallet, inventory in cases: + with ( + self.subTest(label=label), + patch.object(AccessService, 'assert_warehouse_access', new=AsyncMock()), + patch.object( + MasterDao, + 'get_warehouse', + new=AsyncMock(return_value=SimpleNamespace(warehouse_id=7, 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)), + self.assertRaises(ServiceException), + ): + await InboundService.scan_inbound_pallet_services( + self.db, + 'persisted-token', + self.query, + self.user, + ) + + async def test_scan_rejects_unauthorized_disabled_or_missing_target_warehouse_first(self) -> None: + get_pallet = AsyncMock() + with ( + patch.object( + AccessService, + 'assert_warehouse_access', + new=AsyncMock(side_effect=PermissionException(data='', message='无权访问该仓库')), + ), + patch.object(MasterDao, 'get_warehouse', new=AsyncMock()), + patch.object(InventoryDao, 'get_pallet_by_qr_token', new=get_pallet), + self.assertRaises(PermissionException), + ): + await InboundService.scan_inbound_pallet_services(self.db, 'token', self.query, self.user) + get_pallet.assert_not_awaited() + + for warehouse in (None, SimpleNamespace(warehouse_id=7, status='1')): + with self.subTest(warehouse=warehouse): + get_pallet.reset_mock() + with ( + patch.object(AccessService, 'assert_warehouse_access', new=AsyncMock()), + patch.object(MasterDao, 'get_warehouse', new=AsyncMock(return_value=warehouse)), + patch.object(InventoryDao, 'get_pallet_by_qr_token', new=get_pallet), + self.assertRaises(ServiceException), + ): + await InboundService.scan_inbound_pallet_services(self.db, 'token', self.query, self.user) + get_pallet.assert_not_awaited() + + +class TestInboundConfirmService(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, 14, 20) + self.request_hash = InboundService._build_confirm_request_hash(11) + + async def _run_successful_confirm(self, *, commit_side_effect: Exception | None = None) -> dict[str, Any]: + inbound = make_inbound() + inventories = [make_inventory(2, version=4), make_inventory(5, version=8)] + pallets = [make_pallet(2), make_pallet(5)] + actions: list[str] = [] + events: list[BizPalletEvent] = [] + + async def detail(*_args: Any) -> BizInboundOrder: + actions.append('detail') + return inbound + + async def access(*_args: Any) -> None: + actions.append('access') + + async def by_key(*_args: Any) -> None: + actions.append('idempotency-read') + + async def lock_order(*_args: Any) -> BizInboundOrder: + actions.append('order-lock') + return inbound + + 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 + + def transition(current: PalletFlowState, target: PalletFlowState) -> None: + actions.append(f'transition-{current.value}-{target.value}') + + async def update(*_args: Any) -> BizInboundOrder: + actions.append('update-order') + return inbound + + async def add_event( + _db: AsyncSession, + inventory: BizPalletInventory, + event: BizPalletEvent, + ) -> BizPalletEvent: + actions.append(f'event-{inventory.pallet_id}') + events.append(event) + event.event_id = 100 + inventory.pallet_id + inventory.last_event_id = event.event_id + return event + + async def commit() -> None: + actions.append('commit') + if commit_side_effect is not None: + raise commit_side_effect + + self.db.commit.side_effect = commit + patches = ( + patch.object(OperationDao, 'get_inbound_by_id', new=AsyncMock(side_effect=detail)), + patch.object(AccessService, 'assert_warehouse_access', new=AsyncMock(side_effect=access)), + patch.object( + OperationDao, + 'get_inbound_by_confirm_idempotency_key', + new=AsyncMock(side_effect=by_key), + ), + patch.object(OperationDao, 'get_inbound_for_update', new=AsyncMock(side_effect=lock_order)), + patch.object( + InventoryDao, + 'get_inventories_for_update', + new=AsyncMock(side_effect=lock_inventories), + ), + patch.object(InventoryDao, 'get_pallets_by_ids', new=AsyncMock(side_effect=lock_pallets)), + patch.object(PalletStateMachine, 'ensure_transition', side_effect=transition), + patch.object(OperationDao, 'update_inbound', new=AsyncMock(side_effect=update)), + patch.object(InventoryDao, 'add_event', new=AsyncMock(side_effect=add_event)), + patch.object(inbound_service_module, 'datetime'), + ) + with ( + patches[0], + patches[1], + patches[2], + patches[3], + patches[4] as inventory_lock, + patches[5] as pallet_lock, + patches[6] as ensure_transition, + patches[7], + patches[8], + patches[9] as service_datetime, + ): + service_datetime.now.return_value = self.fixed_time + result = await InboundService.confirm_inbound_services(self.db, 11, self.command, self.user) + + return { + 'result': result, + 'inbound': inbound, + 'inventories': inventories, + 'events': events, + 'actions': actions, + 'inventory_lock': inventory_lock, + 'pallet_lock': pallet_lock, + 'ensure_transition': ensure_transition, + } + + async def test_confirm_uses_complete_lock_order_state_machine_version_and_events(self) -> None: + captured = await self._run_successful_confirm() + + self.assertEqual( + [ + 'detail', + 'access', + 'idempotency-read', + 'order-lock', + 'access', + 'inventory-lock', + 'pallet-lock', + 'transition-UNSTOCKED-EMPTY_IN_WAREHOUSE', + 'transition-UNSTOCKED-EMPTY_IN_WAREHOUSE', + 'update-order', + 'event-2', + 'event-5', + 'commit', + ], + captured['actions'], + ) + captured['inventory_lock'].assert_awaited_once_with(self.db, [2, 5]) + captured['pallet_lock'].assert_awaited_once_with(self.db, [2, 5]) + captured['ensure_transition'].assert_has_calls( + [ + unittest.mock.call(PalletFlowState.UNSTOCKED, PalletFlowState.EMPTY_IN_WAREHOUSE), + unittest.mock.call(PalletFlowState.UNSTOCKED, PalletFlowState.EMPTY_IN_WAREHOUSE), + ] + ) + inbound = captured['inbound'] + self.assertEqual('CONFIRMED', inbound.status) + self.assertEqual('confirm-key', inbound.confirm_idempotency_key) + self.assertEqual(self.request_hash, inbound.confirm_request_hash) + self.assertEqual(7, inbound.confirmed_by) + self.assertEqual(self.fixed_time, inbound.confirmed_at) + self.assertEqual('user-7', inbound.update_by) + self.assertEqual([5, 9], [inventory.version for inventory in captured['inventories']]) + for inventory in captured['inventories']: + self.assertEqual('EMPTY_IN_WAREHOUSE', inventory.flow_state) + self.assertEqual(7, inventory.warehouse_id) + self.assertIsNone(inventory.site_id) + self.assertIsNone(inventory.current_line_id) + self.assertIsNone(inventory.active_filling_id) + self.assertEqual(0, inventory.remaining_packages) + self.assertEqual(self.fixed_time, inventory.updated_at) + self.assertEqual([2, 5], [event.pallet_id for event in captured['events']]) + for event in captured['events']: + self.assertEqual(PalletEventType.INBOUND_CONFIRMED.value, event.event_type) + self.assertEqual('INBOUND', event.source_type) + self.assertEqual(11, event.source_id) + self.assertEqual('UNSTOCKED', event.before_state) + self.assertEqual('EMPTY_IN_WAREHOUSE', event.after_state) + self.assertIsNone(event.before_warehouse_id) + self.assertIsNone(event.before_site_id) + self.assertIsNone(event.before_line_id) + self.assertEqual(7, event.after_warehouse_id) + self.assertIsNone(event.after_site_id) + self.assertIsNone(event.after_line_id) + self.assertEqual(0, event.before_remaining_packages) + self.assertEqual(0, 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() + self.assertTrue(captured['result'].is_success) + + async def test_confirm_rejects_incomplete_locked_sets_before_any_mutation(self) -> None: + inbound = make_inbound() + inventories = [make_inventory(2)] + pallets = [make_pallet(2), make_pallet(5)] + update = AsyncMock() + add_event = AsyncMock() + with ( + patch.object(OperationDao, 'get_inbound_by_id', new=AsyncMock(return_value=inbound)), + patch.object(AccessService, 'assert_warehouse_access', new=AsyncMock()), + patch.object(OperationDao, 'get_inbound_by_confirm_idempotency_key', new=AsyncMock(return_value=None)), + patch.object(OperationDao, 'get_inbound_for_update', new=AsyncMock(return_value=inbound)), + patch.object(InventoryDao, 'get_inventories_for_update', new=AsyncMock(return_value=inventories)), + patch.object(InventoryDao, 'get_pallets_by_ids', new=AsyncMock(return_value=pallets)) as pallet_lock, + patch.object(OperationDao, 'update_inbound', new=update), + patch.object(InventoryDao, 'add_event', new=add_event), + self.assertRaises(ServiceException), + ): + await InboundService.confirm_inbound_services(self.db, 11, self.command, self.user) + + pallet_lock.assert_awaited_once_with(self.db, [2, 5]) + self.assertEqual('DRAFT', inbound.status) + self.assertEqual('UNSTOCKED', inventories[0].flow_state) + 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_state_machine_failure_rolls_back_before_writes(self) -> None: + inbound = make_inbound(pallet_ids=(2,)) + inventory = make_inventory(2) + update = AsyncMock() + add_event = AsyncMock() + with ( + patch.object(OperationDao, 'get_inbound_by_id', new=AsyncMock(return_value=inbound)), + patch.object(AccessService, 'assert_warehouse_access', new=AsyncMock()), + patch.object(OperationDao, 'get_inbound_by_confirm_idempotency_key', new=AsyncMock(return_value=None)), + patch.object(OperationDao, 'get_inbound_for_update', new=AsyncMock(return_value=inbound)), + 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(PalletStateMachine, 'ensure_transition', side_effect=ValueError('invalid')), + patch.object(OperationDao, 'update_inbound', new=update), + patch.object(InventoryDao, 'add_event', new=add_event), + self.assertRaisesRegex(ValueError, 'invalid'), + ): + await InboundService.confirm_inbound_services(self.db, 11, self.command, self.user) + + self.assertEqual('DRAFT', inbound.status) + self.assertEqual('UNSTOCKED', inventory.flow_state) + update.assert_not_awaited() + add_event.assert_not_awaited() + self.db.rollback.assert_awaited_once() + + async def test_same_key_retry_rechecks_locked_document_and_ends_read_transaction(self) -> None: + confirmed = make_inbound( + status='CONFIRMED', + idempotency_key='confirm-key', + request_hash=self.request_hash, + ) + inventory_lock = AsyncMock() + with ( + patch.object(OperationDao, 'get_inbound_by_id', new=AsyncMock(return_value=confirmed)), + patch.object(AccessService, 'assert_warehouse_access', new=AsyncMock()) as access, + patch.object( + OperationDao, + 'get_inbound_by_confirm_idempotency_key', + new=AsyncMock(return_value=confirmed), + ), + patch.object(OperationDao, 'get_inbound_for_update', new=AsyncMock(return_value=confirmed)) as lock, + patch.object(InventoryDao, 'get_inventories_for_update', new=inventory_lock), + ): + result = await InboundService.confirm_inbound_services(self.db, 11, self.command, self.user) + + lock.assert_awaited_once_with(self.db, 11) + self.assertEqual(3, access.await_count) + inventory_lock.assert_not_awaited() + self.db.commit.assert_not_awaited() + self.db.rollback.assert_awaited_once() + self.assertTrue(result.is_success) + + async def test_same_key_for_different_inbound_is_rejected_without_lock_or_write(self) -> None: + target = make_inbound() + existing = make_inbound( + inbound_id=12, + warehouse_id=8, + status='CONFIRMED', + idempotency_key='confirm-key', + request_hash=InboundService._build_confirm_request_hash(12), + ) + lock = AsyncMock() + with ( + patch.object(OperationDao, 'get_inbound_by_id', new=AsyncMock(return_value=target)), + patch.object(AccessService, 'assert_warehouse_access', new=AsyncMock()) as access, + patch.object( + OperationDao, + 'get_inbound_by_confirm_idempotency_key', + new=AsyncMock(return_value=existing), + ), + patch.object(OperationDao, 'get_inbound_for_update', new=lock), + self.assertRaises(ServiceException), + ): + await InboundService.confirm_inbound_services(self.db, 11, self.command, self.user) + + self.assertEqual(2, access.await_count) + lock.assert_not_awaited() + self.db.rollback.assert_awaited_once() + + async def test_concurrent_status_change_is_rechecked_after_parent_lock(self) -> None: + initial = make_inbound() + locked = make_inbound( + status='CONFIRMED', + idempotency_key='other-key', + request_hash=self.request_hash, + ) + inventory_lock = AsyncMock() + with ( + patch.object(OperationDao, 'get_inbound_by_id', new=AsyncMock(return_value=initial)), + patch.object(AccessService, 'assert_warehouse_access', new=AsyncMock()), + patch.object(OperationDao, 'get_inbound_by_confirm_idempotency_key', new=AsyncMock(return_value=None)), + patch.object(OperationDao, 'get_inbound_for_update', new=AsyncMock(return_value=locked)), + patch.object(InventoryDao, 'get_inventories_for_update', new=inventory_lock), + self.assertRaises(ServiceException), + ): + await InboundService.confirm_inbound_services(self.db, 11, self.command, self.user) + + inventory_lock.assert_not_awaited() + self.db.rollback.assert_awaited_once() + + async def test_confirm_uses_locked_current_items_instead_of_initial_snapshot(self) -> None: + initial = make_inbound(pallet_ids=(2,)) + locked = make_inbound(pallet_ids=(5,)) + inventory = make_inventory(5) + get_detail = AsyncMock(return_value=initial) + inventory_lock = AsyncMock(return_value=[inventory]) + with ( + patch.object(OperationDao, 'get_inbound_by_id', new=get_detail), + patch.object(AccessService, 'assert_warehouse_access', new=AsyncMock()), + patch.object(OperationDao, 'get_inbound_by_confirm_idempotency_key', new=AsyncMock(return_value=None)), + patch.object(OperationDao, 'get_inbound_for_update', new=AsyncMock(return_value=locked)), + patch.object(InventoryDao, 'get_inventories_for_update', new=inventory_lock), + patch.object(InventoryDao, 'get_pallets_by_ids', new=AsyncMock(return_value=[make_pallet(5)])), + patch.object(OperationDao, 'update_inbound', new=AsyncMock()), + patch.object(InventoryDao, 'add_event', new=AsyncMock()), + ): + result = await InboundService.confirm_inbound_services(self.db, 11, self.command, self.user) + + get_detail.assert_awaited_once_with(self.db, 11) + inventory_lock.assert_awaited_once_with(self.db, [5]) + self.assertEqual('EMPTY_IN_WAREHOUSE', inventory.flow_state) + self.assertTrue(result.is_success) + + async def test_integrity_conflict_recovers_only_exact_confirmed_result_and_ends_read_transaction(self) -> None: + inbound = make_inbound(pallet_ids=(2,)) + inventory = make_inventory(2) + recovered = make_inbound( + pallet_ids=(2,), + status='CONFIRMED', + idempotency_key='confirm-key', + request_hash=self.request_hash, + ) + key_reads = AsyncMock(side_effect=[None, recovered]) + self.db.commit.side_effect = integrity_error() + with ( + patch.object(OperationDao, 'get_inbound_by_id', new=AsyncMock(return_value=inbound)), + patch.object(AccessService, 'assert_warehouse_access', new=AsyncMock()) as access, + patch.object(OperationDao, 'get_inbound_by_confirm_idempotency_key', new=key_reads), + patch.object(OperationDao, 'get_inbound_for_update', new=AsyncMock(return_value=inbound)), + 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_inbound', new=AsyncMock()), + patch.object(InventoryDao, 'add_event', new=AsyncMock()), + ): + result = await InboundService.confirm_inbound_services(self.db, 11, self.command, self.user) + + self.assertEqual(3, access.await_count) + self.assertEqual(2, key_reads.await_count) + self.assertEqual(2, self.db.rollback.await_count) + self.assertTrue(result.is_success) + + async def test_integrity_conflict_with_mismatched_recovery_is_stable_failure(self) -> None: + inbound = make_inbound(pallet_ids=(2,)) + inventory = make_inventory(2) + recovered = make_inbound( + pallet_ids=(2,), + status='CONFIRMED', + idempotency_key='confirm-key', + request_hash='different-hash', + ) + self.db.commit.side_effect = integrity_error() + with ( + patch.object(OperationDao, 'get_inbound_by_id', new=AsyncMock(return_value=inbound)), + patch.object(AccessService, 'assert_warehouse_access', new=AsyncMock()), + patch.object( + OperationDao, + 'get_inbound_by_confirm_idempotency_key', + new=AsyncMock(side_effect=[None, recovered]), + ), + patch.object(OperationDao, 'get_inbound_for_update', new=AsyncMock(return_value=inbound)), + 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_inbound', new=AsyncMock()), + patch.object(InventoryDao, 'add_event', new=AsyncMock()), + self.assertRaises(ServiceException), + ): + await InboundService.confirm_inbound_services(self.db, 11, self.command, self.user) + + self.assertEqual(2, self.db.rollback.await_count) + + +class TestInboundController(unittest.TestCase): + @staticmethod + def _routes() -> dict[tuple[str, str], APIRoute]: + return { + (route.path, next(iter(route.methods))): route + for route in inbound_controller_module.inbound_controller.routes + if isinstance(route, APIRoute) + } + + def test_openapi_has_exact_methods_query_aliases_and_response_models(self) -> None: + app = FastAPI() + app.include_router(inbound_controller_module.inbound_controller) + schema = app.openapi() + expected = { + ('/business/inbound/list', 'get'), + ('/business/inbound/warehouse-options', 'get'), + ('/business/inbound/scan/{qr_token}', 'get'), + ('/business/inbound/{inbound_id}', 'get'), + ('/business/inbound', 'post'), + ('/business/inbound', 'put'), + ('/business/inbound/{inbound_ids}', 'delete'), + ('/business/inbound/{inbound_id}/confirm', 'put'), + } + self.assertEqual(expected, {(path, method) for path, methods in schema['paths'].items() for method in methods}) + scan_parameters = schema['paths']['/business/inbound/scan/{qr_token}']['get']['parameters'] + self.assertIn('warehouseId', {parameter['name'] for parameter in scan_parameters}) + + routes = self._routes() + self.assertEqual(PageResponseModel[InboundModel], routes[('/business/inbound/list', 'GET')].response_model) + self.assertEqual( + DataResponseModel[list[BusinessOptionModel]], + routes[('/business/inbound/warehouse-options', 'GET')].response_model, + ) + self.assertEqual( + DataResponseModel[PalletScanResponseModel], + routes[('/business/inbound/scan/{qr_token}', 'GET')].response_model, + ) + self.assertEqual( + DataResponseModel[InboundModel], + routes[('/business/inbound/{inbound_id}', 'GET')].response_model, + ) + for key in ( + ('/business/inbound', 'POST'), + ('/business/inbound', 'PUT'), + ('/business/inbound/{inbound_ids}', 'DELETE'), + ('/business/inbound/{inbound_id}/confirm', 'PUT'), + ): + self.assertEqual(ResponseBaseModel, routes[key].response_model) + + def test_static_routes_are_declared_before_dynamic_detail(self) -> None: + paths = [route.path for route in inbound_controller_module.inbound_controller.routes] + detail_index = paths.index('/business/inbound/{inbound_id}') + self.assertLess(paths.index('/business/inbound/warehouse-options'), detail_index) + self.assertLess(paths.index('/business/inbound/scan/{qr_token}'), detail_index) + + def test_each_route_has_exact_permission_and_six_unique_permission_values(self) -> None: + expected = { + ('/business/inbound/list', 'GET'): 'business:inbound:list', + ('/business/inbound/warehouse-options', 'GET'): 'business:inbound:query', + ('/business/inbound/scan/{qr_token}', 'GET'): 'business:inbound:query', + ('/business/inbound/{inbound_id}', 'GET'): 'business:inbound:query', + ('/business/inbound', 'POST'): 'business:inbound:add', + ('/business/inbound', 'PUT'): 'business:inbound:edit', + ('/business/inbound/{inbound_ids}', 'DELETE'): 'business:inbound:remove', + ('/business/inbound/{inbound_id}/confirm', 'PUT'): 'business:inbound: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)) + self.assertTrue(any(isinstance(dependency.dependency, PreAuth) for dependency in route.dependencies)) + actual[key] = permissions[0].perm + self.assertEqual(expected, actual) + self.assertEqual( + { + 'business:inbound:list', + 'business:inbound:query', + 'business:inbound:add', + 'business:inbound:edit', + 'business:inbound:remove', + 'business:inbound:confirm', + }, + set(actual.values()), + ) + + def test_all_data_scoped_reads_and_writes_inject_current_user_and_writes_have_log(self) -> None: + routes = self._routes() + for key, route in routes.items(): + 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) + + def test_controller_does_not_add_scan_or_export_permission(self) -> None: + source = inspect.getsource(inbound_controller_module) + self.assertNotIn('business:inbound:scan', source) + self.assertNotIn('business:inbound:export', source) + + +if __name__ == '__main__': + unittest.main() 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 e4f78c6..99e6099 100644 --- a/shierfeng-fastapi-backend/tests/module_business/test_inventory_dao.py +++ b/shierfeng-fastapi-backend/tests/module_business/test_inventory_dao.py @@ -156,8 +156,15 @@ class TestDaoPublicContract(unittest.TestCase): signature = inspect.signature(getattr(OperationDao, method_name)) self.assertEqual(parameter_names, tuple(signature.parameters)) + inbound_list_signature = inspect.signature(OperationDao.get_inbound_list) + self.assertEqual( + ('db', 'query_object', 'is_page', 'warehouse_ids'), + tuple(inbound_list_signature.parameters), + ) + self.assertIs(True, inbound_list_signature.parameters['is_page'].default) + self.assertIsNone(inbound_list_signature.parameters['warehouse_ids'].default) + for method_name in ( - 'get_inbound_list', 'get_filling_list', 'get_shipment_list', 'get_usage_list', @@ -360,13 +367,6 @@ class TestOperationDaoReads(unittest.IsolatedAsyncioTestCase): async def test_for_update_queries_target_primary_keys_and_lock_rows(self) -> None: cases = ( - ( - OperationDao.get_inbound_for_update, - BizInboundOrder, - 'inbound_id', - 11, - (), - ), ( OperationDao.get_shipment_for_update, BizShipmentOrder, @@ -410,6 +410,42 @@ class TestOperationDaoReads(unittest.IsolatedAsyncioTestCase): for fragment in extra_fragments: self.assertIn(fragment, sql) + async def test_inbound_for_update_locks_parent_then_current_items(self) -> None: + inbound = SimpleNamespace(inbound_id=11) + items = [ + SimpleNamespace(inbound_item_id=2, inbound_id=11, pallet_id=2), + SimpleNamespace(inbound_item_id=5, inbound_id=11, pallet_id=5), + ] + inbound_result = MagicMock() + inbound_result.scalar_one_or_none.return_value = inbound + item_result = MagicMock() + item_result.scalars.return_value.all.return_value = items + self.db.execute.side_effect = [inbound_result, item_result] + + actual = await OperationDao.get_inbound_for_update(self.db, inbound_id=11) + + self.assertIs(inbound, actual) + self.assertEqual(items, 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(BizInboundOrder.__table__)) + self.assertTrue(item_statement.get_final_froms()[0].compare(BizInboundItem.__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_inbound_order.inbound_id = 11', parent_sql) + self.assertIn('FOR UPDATE', parent_sql) + self.assertIn('biz_inbound_item.inbound_id = 11', item_sql) + self.assertIn( + 'ORDER BY biz_inbound_item.pallet_id, biz_inbound_item.inbound_item_id', + item_sql, + ) + self.assertIn('FOR UPDATE', item_sql) + async def test_idempotency_queries_use_the_exact_declared_column(self) -> None: cases = ( ( @@ -588,6 +624,58 @@ class TestOperationDaoReads(unittest.IsolatedAsyncioTestCase): for fragment in fragments: self.assertIn(fragment, sql) + async def test_inbound_list_applies_warehouse_scope_inside_paginated_statement(self) -> None: + query = InboundPageQueryModel(page_num=2, page_size=7) + marker = object() + with patch.object(PageUtil, 'paginate', new=AsyncMock(return_value=marker)) as paginate: + actual = await OperationDao.get_inbound_list( + self.db, + query, + is_page=True, + warehouse_ids=[5, 2, 5], + ) + + 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_inbound_order.warehouse_id IN (2, 5)', sql) + self.assertIn('ORDER BY biz_inbound_order.inbound_id', sql) + + async def test_inbound_list_empty_scope_compiles_to_false_for_both_dialects(self) -> None: + query = InboundPageQueryModel(page_num=4, page_size=6) + marker = object() + with patch.object(PageUtil, 'paginate', new=AsyncMock(return_value=marker)) as paginate: + actual = await OperationDao.get_inbound_list( + self.db, + query, + is_page=True, + warehouse_ids=[], + ) + + self.assertIs(marker, actual) + statement = paginate.await_args.args[1] + mysql_sql = compile_sql(statement, mysql.dialect()) + postgresql_sql = compile_sql(statement, postgresql.dialect()) + self.assertIn('WHERE false = 1', mysql_sql) + self.assertIn('WHERE false', postgresql_sql) + self.assertEqual((query.page_num, query.page_size, True), paginate.await_args.args[2:]) + + async def test_inbound_list_none_scope_has_no_authorization_predicate(self) -> None: + query = InboundPageQueryModel() + with patch.object(PageUtil, 'paginate', new=AsyncMock(return_value=object())) as paginate: + await OperationDao.get_inbound_list( + self.db, + query, + is_page=True, + warehouse_ids=None, + ) + + sql = compile_sql(paginate.await_args.args[1]) + self.assertNotIn('warehouse_id IN', sql) + self.assertNotIn('WHERE false', sql) + def test_time_conditions_include_fractional_last_second_with_exclusive_next_day(self) -> None: lower, upper = OperationDao._time_conditions( BizUsageRecord.occurred_at,