From 6851499be2f28a08e0d1ec2e84005a799ff37a4a 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 11:48:22 +0800 Subject: [PATCH] =?UTF-8?q?=E5=8A=9F=E8=83=BD=EF=BC=9A=E6=96=B0=E5=A2=9E?= =?UTF-8?q?=E6=89=98=E7=9B=98=E4=BA=8C=E7=BB=B4=E7=A0=81=E4=B8=8E=E7=94=A8?= =?UTF-8?q?=E6=88=B7=E5=88=86=E9=85=8D?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../controller/pallet_controller.py | 133 +++ .../controller/site_controller.py | 50 +- .../controller/warehouse_controller.py | 50 +- .../module_business/dao/master_dao.py | 127 +++ .../module_business/entity/vo/master_vo.py | 16 + .../module_business/service/pallet_service.py | 282 +++++ .../module_business/test_pallet_service.py | 979 ++++++++++++++++++ 7 files changed, 1635 insertions(+), 2 deletions(-) create mode 100644 shierfeng-fastapi-backend/module_business/controller/pallet_controller.py create mode 100644 shierfeng-fastapi-backend/module_business/service/pallet_service.py create mode 100644 shierfeng-fastapi-backend/tests/module_business/test_pallet_service.py diff --git a/shierfeng-fastapi-backend/module_business/controller/pallet_controller.py b/shierfeng-fastapi-backend/module_business/controller/pallet_controller.py new file mode 100644 index 0000000..6418d54 --- /dev/null +++ b/shierfeng-fastapi-backend/module_business/controller/pallet_controller.py @@ -0,0 +1,133 @@ +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.master_vo import ( + DeletePalletModel, + PalletCreateModel, + PalletEditModel, + PalletModel, + PalletPageQueryModel, +) +from module_business.service.pallet_service import PalletService +from utils.response_util import ResponseUtil + +pallet_controller = APIRouterPro( + prefix='/master/pallet', tags=['基础资料-托盘管理'], dependencies=[PreAuthDependency()] +) + + +@pallet_controller.get( + '/list', + summary='获取托盘分页列表接口', + description='用于获取托盘分页列表', + response_model=PageResponseModel[PalletModel], + dependencies=[UserInterfaceAuthDependency('master:pallet:list')], +) +async def get_master_pallet_list( + request: Request, + pallet_page_query: Annotated[PalletPageQueryModel, Query()], + query_db: Annotated[AsyncSession, DBSessionDependency()], +) -> Response: + result = await PalletService.list_pallet(query_db, pallet_page_query, is_page=True) + return ResponseUtil.success(model_content=result) + + +@pallet_controller.get( + '/{pallet_id}/qrcode', + summary='补打托盘二维码接口', + description='使用托盘原有固定标识重新生成二维码', + response_class=Response, + responses={200: {'content': {'image/png': {}}}}, + dependencies=[UserInterfaceAuthDependency('master:pallet:print')], +) +async def query_master_pallet_qrcode( + request: Request, + pallet_id: Annotated[int, Path(gt=0, description='托盘ID')], + query_db: Annotated[AsyncSession, DBSessionDependency()], +) -> Response: + png = await PalletService.get_pallet_qrcode(query_db, pallet_id) + return Response(content=png, media_type='image/png') + + +@pallet_controller.get( + '/{pallet_id}', + summary='获取托盘详情接口', + description='用于获取指定托盘的详细信息', + response_model=DataResponseModel[PalletModel], + dependencies=[UserInterfaceAuthDependency('master:pallet:query')], +) +async def query_detail_master_pallet( + request: Request, + pallet_id: Annotated[int, Path(gt=0, description='托盘ID')], + query_db: Annotated[AsyncSession, DBSessionDependency()], +) -> Response: + result = await PalletService.detail_pallet(query_db, pallet_id) + return ResponseUtil.success(data=result) + + +@pallet_controller.post( + '', + summary='新增托盘接口', + description='用于新增托盘及初始库存', + response_model=ResponseBaseModel, + dependencies=[UserInterfaceAuthDependency('master:pallet:add')], +) +@Log(title='托盘管理', business_type=BusinessType.INSERT) +async def add_master_pallet( + request: Request, + add_pallet: PalletCreateModel, + query_db: Annotated[AsyncSession, DBSessionDependency()], + current_user: Annotated[CurrentUserModel, CurrentUserDependency()], +) -> Response: + result = await PalletService.add_pallet(query_db, add_pallet, current_user.user.user_name) + return ResponseUtil.success(msg=result.message) + + +@pallet_controller.put( + '', + summary='编辑托盘接口', + description='仅编辑托盘容量、资产状态和备注', + response_model=ResponseBaseModel, + dependencies=[UserInterfaceAuthDependency('master:pallet:edit')], +) +@Log(title='托盘管理', business_type=BusinessType.UPDATE) +async def edit_master_pallet( + request: Request, + edit_pallet: PalletEditModel, + query_db: Annotated[AsyncSession, DBSessionDependency()], + current_user: Annotated[CurrentUserModel, CurrentUserDependency()], +) -> Response: + result = await PalletService.edit_pallet(query_db, edit_pallet, current_user.user.user_name) + return ResponseUtil.success(msg=result.message) + + +@pallet_controller.delete( + '/{pallet_ids}', + summary='删除托盘接口', + description='用于删除尚未入库且没有流转事件的托盘', + response_model=ResponseBaseModel, + dependencies=[UserInterfaceAuthDependency('master:pallet:remove')], +) +@Log(title='托盘管理', business_type=BusinessType.DELETE) +async def delete_master_pallet( + request: Request, + pallet_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: + delete_pallet = DeletePalletModel(palletIds=[int(pallet_id) for pallet_id in pallet_ids.split(',')]) + result = await PalletService.remove_pallet(query_db, delete_pallet) + return ResponseUtil.success(msg=result.message) diff --git a/shierfeng-fastapi-backend/module_business/controller/site_controller.py b/shierfeng-fastapi-backend/module_business/controller/site_controller.py index c27d975..8bee104 100644 --- a/shierfeng-fastapi-backend/module_business/controller/site_controller.py +++ b/shierfeng-fastapi-backend/module_business/controller/site_controller.py @@ -12,13 +12,61 @@ 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.master_vo import DeleteSiteModel, SiteModel, SitePageQueryModel +from module_business.entity.vo.master_vo import ( + BusinessUserOptionModel, + DeleteSiteModel, + SiteModel, + SitePageQueryModel, + UserAssignModel, +) from module_business.service.master_service import MasterService +from module_business.service.pallet_service import PalletService from utils.response_util import ResponseUtil +ASSIGN_LOG = Log +ASSIGN_CURRENT_USER = CurrentUserDependency + site_controller = APIRouterPro( prefix='/master/site', tags=['基础资料-客户现场管理'], dependencies=[PreAuthDependency()] ) +site_user_controller = APIRouterPro( + prefix='/master/site', tags=['基础资料-客户现场用户分配'], dependencies=[PreAuthDependency()] +) + + +@site_user_controller.get( + '/{site_id}/users', + summary='获取客户现场可分配用户接口', + description='获取启用且未删除的用户及客户现场分配状态', + response_model=DataResponseModel[list[BusinessUserOptionModel]], + dependencies=[UserInterfaceAuthDependency('master:site:assign')], +) +async def get_master_site_users( + request: Request, + site_id: Annotated[int, Path(gt=0, description='客户现场ID')], + query_db: Annotated[AsyncSession, DBSessionDependency()], +) -> Response: + users = await PalletService.list_site_users(query_db, site_id) + return ResponseUtil.success(data=users) + + +@site_user_controller.put( + '/{site_id}/users', + summary='分配客户现场用户接口', + description='完整替换客户现场的用户关联', + response_model=ResponseBaseModel, + dependencies=[UserInterfaceAuthDependency('master:site:assign')], +) +@ASSIGN_LOG(title='客户现场用户分配', business_type=BusinessType.GRANT) +async def assign_master_site_users( + request: Request, + site_id: Annotated[int, Path(gt=0, description='客户现场ID')], + assignment: UserAssignModel, + query_db: Annotated[AsyncSession, DBSessionDependency()], + current_user: Annotated[CurrentUserModel, ASSIGN_CURRENT_USER()], +) -> Response: + result = await PalletService.assign_site_users(query_db, site_id, assignment) + return ResponseUtil.success(msg=result.message) @site_controller.get( diff --git a/shierfeng-fastapi-backend/module_business/controller/warehouse_controller.py b/shierfeng-fastapi-backend/module_business/controller/warehouse_controller.py index ab12ae5..511d2be 100644 --- a/shierfeng-fastapi-backend/module_business/controller/warehouse_controller.py +++ b/shierfeng-fastapi-backend/module_business/controller/warehouse_controller.py @@ -12,13 +12,61 @@ 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.master_vo import DeleteWarehouseModel, WarehouseModel, WarehousePageQueryModel +from module_business.entity.vo.master_vo import ( + BusinessUserOptionModel, + DeleteWarehouseModel, + UserAssignModel, + WarehouseModel, + WarehousePageQueryModel, +) from module_business.service.master_service import MasterService +from module_business.service.pallet_service import PalletService from utils.response_util import ResponseUtil +ASSIGN_LOG = Log +ASSIGN_CURRENT_USER = CurrentUserDependency + warehouse_controller = APIRouterPro( prefix='/master/warehouse', tags=['基础资料-仓库管理'], dependencies=[PreAuthDependency()] ) +warehouse_user_controller = APIRouterPro( + prefix='/master/warehouse', tags=['基础资料-仓库用户分配'], dependencies=[PreAuthDependency()] +) + + +@warehouse_user_controller.get( + '/{warehouse_id}/users', + summary='获取仓库可分配用户接口', + description='获取启用且未删除的用户及仓库分配状态', + response_model=DataResponseModel[list[BusinessUserOptionModel]], + dependencies=[UserInterfaceAuthDependency('master:warehouse:assign')], +) +async def get_master_warehouse_users( + request: Request, + warehouse_id: Annotated[int, Path(gt=0, description='仓库ID')], + query_db: Annotated[AsyncSession, DBSessionDependency()], +) -> Response: + users = await PalletService.list_warehouse_users(query_db, warehouse_id) + return ResponseUtil.success(data=users) + + +@warehouse_user_controller.put( + '/{warehouse_id}/users', + summary='分配仓库用户接口', + description='完整替换仓库的用户关联', + response_model=ResponseBaseModel, + dependencies=[UserInterfaceAuthDependency('master:warehouse:assign')], +) +@ASSIGN_LOG(title='仓库用户分配', business_type=BusinessType.GRANT) +async def assign_master_warehouse_users( + request: Request, + warehouse_id: Annotated[int, Path(gt=0, description='仓库ID')], + assignment: UserAssignModel, + query_db: Annotated[AsyncSession, DBSessionDependency()], + current_user: Annotated[CurrentUserModel, ASSIGN_CURRENT_USER()], +) -> Response: + result = await PalletService.assign_warehouse_users(query_db, warehouse_id, assignment) + return ResponseUtil.success(msg=result.message) @warehouse_controller.get( diff --git a/shierfeng-fastapi-backend/module_business/dao/master_dao.py b/shierfeng-fastapi-backend/module_business/dao/master_dao.py index b21d110..e267e93 100644 --- a/shierfeng-fastapi-backend/module_business/dao/master_dao.py +++ b/shierfeng-fastapi-backend/module_business/dao/master_dao.py @@ -1,3 +1,4 @@ +from datetime import datetime from typing import Any, Literal from pydantic import BaseModel @@ -6,9 +7,11 @@ from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.orm.attributes import InstrumentedAttribute from common.vo import PageModel +from module_admin.entity.do.user_do import SysUser from module_business.entity.do.master_do import ( BizCustomer, BizCustomerSite, + BizPallet, BizProductionLine, BizStrain, BizUserSite, @@ -29,6 +32,8 @@ from module_business.entity.vo.master_vo import ( CustomerPageQueryModel, LineModel, LinePageQueryModel, + PalletEditModel, + PalletPageQueryModel, SiteModel, SitePageQueryModel, StrainModel, @@ -357,6 +362,128 @@ class MasterDao: async def delete_warehouse(cls, db: AsyncSession, warehouse_id: int) -> None: await cls._delete(db, BizWarehouse, BizWarehouse.warehouse_id, warehouse_id) + @classmethod + async def list_pallet( + cls, + db: AsyncSession, + query_object: PalletPageQueryModel, + is_page: bool = True, + ) -> PageModel | list[dict[str, Any]]: + conditions = [] + if query_object.pallet_code: + conditions.append(BizPallet.pallet_code.like(f'%{query_object.pallet_code}%')) + if query_object.asset_status: + conditions.append(BizPallet.asset_status == query_object.asset_status) + statement = select(BizPallet).where(*conditions).order_by(BizPallet.pallet_id) + return await cls._paginate(db, statement, query_object, is_page) + + @classmethod + async def get_pallet( + cls, + db: AsyncSession, + pallet_id: int, + *, + for_update: bool = False, + ) -> BizPallet | None: + statement = select(BizPallet).where(BizPallet.pallet_id == pallet_id) + if for_update: + statement = statement.with_for_update() + result = await db.execute(statement) + return result.scalar_one_or_none() + + @classmethod + async def find_pallet_by_code(cls, db: AsyncSession, pallet_code: str) -> BizPallet | None: + return await cls._find(db, BizPallet, BizPallet.pallet_code, pallet_code) + + @classmethod + async def add_pallet(cls, db: AsyncSession, pallet: BizPallet) -> BizPallet: + db.add(pallet) + await db.flush() + return pallet + + @classmethod + async def update_pallet( + cls, + db: AsyncSession, + pallet: PalletEditModel, + update_by: str, + update_time: datetime, + ) -> None: + await db.execute( + update(BizPallet) + .where(BizPallet.pallet_id == pallet.pallet_id) + .values( + capacity_packages=pallet.capacity_packages, + asset_status=pallet.asset_status, + remark=pallet.remark, + update_by=update_by, + update_time=update_time, + ) + ) + + @classmethod + async def delete_pallet(cls, db: AsyncSession, pallet_id: int) -> None: + await cls._delete(db, BizPallet, BizPallet.pallet_id, pallet_id) + + @classmethod + async def get_pallet_inventory( + cls, + db: AsyncSession, + pallet_id: int, + *, + for_update: bool = False, + ) -> BizPalletInventory | None: + statement = select(BizPalletInventory).where(BizPalletInventory.pallet_id == pallet_id) + if for_update: + statement = statement.with_for_update() + result = await db.execute(statement) + return result.scalar_one_or_none() + + @classmethod + async def add_pallet_inventory( + cls, + db: AsyncSession, + inventory: BizPalletInventory, + ) -> BizPalletInventory: + db.add(inventory) + await db.flush() + return inventory + + @classmethod + async def delete_pallet_inventory(cls, db: AsyncSession, pallet_id: int) -> None: + await cls._delete(db, BizPalletInventory, BizPalletInventory.pallet_id, pallet_id) + + @classmethod + async def count_pallet_events(cls, db: AsyncSession, pallet_id: int) -> int: + result = await db.execute( + select(func.count('*')).select_from(BizPalletEvent).where(BizPalletEvent.pallet_id == pallet_id) + ) + return result.scalar() or 0 + + @classmethod + async def list_assignable_users(cls, db: AsyncSession) -> list[Any]: + result = await db.execute( + select(SysUser.user_id, SysUser.user_name, SysUser.nick_name) + .where(SysUser.status == '0', SysUser.del_flag == '0') + .order_by(SysUser.user_id) + ) + return list(result.all()) + + @classmethod + async def list_assignable_users_by_ids(cls, db: AsyncSession, user_ids: list[int]) -> list[Any]: + if not user_ids: + return [] + result = await db.execute( + select(SysUser.user_id, SysUser.user_name, SysUser.nick_name) + .where( + SysUser.user_id.in_(user_ids), + SysUser.status == '0', + SysUser.del_flag == '0', + ) + .order_by(SysUser.user_id) + ) + return list(result.all()) + @classmethod async def count_references( cls, diff --git a/shierfeng-fastapi-backend/module_business/entity/vo/master_vo.py b/shierfeng-fastapi-backend/module_business/entity/vo/master_vo.py index 4960b4e..5b94556 100644 --- a/shierfeng-fastapi-backend/module_business/entity/vo/master_vo.py +++ b/shierfeng-fastapi-backend/module_business/entity/vo/master_vo.py @@ -389,3 +389,19 @@ class UserAssignModel(BaseModel): user_ids: list[PositiveId] = Field(min_length=1, description='用户ID列表') _validate_unique_ids = field_validator('user_ids')(_ensure_unique_ids) + + +class BusinessUserOptionModel(BaseModel): + """仓库或客户现场可分配用户的最小只读模型。""" + + model_config = ConfigDict( + alias_generator=to_camel, + from_attributes=True, + populate_by_name=True, + extra='forbid', + ) + + user_id: PositiveId = Field(description='用户ID') + user_name: str = Field(min_length=1, max_length=30, pattern=r'\S', description='用户账号') + nick_name: str | None = Field(default=None, max_length=30, description='用户昵称') + assigned: bool = Field(description='是否已分配') diff --git a/shierfeng-fastapi-backend/module_business/service/pallet_service.py b/shierfeng-fastapi-backend/module_business/service/pallet_service.py new file mode 100644 index 0000000..d4cd515 --- /dev/null +++ b/shierfeng-fastapi-backend/module_business/service/pallet_service.py @@ -0,0 +1,282 @@ +import secrets +from datetime import datetime +from io import BytesIO +from typing import Any, Literal + +import qrcode +from sqlalchemy.exc import IntegrityError +from sqlalchemy.ext.asyncio import AsyncSession + +from common.vo import CrudResponseModel, PageModel +from exceptions.exception import ServiceException +from module_business.dao.access_dao import AccessDao +from module_business.dao.master_dao import MasterDao +from module_business.domain.enums import PalletFlowState +from module_business.entity.do.master_do import BizPallet +from module_business.entity.do.operation_do import BizPalletInventory +from module_business.entity.vo.master_vo import ( + BusinessUserOptionModel, + DeletePalletModel, + PalletCreateModel, + PalletEditModel, + PalletModel, + PalletPageQueryModel, + UserAssignModel, +) + +AssignmentResource = Literal['warehouse', 'site'] + +_ASSIGNMENT_LABELS: dict[AssignmentResource, str] = { + 'warehouse': '仓库', + 'site': '客户现场', +} + + +class PalletService: + """托盘资产及业务用户分配服务层。""" + + @staticmethod + def render_qr_png(qr_token: str) -> bytes: + """将固定二维码标识渲染为 PNG；补打不会生成新标识。""" + buffer = BytesIO() + qrcode.make(qr_token).save(buffer, format='PNG') + return buffer.getvalue() + + @classmethod + async def get_pallet_qrcode(cls, db: AsyncSession, pallet_id: int) -> bytes: + pallet = await MasterDao.get_pallet(db, pallet_id) + if pallet is None: + raise ServiceException(message='托盘不存在') + return cls.render_qr_png(pallet.qr_token) + + @staticmethod + def _extract_pallet_ids(value: int | DeletePalletModel) -> list[int]: + pallet_ids = [value] if isinstance(value, int) else list(value.pallet_ids) + if not pallet_ids or any(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) + + @classmethod + async def _can_remove(cls, db: AsyncSession, pallet_id: int) -> bool: + inventory = await MasterDao.get_pallet_inventory(db, pallet_id) + if inventory is None or inventory.flow_state != PalletFlowState.UNSTOCKED.value: + return False + return await MasterDao.count_pallet_events(db, pallet_id) == 0 + + @classmethod + async def _to_pallet_model(cls, db: AsyncSession, record: Any) -> PalletModel: + model = PalletModel.model_validate(record) + return model.model_copy(update={'can_remove': await cls._can_remove(db, model.pallet_id)}) + + @classmethod + async def _lock_pallet_inventory(cls, db: AsyncSession, pallet_id: int) -> tuple[BizPallet, BizPalletInventory]: + """按库存、托盘顺序加锁，避免与流转事务形成反序死锁。""" + inventory = await MasterDao.get_pallet_inventory(db, pallet_id, for_update=True) + if inventory is None: + if await MasterDao.get_pallet(db, pallet_id) is None: + raise ServiceException(message='托盘不存在') + raise ServiceException(message='托盘库存不存在') + pallet = await MasterDao.get_pallet(db, pallet_id, for_update=True) + if pallet is None: + raise ServiceException(message='托盘不存在') + return pallet, inventory + + @classmethod + async def list_pallet( + cls, + db: AsyncSession, + query_object: PalletPageQueryModel, + is_page: bool = True, + ) -> PageModel | list[PalletModel]: + result = await MasterDao.list_pallet(db, query_object, is_page) + if isinstance(result, PageModel): + rows = [await cls._to_pallet_model(db, row) for row in result.rows] + return result.model_copy(update={'rows': rows}) + return [await cls._to_pallet_model(db, row) for row in result] + + @classmethod + async def detail_pallet(cls, db: AsyncSession, pallet_id: int) -> PalletModel: + pallet = await MasterDao.get_pallet(db, pallet_id) + if pallet is None: + raise ServiceException(message='托盘不存在') + return await cls._to_pallet_model(db, pallet) + + @classmethod + async def add_pallet( + cls, + db: AsyncSession, + pallet: PalletCreateModel, + current_user_name: str, + ) -> CrudResponseModel: + try: + if await MasterDao.find_pallet_by_code(db, pallet.pallet_code) is not None: + raise ServiceException(message='托盘编号已存在') + + current_time = datetime.now() + pallet_record = BizPallet( + pallet_code=pallet.pallet_code, + qr_token=secrets.token_urlsafe(32), + capacity_packages=pallet.capacity_packages, + asset_status='0', + remark=pallet.remark, + create_by=current_user_name, + create_time=current_time, + update_by=current_user_name, + update_time=current_time, + ) + pallet_record = await MasterDao.add_pallet(db, pallet_record) + inventory = BizPalletInventory( + pallet_id=pallet_record.pallet_id, + flow_state='UNSTOCKED', + remaining_packages=0, + version=0, + updated_at=current_time, + ) + await MasterDao.add_pallet_inventory(db, inventory) + await db.commit() + return CrudResponseModel(is_success=True, message='新增成功') + except IntegrityError: + await db.rollback() + raise ServiceException(message='新增托盘失败，托盘编号或二维码标识已存在') from None + except Exception: + await db.rollback() + raise + + @classmethod + async def edit_pallet( + cls, + db: AsyncSession, + pallet: PalletEditModel, + current_user_name: str, + ) -> CrudResponseModel: + try: + _, inventory = await cls._lock_pallet_inventory(db, pallet.pallet_id) + if pallet.capacity_packages < inventory.remaining_packages: + raise ServiceException(message='托盘容量不能小于当前剩余包数') + await MasterDao.update_pallet( + db, + pallet, + update_by=current_user_name, + update_time=datetime.now(), + ) + await db.commit() + return CrudResponseModel(is_success=True, message='更新成功') + except IntegrityError: + await db.rollback() + raise ServiceException(message='修改托盘失败，托盘状态已变化') from None + except Exception: + await db.rollback() + raise + + @classmethod + async def remove_pallet( + cls, + db: AsyncSession, + value: int | DeletePalletModel, + ) -> CrudResponseModel: + try: + pallet_ids = cls._extract_pallet_ids(value) + for pallet_id in pallet_ids: + _, inventory = await cls._lock_pallet_inventory(db, pallet_id) + if inventory.flow_state != PalletFlowState.UNSTOCKED.value: + raise ServiceException(message='托盘未处于未入库状态，不能删除') + if await MasterDao.count_pallet_events(db, pallet_id) > 0: + raise ServiceException(message='托盘已有流转事件，不能删除') + + for pallet_id in pallet_ids: + await MasterDao.delete_pallet_inventory(db, pallet_id) + await MasterDao.delete_pallet(db, pallet_id) + await db.commit() + return CrudResponseModel(is_success=True, message='删除成功') + except IntegrityError: + await db.rollback() + raise ServiceException(message='删除托盘失败，托盘状态或关联数据已变化') from None + except Exception: + await db.rollback() + raise + + @classmethod + async def _list_resource_users( + cls, + db: AsyncSession, + resource: AssignmentResource, + resource_id: int, + ) -> list[BusinessUserOptionModel]: + if await getattr(MasterDao, f'get_{resource}')(db, resource_id) is None: + raise ServiceException(message=f'{_ASSIGNMENT_LABELS[resource]}不存在') + + users = await MasterDao.list_assignable_users(db) + result = [] + association_method = getattr(AccessDao, f'list_user_{resource}_ids') + for user in users: + assigned_ids = await association_method(db, user.user_id) + result.append( + BusinessUserOptionModel( + user_id=user.user_id, + user_name=user.user_name, + nick_name=user.nick_name, + assigned=resource_id in assigned_ids, + ) + ) + return result + + @classmethod + async def list_warehouse_users( + cls, + db: AsyncSession, + warehouse_id: int, + ) -> list[BusinessUserOptionModel]: + return await cls._list_resource_users(db, 'warehouse', warehouse_id) + + @classmethod + async def list_site_users( + cls, + db: AsyncSession, + site_id: int, + ) -> list[BusinessUserOptionModel]: + return await cls._list_resource_users(db, 'site', site_id) + + @classmethod + async def _assign_resource_users( + cls, + db: AsyncSession, + resource: AssignmentResource, + resource_id: int, + assignment: UserAssignModel, + ) -> CrudResponseModel: + label = _ASSIGNMENT_LABELS[resource] + try: + if await getattr(MasterDao, f'get_{resource}')(db, resource_id) is None: + raise ServiceException(message=f'{label}不存在') + users = await MasterDao.list_assignable_users_by_ids(db, assignment.user_ids) + if {user.user_id for user in users} != set(assignment.user_ids): + raise ServiceException(message='存在不存在、停用或已删除的用户') + await getattr(AccessDao, f'replace_{resource}_users')(db, resource_id, assignment.user_ids) + await db.commit() + return CrudResponseModel(is_success=True, message='分配成功') + except IntegrityError: + await db.rollback() + raise ServiceException(message=f'分配{label}用户失败，关联数据无效') from None + except Exception: + await db.rollback() + raise + + @classmethod + async def assign_warehouse_users( + cls, + db: AsyncSession, + warehouse_id: int, + assignment: UserAssignModel, + ) -> CrudResponseModel: + return await cls._assign_resource_users(db, 'warehouse', warehouse_id, assignment) + + @classmethod + async def assign_site_users( + cls, + db: AsyncSession, + site_id: int, + assignment: UserAssignModel, + ) -> CrudResponseModel: + return await cls._assign_resource_users(db, 'site', site_id, assignment) diff --git a/shierfeng-fastapi-backend/tests/module_business/test_pallet_service.py b/shierfeng-fastapi-backend/tests/module_business/test_pallet_service.py new file mode 100644 index 0000000..29a984e --- /dev/null +++ b/shierfeng-fastapi-backend/tests/module_business/test_pallet_service.py @@ -0,0 +1,979 @@ +import inspect +import secrets +import unittest +from datetime import datetime +from types import SimpleNamespace +from typing import Any +from unittest.mock import AsyncMock, MagicMock, patch + +from fastapi import FastAPI +from fastapi.routing import APIRoute +from pydantic import ValidationError +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 ServiceException +from module_admin.entity.do.user_do import SysUser +from module_admin.service.login_service import LoginService +from module_business.controller import pallet_controller as pallet_controller_module +from module_business.controller import site_controller as site_controller_module +from module_business.controller import warehouse_controller as warehouse_controller_module +from module_business.dao.access_dao import AccessDao +from module_business.dao.master_dao import MasterDao +from module_business.entity.do.master_do import BizPallet +from module_business.entity.do.operation_do import BizPalletInventory +from module_business.entity.vo.master_vo import ( + BusinessUserOptionModel, + DeletePalletModel, + PalletCreateModel, + PalletEditModel, + PalletModel, + PalletPageQueryModel, + UserAssignModel, +) +from module_business.service import pallet_service as pallet_service_module +from module_business.service.pallet_service import PalletService +from utils.page_util import PageUtil + + +def compile_sql(statement: Any) -> str: + return ' '.join(str(statement.compile(compile_kwargs={'literal_binds': True})).split()) + + +def integrity_error() -> IntegrityError: + return IntegrityError('statement', {}, RuntimeError('constraint')) + + +def make_pallet_record(pallet_id: int, capacity_packages: int = 20) -> SimpleNamespace: + return SimpleNamespace( + pallet_id=pallet_id, + pallet_code=f'P-{pallet_id:03d}', + qr_token=f'fixed-token-{pallet_id}', + capacity_packages=capacity_packages, + asset_status='0', + remark=None, + ) + + +class TestPalletServiceContract(unittest.TestCase): + def test_service_is_available(self) -> None: + self.assertIsNotNone(PalletService) + + +class TestPalletVoContract(unittest.TestCase): + def test_business_user_option_only_exposes_safe_fields(self) -> None: + source = SimpleNamespace( + user_id=7, + user_name='operator', + nick_name=None, + assigned=True, + password='secret', + email='operator@example.com', + phonenumber='13800000000', + token='token', + ) + + option = BusinessUserOptionModel.model_validate(source) + + self.assertEqual( + {'userId': 7, 'userName': 'operator', 'nickName': None, 'assigned': True}, + option.model_dump(by_alias=True), + ) + self.assertTrue( + {'password', 'email', 'phonenumber', 'phone', 'token'}.isdisjoint(BusinessUserOptionModel.model_fields) + ) + + def test_business_user_option_validates_identity_fields(self) -> None: + invalid_values = ( + {'userId': 0, 'userName': 'operator', 'assigned': False}, + {'userId': 1, 'userName': '', 'assigned': False}, + {'userId': 1, 'userName': ' ', 'assigned': False}, + ) + for value in invalid_values: + with self.subTest(value=value), self.assertRaises(ValidationError): + BusinessUserOptionModel.model_validate(value) + + def test_pallet_edit_rejects_code_and_qr_token(self) -> None: + base = {'palletId': 1, 'capacityPackages': 20, 'assetStatus': '0'} + for identity_field in ({'palletCode': 'P-001'}, {'qrToken': 'fixed-token'}): + with self.subTest(identity_field=identity_field), self.assertRaises(ValidationError): + PalletEditModel.model_validate({**base, **identity_field}) + + +class TestPalletMasterDao(unittest.IsolatedAsyncioTestCase): + def setUp(self) -> None: + self.db = MagicMock(spec=AsyncSession) + + async def test_list_pallet_filters_and_orders_by_pallet_id(self) -> None: + query = PalletPageQueryModel(palletCode='P-00', assetStatus='0') + expected = object() + + with patch.object(PageUtil, 'paginate', new=AsyncMock(return_value=expected)) as paginate: + actual = await MasterDao.list_pallet(self.db, query, is_page=True) + + self.assertIs(expected, actual) + statement = paginate.await_args.args[1] + sql = compile_sql(statement) + self.assertIn('FROM biz_pallet', sql) + self.assertIn("biz_pallet.pallet_code LIKE '%P-00%'", sql) + self.assertIn("biz_pallet.asset_status = '0'", sql) + self.assertIn('ORDER BY biz_pallet.pallet_id', sql) + + async def test_add_pallet_and_inventory_only_flush_current_transaction(self) -> None: + pallet = BizPallet( + pallet_code='P-001', + qr_token='fixed-token', + capacity_packages=20, + asset_status='0', + ) + inventory = BizPalletInventory( + pallet_id=3, + flow_state='UNSTOCKED', + remaining_packages=0, + version=0, + ) + + actual = await MasterDao.add_pallet(self.db, pallet) + await MasterDao.add_pallet_inventory(self.db, inventory) + + self.assertIs(pallet, actual) + self.assertEqual([pallet, inventory], [call.args[0] for call in self.db.add.call_args_list]) + self.assertEqual(2, self.db.flush.await_count) + self.db.commit.assert_not_awaited() + self.db.rollback.assert_not_awaited() + + async def test_update_pallet_sql_never_contains_identity_fields(self) -> None: + edit = PalletEditModel(palletId=3, capacityPackages=18, assetStatus='1', remark='停用') + updated_at = datetime(2026, 7, 26, 10, 30) + + await MasterDao.update_pallet(self.db, edit, update_by='admin', update_time=updated_at) + + statement = self.db.execute.await_args.args[0] + sql = compile_sql(statement) + self.assertIn('UPDATE biz_pallet', sql) + self.assertIn('capacity_packages=18', sql) + self.assertIn("asset_status='1'", sql) + self.assertIn("remark='停用'", sql) + self.assertIn("update_by='admin'", sql) + self.assertIn('WHERE biz_pallet.pallet_id = 3', sql) + self.assertNotIn('pallet_code=', sql) + self.assertNotIn('qr_token=', sql) + self.db.commit.assert_not_awaited() + self.db.rollback.assert_not_awaited() + + async def test_inventory_delete_and_event_count_target_one_pallet(self) -> None: + count_result = MagicMock() + count_result.scalar.return_value = 2 + self.db.execute.side_effect = [MagicMock(), count_result] + + await MasterDao.delete_pallet_inventory(self.db, pallet_id=9) + count = await MasterDao.count_pallet_events(self.db, pallet_id=9) + + delete_statement = self.db.execute.await_args_list[0].args[0] + self.assertTrue(delete_statement.table.compare(BizPalletInventory.__table__)) + self.assertIn('biz_pallet_inventory.pallet_id = 9', compile_sql(delete_statement)) + count_statement = self.db.execute.await_args_list[1].args[0] + self.assertIn('FROM biz_pallet_event', compile_sql(count_statement)) + self.assertIn('biz_pallet_event.pallet_id = 9', compile_sql(count_statement)) + self.assertEqual(2, count) + self.db.commit.assert_not_awaited() + self.db.rollback.assert_not_awaited() + + async def test_inventory_write_read_uses_row_lock(self) -> None: + result = MagicMock() + inventory = SimpleNamespace(pallet_id=9) + result.scalar_one_or_none.return_value = inventory + self.db.execute.return_value = result + + actual = await MasterDao.get_pallet_inventory(self.db, pallet_id=9, for_update=True) + + self.assertIs(inventory, actual) + statement = self.db.execute.await_args.args[0] + self.assertIn('biz_pallet_inventory.pallet_id = 9', compile_sql(statement)) + self.assertIn('FOR UPDATE', compile_sql(statement)) + + async def test_pallet_write_read_uses_row_lock(self) -> None: + result = MagicMock() + pallet = make_pallet_record(9) + result.scalar_one_or_none.return_value = pallet + self.db.execute.return_value = result + + actual = await MasterDao.get_pallet(self.db, pallet_id=9, for_update=True) + + self.assertIs(pallet, actual) + statement = self.db.execute.await_args.args[0] + self.assertIn('biz_pallet.pallet_id = 9', compile_sql(statement)) + self.assertIn('FOR UPDATE', compile_sql(statement)) + + async def test_assignable_user_queries_filter_disabled_and_deleted_rows(self) -> None: + list_result = MagicMock() + list_result.all.return_value = () + selected_result = MagicMock() + selected_result.all.return_value = () + self.db.execute.side_effect = [list_result, selected_result] + + self.assertEqual([], await MasterDao.list_assignable_users(self.db)) + self.assertEqual([], await MasterDao.list_assignable_users_by_ids(self.db, [2, 5])) + + for statement in (call.args[0] for call in self.db.execute.await_args_list): + sql = compile_sql(statement) + self.assertIn('FROM sys_user', sql) + self.assertIn("sys_user.status = '0'", sql) + self.assertIn("sys_user.del_flag = '0'", sql) + self.assertNotIn('password', sql) + self.assertEqual( + [SysUser.user_id, SysUser.user_name, SysUser.nick_name], + list(statement.selected_columns), + ) + self.assertIn('ORDER BY sys_user.user_id', compile_sql(self.db.execute.await_args_list[0].args[0])) + selected_sql = compile_sql(self.db.execute.await_args_list[1].args[0]) + self.assertIn('sys_user.user_id IN (2, 5)', selected_sql) + + +class TestPalletCreationService(unittest.IsolatedAsyncioTestCase): + def setUp(self) -> None: + self.db = MagicMock(spec=AsyncSession) + self.create = PalletCreateModel(palletCode='P-001', capacityPackages=20, remark='测试托盘') + self.fixed_time = datetime(2026, 7, 26, 11, 30) + + async def test_add_generates_server_token_and_initial_inventory_in_one_transaction(self) -> None: + actions: list[str] = [] + captured: dict[str, Any] = {} + + async def add_pallet(_db: AsyncSession, pallet: BizPallet) -> BizPallet: + actions.append('pallet') + captured['pallet'] = pallet + pallet.pallet_id = 17 + return pallet + + async def add_inventory(_db: AsyncSession, inventory: BizPalletInventory) -> BizPalletInventory: + actions.append('inventory') + captured['inventory'] = inventory + return inventory + + async def commit() -> None: + actions.append('commit') + + self.db.commit.side_effect = commit + with ( + patch.object(MasterDao, 'find_pallet_by_code', new=AsyncMock(return_value=None)), + patch.object(MasterDao, 'add_pallet', new=AsyncMock(side_effect=add_pallet)), + patch.object(MasterDao, 'add_pallet_inventory', new=AsyncMock(side_effect=add_inventory)), + patch.object(secrets, 'token_urlsafe', return_value='server-fixed-token') as token_urlsafe, + patch.object(pallet_service_module, 'datetime', create=True) as service_datetime, + ): + service_datetime.now.return_value = self.fixed_time + result = await PalletService.add_pallet(self.db, self.create, current_user_name='admin') + + token_urlsafe.assert_called_once_with(32) + pallet = captured['pallet'] + self.assertEqual('P-001', pallet.pallet_code) + self.assertEqual('server-fixed-token', pallet.qr_token) + self.assertEqual(20, pallet.capacity_packages) + self.assertEqual('0', pallet.asset_status) + self.assertEqual('admin', pallet.create_by) + self.assertEqual(self.fixed_time, pallet.create_time) + self.assertEqual('admin', pallet.update_by) + self.assertEqual(self.fixed_time, pallet.update_time) + inventory = captured['inventory'] + self.assertEqual(17, inventory.pallet_id) + self.assertEqual('UNSTOCKED', inventory.flow_state) + self.assertEqual(0, inventory.remaining_packages) + self.assertEqual(0, inventory.version) + self.assertEqual(['pallet', 'inventory', 'commit'], actions) + self.db.commit.assert_awaited_once() + self.db.rollback.assert_not_awaited() + self.assertIsInstance(result, CrudResponseModel) + self.assertTrue(result.is_success) + + async def test_add_rolls_back_when_pallet_flush_fails(self) -> None: + failure = RuntimeError('flush failed') + add_inventory = AsyncMock() + with ( + patch.object(MasterDao, 'find_pallet_by_code', new=AsyncMock(return_value=None)), + patch.object(MasterDao, 'add_pallet', new=AsyncMock(side_effect=failure)), + patch.object(MasterDao, 'add_pallet_inventory', new=add_inventory), + self.assertRaisesRegex(RuntimeError, 'flush failed'), + ): + await PalletService.add_pallet(self.db, self.create, current_user_name='admin') + + add_inventory.assert_not_awaited() + self.db.commit.assert_not_awaited() + self.db.rollback.assert_awaited_once() + + async def test_add_rolls_back_when_commit_fails(self) -> None: + async def assign_id(_db: AsyncSession, pallet: BizPallet) -> BizPallet: + pallet.pallet_id = 17 + return pallet + + self.db.commit.side_effect = RuntimeError('commit failed') + with ( + patch.object(MasterDao, 'find_pallet_by_code', new=AsyncMock(return_value=None)), + patch.object(MasterDao, 'add_pallet', new=AsyncMock(side_effect=assign_id)), + patch.object(MasterDao, 'add_pallet_inventory', new=AsyncMock()), + self.assertRaisesRegex(RuntimeError, 'commit failed'), + ): + await PalletService.add_pallet(self.db, self.create, current_user_name='admin') + + self.db.rollback.assert_awaited_once() + + async def test_add_converts_integrity_error_to_stable_service_exception(self) -> None: + with ( + patch.object(MasterDao, 'find_pallet_by_code', new=AsyncMock(return_value=None)), + patch.object(MasterDao, 'add_pallet', new=AsyncMock(side_effect=integrity_error())), + self.assertRaises(ServiceException) as raised, + ): + await PalletService.add_pallet(self.db, self.create, current_user_name='admin') + + self.assertEqual('新增托盘失败，托盘编号或二维码标识已存在', raised.exception.message) + self.db.rollback.assert_awaited_once() + self.db.commit.assert_not_awaited() + + async def test_add_rejects_duplicate_code_before_writing(self) -> None: + add_pallet = AsyncMock() + with ( + patch.object(MasterDao, 'find_pallet_by_code', new=AsyncMock(return_value=object())), + patch.object(MasterDao, 'add_pallet', new=add_pallet), + self.assertRaises(ServiceException) as raised, + ): + await PalletService.add_pallet(self.db, self.create, current_user_name='admin') + + self.assertEqual('托盘编号已存在', raised.exception.message) + add_pallet.assert_not_awaited() + self.db.rollback.assert_awaited_once() + + +class TestPalletQrService(unittest.IsolatedAsyncioTestCase): + def test_render_qr_png_returns_real_stable_png(self) -> None: + first = PalletService.render_qr_png('fixed-original-token') + second = PalletService.render_qr_png('fixed-original-token') + + self.assertEqual(b'\x89PNG\r\n\x1a\n', first[:8]) + self.assertEqual(first, second) + + async def test_get_qrcode_uses_original_persisted_token(self) -> None: + db = MagicMock(spec=AsyncSession) + with ( + patch.object( + MasterDao, + 'get_pallet', + new=AsyncMock(return_value=SimpleNamespace(qr_token='persisted-token')), + ), + patch.object(PalletService, 'render_qr_png', return_value=b'png-bytes') as render, + patch.object(secrets, 'token_urlsafe') as token_urlsafe, + ): + actual = await PalletService.get_pallet_qrcode(db, pallet_id=7) + + self.assertEqual(b'png-bytes', actual) + render.assert_called_once_with('persisted-token') + token_urlsafe.assert_not_called() + + async def test_get_qrcode_rejects_missing_pallet(self) -> None: + db = MagicMock(spec=AsyncSession) + with ( + patch.object(MasterDao, 'get_pallet', new=AsyncMock(return_value=None)), + self.assertRaises(ServiceException) as raised, + ): + await PalletService.get_pallet_qrcode(db, pallet_id=404) + + self.assertEqual('托盘不存在', raised.exception.message) + + +class TestPalletQueryAndEditService(unittest.IsolatedAsyncioTestCase): + def setUp(self) -> None: + self.db = MagicMock(spec=AsyncSession) + + async def test_list_computes_can_remove_from_inventory_and_events(self) -> None: + page = PageModel[Any]( + rows=[make_pallet_record(1), make_pallet_record(2)], + pageNum=1, + pageSize=10, + total=2, + hasNext=False, + ) + query = PalletPageQueryModel() + with ( + patch.object(MasterDao, 'list_pallet', new=AsyncMock(return_value=page)), + patch.object( + MasterDao, + 'get_pallet_inventory', + new=AsyncMock( + side_effect=[ + SimpleNamespace(flow_state='UNSTOCKED'), + SimpleNamespace(flow_state='UNSTOCKED'), + ] + ), + ), + patch.object(MasterDao, 'count_pallet_events', new=AsyncMock(side_effect=[0, 1])), + ): + actual = await PalletService.list_pallet(self.db, query, is_page=True) + + self.assertIsInstance(actual, PageModel) + self.assertTrue(all(isinstance(row, PalletModel) for row in actual.rows)) + self.assertEqual([True, False], [row.can_remove for row in actual.rows]) + + async def test_detail_computes_false_when_inventory_is_not_unstocked(self) -> None: + with ( + patch.object(MasterDao, 'get_pallet', new=AsyncMock(return_value=make_pallet_record(3))), + patch.object( + MasterDao, + 'get_pallet_inventory', + new=AsyncMock(return_value=SimpleNamespace(flow_state='EMPTY_IN_WAREHOUSE')), + ), + patch.object(MasterDao, 'count_pallet_events', new=AsyncMock(return_value=0)), + ): + actual = await PalletService.detail_pallet(self.db, pallet_id=3) + + self.assertIsInstance(actual, PalletModel) + self.assertFalse(actual.can_remove) + + async def test_detail_rejects_missing_pallet(self) -> None: + with ( + patch.object(MasterDao, 'get_pallet', new=AsyncMock(return_value=None)), + self.assertRaises(ServiceException) as raised, + ): + await PalletService.detail_pallet(self.db, pallet_id=404) + + self.assertEqual('托盘不存在', raised.exception.message) + + async def test_edit_updates_only_mutable_fields_with_server_audit(self) -> None: + edit = PalletEditModel(palletId=3, capacityPackages=12, assetStatus='1', remark='停用') + updated_at = datetime(2026, 7, 26, 12, 30) + update_pallet = AsyncMock() + lock_actions: list[str] = [] + + async def get_pallet(*_args: Any, **_kwargs: Any) -> SimpleNamespace: + lock_actions.append('pallet') + return make_pallet_record(3) + + async def get_inventory(*_args: Any, **_kwargs: Any) -> SimpleNamespace: + lock_actions.append('inventory') + return SimpleNamespace(remaining_packages=12) + + get_pallet_mock = AsyncMock(side_effect=get_pallet) + get_inventory_mock = AsyncMock(side_effect=get_inventory) + with ( + patch.object(MasterDao, 'get_pallet', new=get_pallet_mock), + patch.object(MasterDao, 'get_pallet_inventory', new=get_inventory_mock), + patch.object(MasterDao, 'update_pallet', new=update_pallet), + patch.object(pallet_service_module, 'datetime') as service_datetime, + ): + service_datetime.now.return_value = updated_at + result = await PalletService.edit_pallet(self.db, edit, current_user_name='admin') + + update_pallet.assert_awaited_once_with( + self.db, + edit, + update_by='admin', + update_time=updated_at, + ) + self.assertEqual(['inventory', 'pallet'], lock_actions) + get_pallet_mock.assert_awaited_once_with(self.db, 3, for_update=True) + get_inventory_mock.assert_awaited_once_with(self.db, 3, for_update=True) + self.db.commit.assert_awaited_once() + self.db.rollback.assert_not_awaited() + self.assertTrue(result.is_success) + + async def test_edit_rejects_capacity_below_remaining_packages(self) -> None: + edit = PalletEditModel(palletId=3, capacityPackages=11, assetStatus='0') + update_pallet = AsyncMock() + with ( + patch.object(MasterDao, 'get_pallet', new=AsyncMock(return_value=make_pallet_record(3))), + patch.object( + MasterDao, + 'get_pallet_inventory', + new=AsyncMock(return_value=SimpleNamespace(remaining_packages=12)), + ), + patch.object(MasterDao, 'update_pallet', new=update_pallet), + self.assertRaises(ServiceException) as raised, + ): + await PalletService.edit_pallet(self.db, edit, current_user_name='admin') + + self.assertEqual('托盘容量不能小于当前剩余包数', raised.exception.message) + update_pallet.assert_not_awaited() + self.db.commit.assert_not_awaited() + self.db.rollback.assert_awaited_once() + + async def test_edit_rejects_missing_inventory(self) -> None: + edit = PalletEditModel(palletId=3, capacityPackages=20, assetStatus='0') + with ( + patch.object(MasterDao, 'get_pallet', new=AsyncMock(return_value=make_pallet_record(3))), + patch.object(MasterDao, 'get_pallet_inventory', new=AsyncMock(return_value=None)), + self.assertRaises(ServiceException) as raised, + ): + await PalletService.edit_pallet(self.db, edit, current_user_name='admin') + + self.assertEqual('托盘库存不存在', raised.exception.message) + self.db.rollback.assert_awaited_once() + + +class TestPalletRemovalService(unittest.IsolatedAsyncioTestCase): + def setUp(self) -> None: + self.db = MagicMock(spec=AsyncSession) + + async def test_batch_remove_preflights_all_then_deletes_inventory_before_pallet(self) -> None: + actions: list[str] = [] + + async def get_pallet( + _db: AsyncSession, + pallet_id: int, + *, + for_update: bool = False, + ) -> SimpleNamespace: + actions.append(f'get-pallet-{pallet_id}-{for_update}') + return make_pallet_record(pallet_id) + + async def get_inventory( + _db: AsyncSession, + pallet_id: int, + *, + for_update: bool = False, + ) -> SimpleNamespace: + actions.append(f'get-inventory-{pallet_id}-{for_update}') + return SimpleNamespace(flow_state='UNSTOCKED') + + async def count_events(_db: AsyncSession, pallet_id: int) -> int: + actions.append(f'count-events-{pallet_id}') + return 0 + + async def delete_inventory(_db: AsyncSession, pallet_id: int) -> None: + actions.append(f'delete-inventory-{pallet_id}') + + async def delete_pallet(_db: AsyncSession, pallet_id: int) -> None: + actions.append(f'delete-pallet-{pallet_id}') + + async def commit() -> None: + actions.append('commit') + + self.db.commit.side_effect = commit + with ( + patch.object(MasterDao, 'get_pallet', new=AsyncMock(side_effect=get_pallet)), + patch.object(MasterDao, 'get_pallet_inventory', new=AsyncMock(side_effect=get_inventory)), + patch.object(MasterDao, 'count_pallet_events', new=AsyncMock(side_effect=count_events)), + patch.object(MasterDao, 'delete_pallet_inventory', new=AsyncMock(side_effect=delete_inventory)), + patch.object(MasterDao, 'delete_pallet', new=AsyncMock(side_effect=delete_pallet)), + ): + result = await PalletService.remove_pallet(self.db, DeletePalletModel(palletIds=[5, 4])) + + self.assertEqual( + [ + 'get-inventory-4-True', + 'get-pallet-4-True', + 'count-events-4', + 'get-inventory-5-True', + 'get-pallet-5-True', + 'count-events-5', + 'delete-inventory-4', + 'delete-pallet-4', + 'delete-inventory-5', + 'delete-pallet-5', + 'commit', + ], + actions, + ) + self.db.commit.assert_awaited_once() + self.db.rollback.assert_not_awaited() + self.assertTrue(result.is_success) + + async def test_remove_rejects_invalid_state_before_any_delete(self) -> None: + cases = ( + ('托盘不存在', None, None, 0), + ('托盘库存不存在', make_pallet_record(4), None, 0), + ( + '托盘未处于未入库状态，不能删除', + make_pallet_record(4), + SimpleNamespace(flow_state='EMPTY_IN_WAREHOUSE'), + 0, + ), + ( + '托盘已有流转事件，不能删除', + make_pallet_record(4), + SimpleNamespace(flow_state='UNSTOCKED'), + 1, + ), + ) + for message, pallet, inventory, event_count in cases: + with self.subTest(message=message): + self.db.reset_mock() + delete_inventory = AsyncMock() + delete_pallet = AsyncMock() + with ( + patch.object(MasterDao, 'get_pallet', new=AsyncMock(return_value=pallet)), + patch.object(MasterDao, 'get_pallet_inventory', new=AsyncMock(return_value=inventory)), + patch.object(MasterDao, 'count_pallet_events', new=AsyncMock(return_value=event_count)), + patch.object(MasterDao, 'delete_pallet_inventory', new=delete_inventory), + patch.object(MasterDao, 'delete_pallet', new=delete_pallet), + self.assertRaises(ServiceException) as raised, + ): + await PalletService.remove_pallet(self.db, 4) + + self.assertEqual(message, raised.exception.message) + delete_inventory.assert_not_awaited() + delete_pallet.assert_not_awaited() + self.db.commit.assert_not_awaited() + self.db.rollback.assert_awaited_once() + + async def test_remove_integrity_error_rolls_back_with_stable_failure(self) -> None: + with ( + patch.object(MasterDao, 'get_pallet', new=AsyncMock(return_value=make_pallet_record(4))), + patch.object( + MasterDao, + 'get_pallet_inventory', + new=AsyncMock(return_value=SimpleNamespace(flow_state='UNSTOCKED')), + ), + patch.object(MasterDao, 'count_pallet_events', new=AsyncMock(return_value=0)), + patch.object(MasterDao, 'delete_pallet_inventory', new=AsyncMock()), + patch.object(MasterDao, 'delete_pallet', new=AsyncMock(side_effect=integrity_error())), + self.assertRaises(ServiceException) as raised, + ): + await PalletService.remove_pallet(self.db, 4) + + self.assertEqual('删除托盘失败，托盘状态或关联数据已变化', raised.exception.message) + self.db.rollback.assert_awaited_once() + self.db.commit.assert_not_awaited() + + +class TestBusinessUserAssignmentService(unittest.IsolatedAsyncioTestCase): + def setUp(self) -> None: + self.db = MagicMock(spec=AsyncSession) + self.users = [ + SimpleNamespace( + user_id=2, + user_name='alice', + nick_name='艾丽丝', + password='secret-a', + email='alice@example.com', + phonenumber='13800000000', + ), + SimpleNamespace( + user_id=5, + user_name='bob', + nick_name=None, + password='secret-b', + email='bob@example.com', + phonenumber='13900000000', + ), + ] + + async def test_assignment_get_returns_only_safe_fields_with_assigned_marker(self) -> None: + cases = ( + ('warehouse', 7, [7], []), + ('site', 9, [9], []), + ) + for resource, resource_id, first_ids, second_ids in cases: + with self.subTest(resource=resource): + association_method = f'list_user_{resource}_ids' + with ( + patch.object( + MasterDao, + f'get_{resource}', + new=AsyncMock(return_value=SimpleNamespace()), + ), + patch.object(MasterDao, 'list_assignable_users', new=AsyncMock(return_value=self.users)), + patch.object( + AccessDao, + association_method, + new=AsyncMock(side_effect=[first_ids, second_ids]), + ) as list_ids, + ): + actual = await getattr(PalletService, f'list_{resource}_users')(self.db, resource_id) + + self.assertEqual( + [ + {'userId': 2, 'userName': 'alice', 'nickName': '艾丽丝', 'assigned': True}, + {'userId': 5, 'userName': 'bob', 'nickName': None, 'assigned': False}, + ], + [option.model_dump(by_alias=True) for option in actual], + ) + self.assertTrue(all(isinstance(option, BusinessUserOptionModel) for option in actual)) + self.assertEqual( + [(self.db, 2), (self.db, 5)], + [call.args for call in list_ids.await_args_list], + ) + + async def test_assignment_get_rejects_missing_resource_before_user_query(self) -> None: + cases = (('warehouse', 7, '仓库不存在'), ('site', 9, '客户现场不存在')) + for resource, resource_id, message in cases: + with self.subTest(resource=resource): + list_users = AsyncMock() + with ( + patch.object(MasterDao, f'get_{resource}', new=AsyncMock(return_value=None)), + patch.object(MasterDao, 'list_assignable_users', new=list_users), + self.assertRaises(ServiceException) as raised, + ): + await getattr(PalletService, f'list_{resource}_users')(self.db, resource_id) + + self.assertEqual(message, raised.exception.message) + list_users.assert_not_awaited() + + async def test_assignment_put_validates_users_replaces_and_commits(self) -> None: + assignment = UserAssignModel(userIds=[2, 5]) + for resource, resource_id in (('warehouse', 7), ('site', 9)): + with self.subTest(resource=resource): + self.db.reset_mock() + replace = AsyncMock() + with ( + patch.object( + MasterDao, + f'get_{resource}', + new=AsyncMock(return_value=SimpleNamespace()), + ), + patch.object( + MasterDao, + 'list_assignable_users_by_ids', + new=AsyncMock(return_value=self.users), + ) as list_users, + patch.object(AccessDao, f'replace_{resource}_users', new=replace), + ): + result = await getattr(PalletService, f'assign_{resource}_users')( + self.db, + resource_id, + assignment, + ) + + list_users.assert_awaited_once_with(self.db, [2, 5]) + replace.assert_awaited_once_with(self.db, resource_id, [2, 5]) + self.db.commit.assert_awaited_once() + self.db.rollback.assert_not_awaited() + self.assertTrue(result.is_success) + + async def test_assignment_put_rejects_missing_disabled_or_deleted_user(self) -> None: + replace = AsyncMock() + assignment = UserAssignModel(userIds=[2, 5]) + with ( + patch.object(MasterDao, 'get_warehouse', new=AsyncMock(return_value=SimpleNamespace())), + patch.object( + MasterDao, + 'list_assignable_users_by_ids', + new=AsyncMock(return_value=[self.users[0]]), + ), + patch.object(AccessDao, 'replace_warehouse_users', new=replace), + self.assertRaises(ServiceException) as raised, + ): + await PalletService.assign_warehouse_users(self.db, 7, assignment) + + self.assertEqual('存在不存在、停用或已删除的用户', raised.exception.message) + replace.assert_not_awaited() + self.db.commit.assert_not_awaited() + self.db.rollback.assert_awaited_once() + + async def test_replace_failure_rolls_back_and_preserves_original_exception(self) -> None: + assignment = UserAssignModel(userIds=[2, 5]) + with ( + patch.object(MasterDao, 'get_site', new=AsyncMock(return_value=SimpleNamespace())), + patch.object( + MasterDao, + 'list_assignable_users_by_ids', + new=AsyncMock(return_value=self.users), + ), + patch.object( + AccessDao, + 'replace_site_users', + new=AsyncMock(side_effect=RuntimeError('replace failed')), + ), + self.assertRaisesRegex(RuntimeError, 'replace failed'), + ): + await PalletService.assign_site_users(self.db, 9, assignment) + + self.db.commit.assert_not_awaited() + self.db.rollback.assert_awaited_once() + + async def test_assignment_commit_failure_rolls_back(self) -> None: + assignment = UserAssignModel(userIds=[2, 5]) + self.db.commit.side_effect = RuntimeError('commit failed') + with ( + patch.object(MasterDao, 'get_warehouse', new=AsyncMock(return_value=SimpleNamespace())), + patch.object( + MasterDao, + 'list_assignable_users_by_ids', + new=AsyncMock(return_value=self.users), + ), + patch.object(AccessDao, 'replace_warehouse_users', new=AsyncMock()), + self.assertRaisesRegex(RuntimeError, 'commit failed'), + ): + await PalletService.assign_warehouse_users(self.db, 7, assignment) + + self.db.rollback.assert_awaited_once() + + async def test_assignment_integrity_error_has_stable_message_and_rolls_back(self) -> None: + assignment = UserAssignModel(userIds=[2, 5]) + with ( + patch.object(MasterDao, 'get_warehouse', new=AsyncMock(return_value=SimpleNamespace())), + patch.object( + MasterDao, + 'list_assignable_users_by_ids', + new=AsyncMock(return_value=self.users), + ), + patch.object( + AccessDao, + 'replace_warehouse_users', + new=AsyncMock(side_effect=integrity_error()), + ), + self.assertRaises(ServiceException) as raised, + ): + await PalletService.assign_warehouse_users(self.db, 7, assignment) + + self.assertEqual('分配仓库用户失败，关联数据无效', raised.exception.message) + self.db.rollback.assert_awaited_once() + + +class TestPalletAndAssignmentControllers(unittest.IsolatedAsyncioTestCase): + @staticmethod + def _routers() -> tuple[Any, ...]: + return ( + pallet_controller_module.pallet_controller, + warehouse_controller_module.warehouse_user_controller, + site_controller_module.site_user_controller, + ) + + @classmethod + def _routes(cls) -> dict[tuple[str, str], APIRoute]: + routes = {} + for router in cls._routers(): + for route in router.routes: + if isinstance(route, APIRoute): + routes[(route.path, next(iter(route.methods)))] = route + return routes + + def test_openapi_has_all_pallet_and_assignment_operations(self) -> None: + app = FastAPI() + for router in self._routers(): + app.include_router(router) + schema = app.openapi() + + expected_operations = { + ('/master/pallet/list', 'get'), + ('/master/pallet/{pallet_id}', 'get'), + ('/master/pallet', 'post'), + ('/master/pallet', 'put'), + ('/master/pallet/{pallet_ids}', 'delete'), + ('/master/pallet/{pallet_id}/qrcode', 'get'), + ('/master/warehouse/{warehouse_id}/users', 'get'), + ('/master/warehouse/{warehouse_id}/users', 'put'), + ('/master/site/{site_id}/users', 'get'), + ('/master/site/{site_id}/users', 'put'), + } + for path, method in expected_operations: + with self.subTest(path=path, method=method): + self.assertIn(path, schema['paths']) + self.assertIn(method, schema['paths'][path]) + self.assertIn( + 'image/png', + schema['paths']['/master/pallet/{pallet_id}/qrcode']['get']['responses']['200']['content'], + ) + + def test_route_response_models_are_exact(self) -> None: + routes = self._routes() + self.assertEqual( + PageResponseModel[PalletModel], + routes[('/master/pallet/list', 'GET')].response_model, + ) + self.assertEqual( + DataResponseModel[PalletModel], + routes[('/master/pallet/{pallet_id}', 'GET')].response_model, + ) + for path, method in ( + ('/master/pallet', 'POST'), + ('/master/pallet', 'PUT'), + ('/master/pallet/{pallet_ids}', 'DELETE'), + ('/master/warehouse/{warehouse_id}/users', 'PUT'), + ('/master/site/{site_id}/users', 'PUT'), + ): + self.assertEqual(ResponseBaseModel, routes[(path, method)].response_model) + user_response = DataResponseModel[list[BusinessUserOptionModel]] + self.assertEqual( + user_response, + routes[('/master/warehouse/{warehouse_id}/users', 'GET')].response_model, + ) + self.assertEqual( + user_response, + routes[('/master/site/{site_id}/users', 'GET')].response_model, + ) + self.assertIsNone(routes[('/master/pallet/{pallet_id}/qrcode', 'GET')].response_model) + + def test_permissions_are_exact_and_all_routes_require_pre_auth(self) -> None: + expected_permissions = { + ('/master/pallet/list', 'GET'): 'master:pallet:list', + ('/master/pallet/{pallet_id}', 'GET'): 'master:pallet:query', + ('/master/pallet', 'POST'): 'master:pallet:add', + ('/master/pallet', 'PUT'): 'master:pallet:edit', + ('/master/pallet/{pallet_ids}', 'DELETE'): 'master:pallet:remove', + ('/master/pallet/{pallet_id}/qrcode', 'GET'): 'master:pallet:print', + ('/master/warehouse/{warehouse_id}/users', 'GET'): 'master:warehouse:assign', + ('/master/warehouse/{warehouse_id}/users', 'PUT'): 'master:warehouse:assign', + ('/master/site/{site_id}/users', 'GET'): 'master:site:assign', + ('/master/site/{site_id}/users', 'PUT'): 'master:site:assign', + } + + actual_permissions = {} + for key, route in self._routes().items(): + dependencies = [ + dependency.dependency + for dependency in route.dependencies + if isinstance(dependency.dependency, CheckUserInterfaceAuth) + ] + self.assertEqual(1, len(dependencies)) + self.assertTrue(any(isinstance(dependency.dependency, PreAuth) for dependency in route.dependencies)) + actual_permissions[key] = dependencies[0].perm + + self.assertEqual(expected_permissions, actual_permissions) + self.assertEqual(8, len(set(actual_permissions.values()))) + + def test_write_routes_have_log_and_current_user(self) -> None: + write_routes = ( + ('/master/pallet', 'POST'), + ('/master/pallet', 'PUT'), + ('/master/pallet/{pallet_ids}', 'DELETE'), + ('/master/warehouse/{warehouse_id}/users', 'PUT'), + ('/master/site/{site_id}/users', 'PUT'), + ) + routes = self._routes() + for key in write_routes: + route = routes[key] + self.assertTrue(hasattr(route.endpoint, '__wrapped__')) + self.assertTrue( + any( + dependency.name == 'current_user' and dependency.call == LoginService.get_current_user + for dependency in route.dependant.dependencies + ) + ) + + def test_qrcode_route_is_declared_before_dynamic_detail(self) -> None: + route_paths = [route.path for route in pallet_controller_module.pallet_controller.routes] + self.assertLess( + route_paths.index('/master/pallet/{pallet_id}/qrcode'), + route_paths.index('/master/pallet/{pallet_id}'), + ) + + async def test_qrcode_endpoint_returns_real_image_png_body(self) -> None: + png = PalletService.render_qr_png('persisted-token') + db = MagicMock(spec=AsyncSession) + with patch.object(PalletService, 'get_pallet_qrcode', new=AsyncMock(return_value=png)) as get_qrcode: + response = await pallet_controller_module.query_master_pallet_qrcode( + request=MagicMock(), + pallet_id=7, + query_db=db, + ) + + get_qrcode.assert_awaited_once_with(db, 7) + self.assertEqual('image/png', response.media_type) + self.assertEqual(b'\x89PNG\r\n\x1a\n', response.body[:8]) + + def test_controller_sources_do_not_require_system_user_list_or_cache(self) -> None: + sources = '\n'.join( + inspect.getsource(module) + for module in ( + pallet_controller_module, + warehouse_controller_module, + site_controller_module, + ) + ) + self.assertNotIn('system:user:list', sources) + self.assertNotIn('ApiCache', sources) + + +if __name__ == '__main__': + unittest.main()