diff --git a/shierfeng-fastapi-backend/module_business/controller/filling_controller.py b/shierfeng-fastapi-backend/module_business/controller/filling_controller.py new file mode 100644 index 0000000..01c5faa --- /dev/null +++ b/shierfeng-fastapi-backend/module_business/controller/filling_controller.py @@ -0,0 +1,125 @@ +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, + FillingCreateModel, + FillingModel, + FillingPageQueryModel, + PalletScanResponseModel, +) +from module_business.service.filling_service import FillingService +from utils.response_util import ResponseUtil + +filling_controller = APIRouterPro( + prefix='/business/filling', + tags=['业务管理-托盘装填'], + dependencies=[PreAuthDependency()], +) + + +@filling_controller.get( + '/list', + summary='获取托盘装填记录分页列表接口', + description='按当前用户仓库数据范围获取托盘装填记录分页列表', + response_model=PageResponseModel[FillingModel], + dependencies=[UserInterfaceAuthDependency('business:filling:list')], +) +async def get_business_filling_list( + request: Request, + filling_page_query: Annotated[FillingPageQueryModel, Query()], + query_db: Annotated[AsyncSession, DBSessionDependency()], + current_user: Annotated[CurrentUserModel, CurrentUserDependency()], +) -> Response: + result = await FillingService.get_filling_list_services( + query_db, + filling_page_query, + current_user, + is_page=True, + ) + return ResponseUtil.success(model_content=result) + + +@filling_controller.get( + '/strain-options', + summary='获取装填菌种选项接口', + description='获取全部启用菌种的业务选项', + response_model=DataResponseModel[list[BusinessOptionModel]], + dependencies=[UserInterfaceAuthDependency('business:filling:query')], +) +async def get_business_filling_strain_options( + request: Request, + query_db: Annotated[AsyncSession, DBSessionDependency()], + current_user: Annotated[CurrentUserModel, CurrentUserDependency()], +) -> Response: + result = await FillingService.get_strain_options_services(query_db, current_user) + return ResponseUtil.success(data=result) + + +@filling_controller.get( + '/scan/{qr_token}', + summary='扫描待装填托盘接口', + description='按托盘当前仓库数据范围判断空托盘是否可装填', + response_model=DataResponseModel[PalletScanResponseModel], + dependencies=[UserInterfaceAuthDependency('business:filling:query')], +) +async def scan_business_filling_pallet( + request: Request, + qr_token: Annotated[ + str, + Path(min_length=1, max_length=128, pattern=r'\S', description='二维码固定标识'), + ], + query_db: Annotated[AsyncSession, DBSessionDependency()], + current_user: Annotated[CurrentUserModel, CurrentUserDependency()], +) -> Response: + result = await FillingService.scan_filling_pallet_services( + query_db, + qr_token, + current_user, + ) + return ResponseUtil.success(data=result) + + +@filling_controller.get( + '/{filling_id}', + summary='获取托盘装填记录详情接口', + description='按仓库数据范围获取指定托盘装填记录', + response_model=DataResponseModel[FillingModel], + dependencies=[UserInterfaceAuthDependency('business:filling:query')], +) +async def query_detail_business_filling( + request: Request, + filling_id: Annotated[int, Path(gt=0, description='装填记录ID')], + query_db: Annotated[AsyncSession, DBSessionDependency()], + current_user: Annotated[CurrentUserModel, CurrentUserDependency()], +) -> Response: + result = await FillingService.filling_detail_services(query_db, filling_id, current_user) + return ResponseUtil.success(data=result) + + +@filling_controller.post( + '', + summary='单托盘装填接口', + description='在仓库内即时确认单个空托盘装填', + response_model=ResponseBaseModel, + dependencies=[UserInterfaceAuthDependency('business:filling:add')], +) +@Log(title='托盘装填', business_type=BusinessType.INSERT) +async def add_business_filling( + request: Request, + add_filling: FillingCreateModel, + query_db: Annotated[AsyncSession, DBSessionDependency()], + current_user: Annotated[CurrentUserModel, CurrentUserDependency()], +) -> Response: + result = await FillingService.fill_pallet_services(query_db, add_filling, 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 a6be1b1..9df1773 100644 --- a/shierfeng-fastapi-backend/module_business/dao/operation_dao.py +++ b/shierfeng-fastapi-backend/module_business/dao/operation_dao.py @@ -60,6 +60,15 @@ class OperationDao: scoped_warehouse_ids = sorted(set(warehouse_ids)) return [BizInboundOrder.warehouse_id.in_(scoped_warehouse_ids) if scoped_warehouse_ids else false()] + @staticmethod + def _filling_scope_conditions( + warehouse_ids: list[int] | None, + ) -> list[ColumnElement[bool]]: + if warehouse_ids is None: + return [] + scoped_warehouse_ids = sorted(set(warehouse_ids)) + return [BizFillingRecord.warehouse_id.in_(scoped_warehouse_ids) if scoped_warehouse_ids else false()] + @staticmethod async def _paginate( db: AsyncSession, @@ -97,8 +106,9 @@ class OperationDao: model: type[Any], key_column: InstrumentedAttribute, idempotency_key: str, + *extra_conditions: ColumnElement[bool], ) -> Any | None: - result = await db.execute(select(model).where(key_column == idempotency_key)) + result = await db.execute(select(model).where(key_column == idempotency_key, *extra_conditions)) return result.scalar_one_or_none() @staticmethod @@ -300,12 +310,14 @@ class OperationDao: cls, db: AsyncSession, idempotency_key: str, + warehouse_ids: list[int] | None = None, ) -> BizFillingRecord | None: return await cls._get_by_key( db, BizFillingRecord, BizFillingRecord.idempotency_key, idempotency_key, + *cls._filling_scope_conditions(warehouse_ids), ) @classmethod @@ -437,8 +449,9 @@ class OperationDao: db: AsyncSession, query_object: FillingPageQueryModel, is_page: bool = True, + warehouse_ids: list[int] | None = None, ) -> PageModel | list[dict[str, Any]]: - conditions = [] + conditions = cls._filling_scope_conditions(warehouse_ids) if query_object.filling_no: conditions.append(BizFillingRecord.filling_no.like(f'%{query_object.filling_no}%')) if query_object.warehouse_id: @@ -470,12 +483,14 @@ class OperationDao: cls, db: AsyncSession, filling_id: int, + warehouse_ids: list[int] | None = None, ) -> BizFillingRecord | None: return await cls._get_by_id( db, BizFillingRecord, BizFillingRecord.filling_id, filling_id, + *cls._filling_scope_conditions(warehouse_ids), ) @classmethod diff --git a/shierfeng-fastapi-backend/module_business/service/filling_service.py b/shierfeng-fastapi-backend/module_business/service/filling_service.py new file mode 100644 index 0000000..c8e4515 --- /dev/null +++ b/shierfeng-fastapi-backend/module_business/service/filling_service.py @@ -0,0 +1,464 @@ +import hashlib +import json +import secrets +from datetime import datetime +from typing import Any, NoReturn + +from sqlalchemy.exc import IntegrityError +from sqlalchemy.ext.asyncio import AsyncSession + +from common.vo import CrudResponseModel, PageModel +from exceptions.exception import PermissionException, ServiceException +from module_admin.entity.vo.user_vo import CurrentUserModel +from module_business.dao.access_dao import AccessDao +from module_business.dao.inventory_dao import InventoryDao +from module_business.dao.master_dao import MasterDao +from module_business.dao.operation_dao import OperationDao +from module_business.domain.enums import PalletEventType, PalletFlowState +from module_business.domain.pallet_state_machine import PalletStateMachine +from module_business.entity.do.master_do import BizPallet, BizStrain +from module_business.entity.do.operation_do import BizFillingRecord, BizPalletEvent, BizPalletInventory +from module_business.entity.vo.master_vo import StrainPageQueryModel +from module_business.entity.vo.operation_vo import ( + BusinessOptionModel, + FillingCreateModel, + FillingModel, + FillingPageQueryModel, + PalletScanResponseModel, +) +from module_business.service.access_service import AccessService + + +class FillingService: + """仓库内单托盘即时装填服务层。""" + + @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_request_hash(page_object: FillingCreateModel) -> str: + payload = json.dumps( + { + 'qr_token': page_object.qr_token.strip(), + 'warehouse_id': page_object.warehouse_id, + 'strain_id': page_object.strain_id, + 'batch_no': page_object.batch_no.strip(), + 'package_spec': page_object.package_spec.strip(), + 'initial_packages': page_object.initial_packages, + }, + 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))) + + @classmethod + def _raise_filling_not_accessible(cls, current_user: CurrentUserModel) -> NoReturn: + if cls._is_admin(current_user): + raise ServiceException(message='装填记录不存在') + raise PermissionException(data='', message='无权访问该装填记录') + + @classmethod + def _raise_pallet_not_accessible( + cls, + current_user: CurrentUserModel, + admin_message: str = '托盘不存在', + ) -> NoReturn: + if cls._is_admin(current_user): + raise ServiceException(message=admin_message) + raise PermissionException(data='', message='无权访问该托盘') + + @staticmethod + def _validate_empty_warehouse_pallet( + pallet: BizPallet, + inventory: BizPalletInventory, + warehouse_id: int, + qr_token: str, + ) -> None: + if pallet.pallet_id != inventory.pallet_id or pallet.qr_token != qr_token: + raise ServiceException(message='托盘二维码与库存不一致') + if pallet.asset_status != '0': + raise ServiceException(message='托盘资产已停用，不允许装填') + if ( + inventory.flow_state != PalletFlowState.EMPTY_IN_WAREHOUSE.value + or inventory.warehouse_id != warehouse_id + 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 type(inventory.version) is not int or inventory.version < 0: + raise ServiceException(message='托盘库存版本无效') + + @staticmethod + def _validate_package_count(initial_packages: int, pallet: BizPallet) -> None: + if type(initial_packages) is not int or initial_packages <= 0: + raise ServiceException(message='装填包数必须为正整数') + if initial_packages > pallet.capacity_packages: + raise ServiceException(message='装填包数不能超过托盘容量') + + @staticmethod + def _validate_active_strain(strain: BizStrain | None) -> BizStrain: + if strain is None: + raise ServiceException(message='菌种不存在') + if strain.status != '0': + raise ServiceException(message='菌种已停用') + return strain + + @staticmethod + def _is_exact_idempotent_result( + filling: BizFillingRecord, + page_object: FillingCreateModel, + request_hash: str, + ) -> bool: + return ( + filling.idempotency_key == page_object.idempotency_key + and filling.warehouse_id == page_object.warehouse_id + and filling.request_hash == request_hash + ) + + @staticmethod + def _success(filling_id: int) -> CrudResponseModel: + return CrudResponseModel(is_success=True, message='装填成功', result=filling_id) + + @classmethod + async def get_filling_list_services( + cls, + query_db: AsyncSession, + query_object: FillingPageQueryModel, + current_user: CurrentUserModel, + is_page: bool = True, + ) -> PageModel | list[FillingModel]: + 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_filling_list( + query_db, + query_object, + is_page, + warehouse_ids=warehouse_ids, + ) + rows = result.rows if isinstance(result, PageModel) else result + models = [FillingModel.model_validate(record) for record in rows] + if isinstance(result, PageModel): + return result.model_copy(update={'rows': models}) + return models + + @classmethod + async def filling_detail_services( + cls, + query_db: AsyncSession, + filling_id: int, + current_user: CurrentUserModel, + ) -> FillingModel: + warehouse_ids = await cls._warehouse_scope(query_db, current_user) + filling = await OperationDao.get_filling_by_id( + query_db, + filling_id, + warehouse_ids=warehouse_ids, + ) + if filling is None: + cls._raise_filling_not_accessible(current_user) + return FillingModel.model_validate(filling) + + @classmethod + async def get_strain_options_services( + cls, + query_db: AsyncSession, + current_user: CurrentUserModel, + ) -> list[BusinessOptionModel]: + cls._current_identity(current_user) + strains = await MasterDao.list_strain( + query_db, + StrainPageQueryModel(status='0'), + False, + ) + options = [ + BusinessOptionModel( + id=cls._record_value(strain, 'strain_id'), + code=cls._record_value(strain, 'strain_code'), + name=cls._record_value(strain, 'strain_name'), + ) + for strain in strains + if cls._record_value(strain, 'status') == '0' + ] + return sorted(options, key=lambda option: option.id) + + @classmethod + async def scan_filling_pallet_services( + cls, + query_db: AsyncSession, + qr_token: str, + current_user: CurrentUserModel, + ) -> PalletScanResponseModel: + pallet = await InventoryDao.get_pallet_by_qr_token(query_db, qr_token) + if pallet is None: + cls._raise_pallet_not_accessible(current_user) + inventory = await MasterDao.get_pallet_inventory(query_db, pallet.pallet_id) + if inventory is None: + cls._raise_pallet_not_accessible(current_user, '托盘库存不存在') + if inventory.warehouse_id is None: + cls._raise_pallet_not_accessible(current_user, '托盘不在仓库') + await AccessService.assert_warehouse_access( + query_db, + current_user, + inventory.warehouse_id, + ) + cls._validate_empty_warehouse_pallet( + pallet, + inventory, + inventory.warehouse_id, + qr_token, + ) + return PalletScanResponseModel( + pallet_id=pallet.pallet_id, + pallet_code=pallet.pallet_code, + flow_state=PalletFlowState.EMPTY_IN_WAREHOUSE, + warehouse_id=inventory.warehouse_id, + 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=['filling'], + ) + + @classmethod + async def _recover_integrity_error( + cls, + query_db: AsyncSession, + page_object: FillingCreateModel, + request_hash: str, + current_user: CurrentUserModel, + ) -> CrudResponseModel: + try: + warehouse_ids = await cls._warehouse_scope(query_db, current_user) + if warehouse_ids is not None and page_object.warehouse_id not in warehouse_ids: + raise PermissionException(data='', message='无权访问该仓库') + recovered = await OperationDao.get_filling_by_idempotency_key( + query_db, + page_object.idempotency_key, + warehouse_ids=warehouse_ids, + ) + if recovered is None: + if not cls._is_admin(current_user): + raise PermissionException(data='', message='无权访问该装填记录') + raise ServiceException(message='装填失败，数据冲突') + await AccessService.assert_warehouse_access( + query_db, + current_user, + recovered.warehouse_id, + ) + if not cls._is_exact_idempotent_result(recovered, page_object, request_hash): + raise ServiceException(message='装填失败，幂等键已被其他请求使用') + return cls._success(recovered.filling_id) + finally: + await query_db.rollback() + + @classmethod + async def fill_pallet_services( # noqa: PLR0912, PLR0915 + cls, + query_db: AsyncSession, + page_object: FillingCreateModel, + current_user: CurrentUserModel, + ) -> CrudResponseModel: + recovery_transaction_ended = False + try: + request_hash = cls._build_request_hash(page_object) + user_id, user_name = cls._current_identity(current_user) + warehouse_ids = await cls._warehouse_scope(query_db, current_user) + if warehouse_ids is not None and page_object.warehouse_id not in warehouse_ids: + raise PermissionException(data='', message='无权访问该仓库') + + existing = await OperationDao.get_filling_by_idempotency_key( + query_db, + page_object.idempotency_key, + warehouse_ids=warehouse_ids, + ) + if existing is not None: + await AccessService.assert_warehouse_access( + query_db, + current_user, + existing.warehouse_id, + ) + if not cls._is_exact_idempotent_result(existing, page_object, request_hash): + raise ServiceException(message='装填失败，幂等键已被其他请求使用') + filling_id = existing.filling_id + await query_db.rollback() + return cls._success(filling_id) + + await AccessService.assert_warehouse_access( + query_db, + current_user, + page_object.warehouse_id, + ) + warehouse = await MasterDao.get_warehouse(query_db, page_object.warehouse_id) + if warehouse is None: + raise ServiceException(message='仓库不存在') + if warehouse.status != '0': + raise ServiceException(message='仓库已停用') + + scanned_pallet = await InventoryDao.get_pallet_by_qr_token( + query_db, + page_object.qr_token, + ) + if scanned_pallet is None: + cls._raise_pallet_not_accessible(current_user) + + inventory = await InventoryDao.get_inventory_for_update( + query_db, + scanned_pallet.pallet_id, + ) + if inventory is None: + cls._raise_pallet_not_accessible(current_user, '托盘库存不存在') + if inventory.warehouse_id is None: + cls._raise_pallet_not_accessible(current_user, '托盘不在仓库') + await AccessService.assert_warehouse_access( + query_db, + current_user, + inventory.warehouse_id, + ) + locked_pallets = await InventoryDao.get_pallets_by_ids( + query_db, + [scanned_pallet.pallet_id], + ) + if len(locked_pallets) != 1 or locked_pallets[0].pallet_id != scanned_pallet.pallet_id: + cls._raise_pallet_not_accessible(current_user) + pallet = locked_pallets[0] + + if inventory.active_filling_id is not None: + concurrent = await OperationDao.get_active_filling_for_update( + query_db, + inventory.active_filling_id, + ) + if concurrent is not None and concurrent.idempotency_key == page_object.idempotency_key: + # 结束首次非锁定读取建立的 RR 快照，再按最新数据范围恢复并发成功结果。 + await query_db.rollback() + recovery_transaction_ended = True + return await cls._recover_integrity_error( + query_db, + page_object, + request_hash, + current_user, + ) + + cls._validate_empty_warehouse_pallet( + pallet, + inventory, + page_object.warehouse_id, + page_object.qr_token, + ) + cls._validate_package_count(page_object.initial_packages, pallet) + PalletStateMachine.ensure_transition( + PalletFlowState(inventory.flow_state), + PalletFlowState.FILLED_IN_WAREHOUSE, + ) + strain = cls._validate_active_strain(await MasterDao.get_strain(query_db, page_object.strain_id)) + + current_time = datetime.now() + filling = BizFillingRecord( + filling_no=f'FL-{secrets.token_hex(16)}', + pallet_id=pallet.pallet_id, + warehouse_id=inventory.warehouse_id, + strain_id=strain.strain_id, + strain_code_snapshot=strain.strain_code, + strain_name_snapshot=strain.strain_name, + batch_no=page_object.batch_no.strip(), + package_spec=page_object.package_spec.strip(), + initial_packages=page_object.initial_packages, + active_slot='ACTIVE', + idempotency_key=page_object.idempotency_key, + request_hash=request_hash, + operator_id=user_id, + operated_at=current_time, + create_by=user_name, + create_time=current_time, + update_by=user_name, + update_time=current_time, + ) + filling = await OperationDao.add_filling_record(query_db, filling) + filling_id = filling.filling_id + + 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.active_filling_id = filling_id + inventory.remaining_packages = page_object.initial_packages + inventory.flow_state = PalletFlowState.FILLED_IN_WAREHOUSE.value + inventory.version += 1 + inventory.updated_at = current_time + event = BizPalletEvent( + pallet_id=inventory.pallet_id, + event_type=PalletEventType.FILLING_CONFIRMED.value, + source_type='FILLING', + source_id=filling_id, + before_state=before_state, + after_state=PalletFlowState.FILLED_IN_WAREHOUSE.value, + before_warehouse_id=before_warehouse_id, + before_site_id=before_site_id, + before_line_id=before_line_id, + after_warehouse_id=inventory.warehouse_id, + after_site_id=inventory.site_id, + after_line_id=inventory.current_line_id, + before_remaining_packages=before_remaining_packages, + after_remaining_packages=page_object.initial_packages, + operator_id=user_id, + occurred_at=current_time, + idempotency_key=page_object.idempotency_key, + request_hash=request_hash, + ) + await InventoryDao.add_event(query_db, inventory, event) + await query_db.commit() + return cls._success(filling_id) + except IntegrityError: + await query_db.rollback() + return await cls._recover_integrity_error( + query_db, + page_object, + request_hash, + current_user, + ) + except Exception: + if not recovery_transaction_ended: + await query_db.rollback() + raise diff --git a/shierfeng-fastapi-backend/tests/module_business/test_filling_service.py b/shierfeng-fastapi-backend/tests/module_business/test_filling_service.py new file mode 100644 index 0000000..ef4a0a0 --- /dev/null +++ b/shierfeng-fastapi-backend/tests/module_business/test_filling_service.py @@ -0,0 +1,919 @@ +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 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 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 filling_controller as filling_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.entity.do.master_do import BizPallet, BizStrain +from module_business.entity.do.operation_do import BizFillingRecord, BizPalletEvent, BizPalletInventory +from module_business.entity.vo.operation_vo import ( + BusinessOptionModel, + FillingCreateModel, + FillingModel, + FillingPageQueryModel, + PalletScanResponseModel, +) +from module_business.service.access_service import AccessService +from module_business.service.filling_service import FillingService + + +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_page( + *, + qr_token: str = 'qr-2', + warehouse_id: int = 7, + strain_id: int = 11, + batch_no: str = 'B-20260726', + package_spec: str = '20kg', + initial_packages: int = 18, + idempotency_key: str = 'fill-key', +) -> FillingCreateModel: + return FillingCreateModel( + qrToken=qr_token, + warehouseId=warehouse_id, + strainId=strain_id, + batchNo=batch_no, + packageSpec=package_spec, + initialPackages=initial_packages, + idempotencyKey=idempotency_key, + ) + + +def make_pallet( + pallet_id: int = 2, + *, + qr_token: str = 'qr-2', + capacity_packages: int = 20, + asset_status: str = '0', +) -> BizPallet: + return BizPallet( + pallet_id=pallet_id, + pallet_code=f'P-{pallet_id:03d}', + qr_token=qr_token, + capacity_packages=capacity_packages, + asset_status=asset_status, + ) + + +def make_inventory( + pallet_id: int = 2, + *, + flow_state: str = PalletFlowState.EMPTY_IN_WAREHOUSE.value, + warehouse_id: int | None = 7, + site_id: int | None = None, + current_line_id: int | None = None, + active_filling_id: int | None = None, + remaining_packages: int = 0, + last_event_id: int | None = 40, + 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, + last_event_id=last_event_id, + version=version, + updated_at=datetime(2026, 7, 25, 9, 0), + ) + + +def make_strain( + strain_id: int = 11, + *, + status: str = '0', + code: str = 'S-011', + name: str = '香菇 11', +) -> BizStrain: + return BizStrain( + strain_id=strain_id, + strain_code=code, + strain_name=name, + status=status, + ) + + +def make_filling( + filling_id: int = 31, + *, + warehouse_id: int = 7, + idempotency_key: str = 'fill-key', + request_hash: str | None = None, +) -> BizFillingRecord: + page = make_page(warehouse_id=warehouse_id, idempotency_key=idempotency_key) + return BizFillingRecord( + filling_id=filling_id, + filling_no=f'FL-{filling_id}', + pallet_id=2, + warehouse_id=warehouse_id, + strain_id=11, + strain_code_snapshot='S-011', + strain_name_snapshot='香菇 11', + batch_no=page.batch_no, + package_spec=page.package_spec, + initial_packages=page.initial_packages, + active_slot='ACTIVE', + idempotency_key=idempotency_key, + request_hash=request_hash or FillingService._build_request_hash(page), + operator_id=7, + operated_at=datetime(2026, 7, 26, 10, 30), + create_by='user-7', + create_time=datetime(2026, 7, 26, 10, 30), + update_by='user-7', + update_time=datetime(2026, 7, 26, 10, 30), + ) + + +class TestFillingPublicContract(unittest.TestCase): + def test_service_and_controller_are_available(self) -> None: + self.assertIsNotNone(FillingService) + self.assertIsNotNone(filling_controller_module.filling_controller) + + def test_fill_signature_and_types_are_exact(self) -> None: + signature = inspect.signature(FillingService.fill_pallet_services) + self.assertEqual(('query_db', 'page_object', 'current_user'), tuple(signature.parameters)) + hints = get_type_hints(FillingService.fill_pallet_services) + self.assertIs(AsyncSession, hints['query_db']) + self.assertIs(FillingCreateModel, hints['page_object']) + self.assertIs(CurrentUserModel, hints['current_user']) + self.assertIs(CrudResponseModel, hints['return']) + + def test_request_hash_is_stable_normalized_sha256_and_excludes_key(self) -> None: + page = make_page(batch_no=' B-20260726 ', package_spec=' 20kg ', idempotency_key='key-a') + expected_payload = json.dumps( + { + 'batch_no': 'B-20260726', + 'initial_packages': 18, + 'package_spec': '20kg', + 'qr_token': 'qr-2', + 'strain_id': 11, + 'warehouse_id': 7, + }, + sort_keys=True, + separators=(',', ':'), + ).encode() + expected = hashlib.sha256(expected_payload).hexdigest() + + self.assertEqual(expected, FillingService._build_request_hash(page)) + self.assertEqual( + expected, FillingService._build_request_hash(page.model_copy(update={'idempotency_key': 'key-b'})) + ) + self.assertNotEqual( + expected, FillingService._build_request_hash(page.model_copy(update={'initial_packages': 17})) + ) + + def test_create_model_requires_strict_positive_integer_packages(self) -> None: + for invalid in (True, 0, -1, 1.5): + with self.subTest(invalid=invalid), self.assertRaises(ValidationError): + make_page(initial_packages=invalid) + + +class TestFillingReadServices(unittest.IsolatedAsyncioTestCase): + def setUp(self) -> None: + self.db = MagicMock(spec=AsyncSession) + self.user = make_user() + + async def test_list_pushes_user_warehouse_scope_and_converts_rows(self) -> None: + query = FillingPageQueryModel(pageNum=1, pageSize=10) + row = make_filling() + page = PageModel(rows=[row], pageNum=1, pageSize=10, total=1, hasNext=False) + get_list = AsyncMock(return_value=page) + with ( + patch.object(AccessDao, 'list_user_warehouse_ids', new=AsyncMock(return_value=[9, 7, 7])), + patch.object(OperationDao, 'get_filling_list', new=get_list), + ): + result = await FillingService.get_filling_list_services(self.db, query, self.user) + + get_list.assert_awaited_once_with(self.db, query, True, warehouse_ids=[7, 9]) + self.assertIsInstance(result.rows[0], FillingModel) + self.assertEqual(1, result.total) + + async def test_list_with_empty_scope_returns_empty_page_without_global_rows(self) -> None: + query = FillingPageQueryModel() + empty_page = PageModel(rows=[], pageNum=1, pageSize=10, 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_filling_list', new=get_list), + ): + result = await FillingService.get_filling_list_services(self.db, query, self.user) + + get_list.assert_awaited_once_with(self.db, query, True, warehouse_ids=[]) + self.assertEqual([], result.rows) + self.assertEqual(0, result.total) + + async def test_list_rejects_explicit_warehouse_outside_scope(self) -> None: + query = FillingPageQueryModel(warehouseId=8) + get_list = AsyncMock() + with ( + patch.object(AccessDao, 'list_user_warehouse_ids', new=AsyncMock(return_value=[7])), + patch.object(OperationDao, 'get_filling_list', new=get_list), + self.assertRaises(PermissionException), + ): + await FillingService.get_filling_list_services(self.db, query, self.user) + get_list.assert_not_awaited() + + async def test_admin_list_is_unscoped(self) -> None: + query = FillingPageQueryModel() + get_list = AsyncMock(return_value=[]) + scope = AsyncMock() + with ( + patch.object(AccessDao, 'list_user_warehouse_ids', new=scope), + patch.object(OperationDao, 'get_filling_list', new=get_list), + ): + result = await FillingService.get_filling_list_services(self.db, query, make_user(1), is_page=False) + scope.assert_not_awaited() + get_list.assert_awaited_once_with(self.db, query, False, warehouse_ids=None) + self.assertEqual([], result) + + async def test_detail_is_scoped_before_lookup_and_hides_missing_for_normal_user(self) -> None: + get_detail = AsyncMock(return_value=None) + with ( + patch.object(AccessDao, 'list_user_warehouse_ids', new=AsyncMock(return_value=[7])), + patch.object(OperationDao, 'get_filling_by_id', new=get_detail), + self.assertRaises(PermissionException) as raised, + ): + await FillingService.filling_detail_services(self.db, 31, self.user) + get_detail.assert_awaited_once_with(self.db, 31, warehouse_ids=[7]) + self.assertEqual('无权访问该装填记录', raised.exception.message) + + async def test_admin_detail_keeps_precise_not_found_error(self) -> None: + get_detail = AsyncMock(return_value=None) + with ( + patch.object(OperationDao, 'get_filling_by_id', new=get_detail), + self.assertRaises(ServiceException) as raised, + ): + await FillingService.filling_detail_services(self.db, 31, make_user(1)) + get_detail.assert_awaited_once_with(self.db, 31, warehouse_ids=None) + self.assertEqual('装填记录不存在', raised.exception.message) + + async def test_strain_options_return_only_enabled_business_options(self) -> None: + strains = [ + {'strainId': 3, 'strainCode': 'S-3', 'strainName': '菌种三', 'status': '0'}, + {'strainId': 2, 'strainCode': 'S-2', 'strainName': '菌种二', 'status': '1'}, + {'strainId': 1, 'strainCode': 'S-1', 'strainName': '菌种一', 'status': '0'}, + ] + list_strain = AsyncMock(return_value=strains) + with patch.object(MasterDao, 'list_strain', new=list_strain): + result = await FillingService.get_strain_options_services(self.db, self.user) + + query = list_strain.await_args.args[1] + self.assertEqual('0', query.status) + self.assertFalse(list_strain.await_args.args[2]) + self.assertEqual([1, 3], [option.id for option in result]) + self.assertTrue(all(isinstance(option, BusinessOptionModel) for option in result)) + + async def test_scan_derives_scope_from_inventory_and_is_read_only(self) -> None: + pallet = make_pallet() + inventory = make_inventory() + access = AsyncMock() + pallet_lock = AsyncMock() + inventory_lock = AsyncMock() + with ( + patch.object(InventoryDao, 'get_pallet_by_qr_token', new=AsyncMock(return_value=pallet)), + patch.object(MasterDao, 'get_pallet_inventory', new=AsyncMock(return_value=inventory)), + patch.object(AccessService, 'assert_warehouse_access', new=access), + patch.object(InventoryDao, 'get_inventory_for_update', new=inventory_lock), + patch.object(InventoryDao, 'get_pallets_by_ids', new=pallet_lock), + ): + result = await FillingService.scan_filling_pallet_services(self.db, 'qr-2', self.user) + + access.assert_awaited_once_with(self.db, self.user, 7) + inventory_lock.assert_not_awaited() + pallet_lock.assert_not_awaited() + self.db.commit.assert_not_awaited() + self.assertEqual(PalletFlowState.EMPTY_IN_WAREHOUSE, result.flow_state) + self.assertEqual(['filling'], result.allowed_actions) + self.assertEqual(20, result.capacity_packages) + self.assertEqual(0, result.remaining_packages) + + async def test_scan_checks_access_before_exposing_invalid_state(self) -> None: + access = AsyncMock(side_effect=PermissionException(data='', message='无权访问该仓库')) + with ( + patch.object(InventoryDao, 'get_pallet_by_qr_token', new=AsyncMock(return_value=make_pallet())), + patch.object( + MasterDao, + 'get_pallet_inventory', + new=AsyncMock(return_value=make_inventory(flow_state=PalletFlowState.FILLED_IN_WAREHOUSE.value)), + ), + patch.object(AccessService, 'assert_warehouse_access', new=access), + self.assertRaises(PermissionException), + ): + await FillingService.scan_filling_pallet_services(self.db, 'qr-2', self.user) + access.assert_awaited_once_with(self.db, self.user, 7) + + async def test_scan_hides_nonexistent_pallet_from_normal_user_but_admin_gets_precise_error(self) -> None: + for user, exception_type in ((self.user, PermissionException), (make_user(1), ServiceException)): + with ( + self.subTest(admin=user.user.admin), + patch.object(InventoryDao, 'get_pallet_by_qr_token', new=AsyncMock(return_value=None)), + self.assertRaises(exception_type), + ): + await FillingService.scan_filling_pallet_services(self.db, 'missing', user) + + async def test_scan_admin_distinguishes_missing_inventory(self) -> None: + with ( + patch.object(InventoryDao, 'get_pallet_by_qr_token', new=AsyncMock(return_value=make_pallet())), + patch.object(MasterDao, 'get_pallet_inventory', new=AsyncMock(return_value=None)), + self.assertRaises(ServiceException) as raised, + ): + await FillingService.scan_filling_pallet_services(self.db, 'qr-2', make_user(1)) + self.assertEqual('托盘库存不存在', raised.exception.message) + + +class TestFillPalletService(unittest.IsolatedAsyncioTestCase): + def setUp(self) -> None: + self.db = MagicMock(spec=AsyncSession) + self.user = make_user() + self.page = make_page() + self.pallet = make_pallet() + self.inventory = make_inventory() + self.strain = make_strain() + self.warehouse = SimpleNamespace(warehouse_id=7, status='0') + self.fixed_time = datetime(2026, 7, 26, 10, 30) + + async def test_fill_locks_inventory_then_pallet_and_writes_one_atomic_event( # noqa: PLR0915 + self, + ) -> None: + actions: list[str] = [] + captured: dict[str, Any] = {} + + async def lock_inventory(*_args: Any) -> BizPalletInventory: + actions.append('inventory-lock') + return self.inventory + + async def lock_pallets(*_args: Any) -> list[BizPallet]: + actions.append('pallet-lock') + return [self.pallet] + + async def get_strain(*_args: Any) -> BizStrain: + actions.append('strain-read') + return self.strain + + async def add_filling(_db: AsyncSession, filling: BizFillingRecord) -> BizFillingRecord: + actions.append('filling-flush') + filling.filling_id = 31 + captured['filling'] = filling + return filling + + async def add_event( + _db: AsyncSession, + inventory: BizPalletInventory, + event: BizPalletEvent, + ) -> BizPalletEvent: + actions.append('event-flush') + event.event_id = 91 + event.event_seq = 6 + inventory.last_event_id = 91 + captured['event'] = event + return event + + async def commit() -> None: + actions.append('commit') + captured['filling'].__dict__.pop('filling_id', None) + + self.db.commit.side_effect = commit + with ( + patch.object(AccessDao, 'list_user_warehouse_ids', new=AsyncMock(return_value=[7])), + patch.object(AccessService, 'assert_warehouse_access', new=AsyncMock()), + patch.object(OperationDao, 'get_filling_by_idempotency_key', new=AsyncMock(return_value=None)), + patch.object(InventoryDao, 'get_pallet_by_qr_token', new=AsyncMock(return_value=self.pallet)), + patch.object(InventoryDao, 'get_inventory_for_update', new=AsyncMock(side_effect=lock_inventory)), + patch.object(InventoryDao, 'get_pallets_by_ids', new=AsyncMock(side_effect=lock_pallets)), + patch.object(MasterDao, 'get_warehouse', new=AsyncMock(return_value=self.warehouse)), + patch.object(MasterDao, 'get_strain', new=AsyncMock(side_effect=get_strain)), + patch.object(OperationDao, 'add_filling_record', new=AsyncMock(side_effect=add_filling)), + patch.object(InventoryDao, 'add_event', new=AsyncMock(side_effect=add_event)), + patch('module_business.service.filling_service.datetime') as mocked_datetime, + ): + mocked_datetime.now.return_value = self.fixed_time + result = await FillingService.fill_pallet_services(self.db, self.page, self.user) + + self.assertEqual( + ['inventory-lock', 'pallet-lock', 'strain-read', 'filling-flush', 'event-flush', 'commit'], + actions, + ) + filling = captured['filling'] + self.assertTrue(filling.filling_no.startswith('FL-')) + self.assertEqual(2, filling.pallet_id) + self.assertEqual(7, filling.warehouse_id) + self.assertEqual(11, filling.strain_id) + self.assertEqual('S-011', filling.strain_code_snapshot) + self.assertEqual('香菇 11', filling.strain_name_snapshot) + self.assertEqual('B-20260726', filling.batch_no) + self.assertEqual('20kg', filling.package_spec) + self.assertEqual(18, filling.initial_packages) + self.assertEqual('ACTIVE', filling.active_slot) + self.assertEqual('fill-key', filling.idempotency_key) + self.assertEqual(FillingService._build_request_hash(self.page), filling.request_hash) + self.assertEqual(7, filling.operator_id) + self.assertEqual(self.fixed_time, filling.operated_at) + self.assertEqual('user-7', filling.create_by) + self.assertEqual(self.fixed_time, filling.create_time) + self.assertEqual('user-7', filling.update_by) + self.assertEqual(self.fixed_time, filling.update_time) + + self.assertEqual(31, self.inventory.active_filling_id) + self.assertEqual(18, self.inventory.remaining_packages) + self.assertEqual(PalletFlowState.FILLED_IN_WAREHOUSE.value, self.inventory.flow_state) + self.assertEqual(4, self.inventory.version) + self.assertEqual(self.fixed_time, self.inventory.updated_at) + self.assertEqual(91, self.inventory.last_event_id) + + event = captured['event'] + self.assertEqual(PalletEventType.FILLING_CONFIRMED.value, event.event_type) + self.assertEqual('FILLING', event.source_type) + self.assertEqual(31, event.source_id) + self.assertEqual(PalletFlowState.EMPTY_IN_WAREHOUSE.value, event.before_state) + self.assertEqual(PalletFlowState.FILLED_IN_WAREHOUSE.value, event.after_state) + self.assertEqual(7, event.before_warehouse_id) + self.assertEqual(7, event.after_warehouse_id) + self.assertIsNone(event.before_site_id) + self.assertIsNone(event.after_site_id) + self.assertIsNone(event.before_line_id) + self.assertIsNone(event.after_line_id) + self.assertEqual(0, event.before_remaining_packages) + self.assertEqual(18, event.after_remaining_packages) + self.assertEqual(7, event.operator_id) + self.assertEqual(self.fixed_time, event.occurred_at) + self.assertEqual('fill-key', event.idempotency_key) + self.assertEqual(filling.request_hash, event.request_hash) + self.db.commit.assert_awaited_once_with() + self.db.rollback.assert_not_awaited() + self.assertTrue(result.is_success) + self.assertEqual('装填成功', result.message) + self.assertEqual(31, result.result) + + async def _assert_fill_rejected( + self, + *, + page: FillingCreateModel | None = None, + pallet: BizPallet | None = None, + inventory: BizPalletInventory | None = None, + strain: BizStrain | None = None, + ) -> None: + add_filling = AsyncMock() + add_event = AsyncMock() + with ( + patch.object(AccessDao, 'list_user_warehouse_ids', new=AsyncMock(return_value=[7])), + patch.object(AccessService, 'assert_warehouse_access', new=AsyncMock()), + patch.object(OperationDao, 'get_filling_by_idempotency_key', new=AsyncMock(return_value=None)), + patch.object(InventoryDao, 'get_pallet_by_qr_token', new=AsyncMock(return_value=pallet or self.pallet)), + patch.object( + InventoryDao, 'get_inventory_for_update', new=AsyncMock(return_value=inventory or self.inventory) + ), + patch.object(InventoryDao, 'get_pallets_by_ids', new=AsyncMock(return_value=[pallet or self.pallet])), + patch.object(OperationDao, 'get_active_filling_for_update', new=AsyncMock(return_value=None)), + patch.object(MasterDao, 'get_warehouse', new=AsyncMock(return_value=self.warehouse)), + patch.object(MasterDao, 'get_strain', new=AsyncMock(return_value=strain or self.strain)), + patch.object(OperationDao, 'add_filling_record', new=add_filling), + patch.object(InventoryDao, 'add_event', new=add_event), + self.assertRaises(ServiceException), + ): + await FillingService.fill_pallet_services(self.db, page or self.page, self.user) + add_filling.assert_not_awaited() + add_event.assert_not_awaited() + self.db.commit.assert_not_awaited() + self.db.rollback.assert_awaited_once() + + async def test_fill_rejects_packages_over_persisted_pallet_capacity(self) -> None: + await self._assert_fill_rejected(page=make_page(initial_packages=21)) + + async def test_fill_rejects_stopped_pallet_and_stopped_or_missing_strain(self) -> None: + await self._assert_fill_rejected(pallet=make_pallet(asset_status='1')) + self.db.reset_mock() + await self._assert_fill_rejected(strain=make_strain(status='1')) + self.db.reset_mock() + add_filling = AsyncMock() + with ( + patch.object(AccessDao, 'list_user_warehouse_ids', new=AsyncMock(return_value=[7])), + patch.object(AccessService, 'assert_warehouse_access', new=AsyncMock()), + patch.object(OperationDao, 'get_filling_by_idempotency_key', new=AsyncMock(return_value=None)), + patch.object(InventoryDao, 'get_pallet_by_qr_token', new=AsyncMock(return_value=self.pallet)), + patch.object(InventoryDao, 'get_inventory_for_update', new=AsyncMock(return_value=self.inventory)), + patch.object(InventoryDao, 'get_pallets_by_ids', new=AsyncMock(return_value=[self.pallet])), + patch.object(MasterDao, 'get_warehouse', new=AsyncMock(return_value=self.warehouse)), + patch.object(MasterDao, 'get_strain', new=AsyncMock(return_value=None)), + patch.object(OperationDao, 'add_filling_record', new=add_filling), + self.assertRaises(ServiceException), + ): + await FillingService.fill_pallet_services(self.db, self.page, self.user) + add_filling.assert_not_awaited() + self.db.rollback.assert_awaited_once() + + async def test_fill_rechecks_persisted_qr_and_all_empty_inventory_invariants_after_locks(self) -> None: + invalid_inventories = ( + make_inventory(flow_state=PalletFlowState.FILLED_IN_WAREHOUSE.value), + make_inventory(warehouse_id=8), + make_inventory(site_id=3), + make_inventory(current_line_id=4), + make_inventory(active_filling_id=30), + make_inventory(remaining_packages=1), + ) + for inventory in invalid_inventories: + with self.subTest(values=vars(inventory)): + self.db.reset_mock() + await self._assert_fill_rejected(inventory=inventory) + self.db.reset_mock() + await self._assert_fill_rejected(pallet=make_pallet(qr_token='persisted-other')) + + async def test_same_idempotency_key_and_hash_returns_existing_result_before_locks(self) -> None: + existing = make_filling() + inventory_lock = AsyncMock() + access = AsyncMock() + + async def rollback() -> None: + existing.__dict__.pop('filling_id', None) + + self.db.rollback.side_effect = rollback + with ( + patch.object(AccessDao, 'list_user_warehouse_ids', new=AsyncMock(return_value=[7])), + patch.object(AccessService, 'assert_warehouse_access', new=access), + patch.object( + OperationDao, 'get_filling_by_idempotency_key', new=AsyncMock(return_value=existing) + ) as get_key, + patch.object(InventoryDao, 'get_inventory_for_update', new=inventory_lock), + ): + result = await FillingService.fill_pallet_services(self.db, self.page, self.user) + + get_key.assert_awaited_once_with(self.db, 'fill-key', warehouse_ids=[7]) + access.assert_awaited_once_with(self.db, self.user, 7) + inventory_lock.assert_not_awaited() + self.db.commit.assert_not_awaited() + self.db.rollback.assert_awaited_once() + self.assertTrue(result.is_success) + self.assertEqual(31, result.result) + + async def test_same_idempotency_key_with_different_hash_is_rejected(self) -> None: + existing = make_filling(request_hash='different-hash') + with ( + patch.object(AccessDao, 'list_user_warehouse_ids', new=AsyncMock(return_value=[7])), + patch.object(AccessService, 'assert_warehouse_access', new=AsyncMock()), + patch.object(OperationDao, 'get_filling_by_idempotency_key', new=AsyncMock(return_value=existing)), + self.assertRaises(ServiceException) as raised, + ): + await FillingService.fill_pallet_services(self.db, self.page, self.user) + self.assertIn('幂等键', raised.exception.message) + self.db.rollback.assert_awaited_once() + + async def test_idempotency_retry_cannot_read_outside_current_scope(self) -> None: + get_key = AsyncMock() + with ( + patch.object(AccessDao, 'list_user_warehouse_ids', new=AsyncMock(return_value=[])), + patch.object(OperationDao, 'get_filling_by_idempotency_key', new=get_key), + self.assertRaises(PermissionException), + ): + await FillingService.fill_pallet_services(self.db, self.page, self.user) + get_key.assert_not_awaited() + self.db.rollback.assert_awaited_once() + + async def test_fill_hides_qr_whose_locked_inventory_is_outside_actual_scope(self) -> None: + foreign_warehouse_id = 8 + inventory = make_inventory( + warehouse_id=foreign_warehouse_id, + flow_state=PalletFlowState.FILLED_IN_WAREHOUSE.value, + ) + pallet_lock = AsyncMock(return_value=[make_pallet(asset_status='1')]) + + async def assert_access(_db: AsyncSession, _user: CurrentUserModel, warehouse_id: int) -> None: + if warehouse_id == foreign_warehouse_id: + raise PermissionException(data='', message='无权访问该仓库') + + access = AsyncMock(side_effect=assert_access) + with ( + patch.object(AccessDao, 'list_user_warehouse_ids', new=AsyncMock(return_value=[7])), + patch.object(AccessService, 'assert_warehouse_access', new=access), + patch.object(OperationDao, 'get_filling_by_idempotency_key', new=AsyncMock(return_value=None)), + patch.object(InventoryDao, 'get_pallet_by_qr_token', new=AsyncMock(return_value=self.pallet)), + patch.object(InventoryDao, 'get_inventory_for_update', new=AsyncMock(return_value=inventory)), + patch.object(InventoryDao, 'get_pallets_by_ids', new=pallet_lock), + patch.object(MasterDao, 'get_warehouse', new=AsyncMock(return_value=self.warehouse)), + self.assertRaises(PermissionException), + ): + await FillingService.fill_pallet_services(self.db, self.page, self.user) + + self.assertEqual([7, foreign_warehouse_id], [call.args[2] for call in access.await_args_list]) + pallet_lock.assert_not_awaited() + self.db.rollback.assert_awaited_once() + + async def test_concurrent_same_key_recovers_after_waiting_for_inventory_lock(self) -> None: + recovered = make_filling() + inventory = make_inventory( + flow_state=PalletFlowState.FILLED_IN_WAREHOUSE.value, + active_filling_id=31, + remaining_packages=18, + version=4, + ) + actions: list[str] = [] + + async def lock_inventory(*_args: Any) -> BizPalletInventory: + actions.append('inventory-lock') + return inventory + + async def lock_pallet(*_args: Any) -> list[BizPallet]: + actions.append('pallet-lock') + return [self.pallet] + + async def lock_active(*_args: Any) -> BizFillingRecord: + actions.append('filling-lock') + return recovered + + add_filling = AsyncMock() + add_event = AsyncMock() + with ( + patch.object(AccessDao, 'list_user_warehouse_ids', new=AsyncMock(return_value=[7])), + patch.object(AccessService, 'assert_warehouse_access', new=AsyncMock()), + patch.object( + OperationDao, + 'get_filling_by_idempotency_key', + new=AsyncMock(side_effect=[None, recovered]), + ) as get_key, + patch.object(InventoryDao, 'get_pallet_by_qr_token', new=AsyncMock(return_value=self.pallet)), + patch.object(InventoryDao, 'get_inventory_for_update', new=AsyncMock(side_effect=lock_inventory)), + patch.object(InventoryDao, 'get_pallets_by_ids', new=AsyncMock(side_effect=lock_pallet)), + patch.object(OperationDao, 'get_active_filling_for_update', new=AsyncMock(side_effect=lock_active)), + patch.object(MasterDao, 'get_warehouse', new=AsyncMock(return_value=self.warehouse)), + patch.object(OperationDao, 'add_filling_record', new=add_filling), + patch.object(InventoryDao, 'add_event', new=add_event), + ): + result = await FillingService.fill_pallet_services(self.db, self.page, self.user) + + self.assertEqual(['inventory-lock', 'pallet-lock', 'filling-lock'], actions) + self.assertEqual(2, get_key.await_count) + self.assertEqual(2, self.db.rollback.await_count) + self.assertEqual(31, result.result) + add_filling.assert_not_awaited() + add_event.assert_not_awaited() + self.db.commit.assert_not_awaited() + + async def test_concurrent_same_key_with_different_hash_is_stably_rejected(self) -> None: + recovered = make_filling(request_hash='different-hash') + inventory = make_inventory( + flow_state=PalletFlowState.FILLED_IN_WAREHOUSE.value, + active_filling_id=31, + remaining_packages=18, + version=4, + ) + add_filling = AsyncMock() + with ( + patch.object(AccessDao, 'list_user_warehouse_ids', new=AsyncMock(return_value=[7])), + patch.object(AccessService, 'assert_warehouse_access', new=AsyncMock()), + patch.object( + OperationDao, + 'get_filling_by_idempotency_key', + new=AsyncMock(side_effect=[None, recovered]), + ) as get_key, + patch.object(InventoryDao, 'get_pallet_by_qr_token', new=AsyncMock(return_value=self.pallet)), + patch.object(InventoryDao, 'get_inventory_for_update', new=AsyncMock(return_value=inventory)), + patch.object(InventoryDao, 'get_pallets_by_ids', new=AsyncMock(return_value=[self.pallet])), + patch.object(OperationDao, 'get_active_filling_for_update', new=AsyncMock(return_value=recovered)), + patch.object(MasterDao, 'get_warehouse', new=AsyncMock(return_value=self.warehouse)), + patch.object(OperationDao, 'add_filling_record', new=add_filling), + self.assertRaises(ServiceException) as raised, + ): + await FillingService.fill_pallet_services(self.db, self.page, self.user) + + self.assertEqual('装填失败，幂等键已被其他请求使用', raised.exception.message) + self.assertEqual(2, get_key.await_count) + self.assertEqual(2, self.db.rollback.await_count) + add_filling.assert_not_awaited() + self.db.commit.assert_not_awaited() + + async def test_any_write_failure_rolls_back_without_commit(self) -> None: + with ( + patch.object(AccessDao, 'list_user_warehouse_ids', new=AsyncMock(return_value=[7])), + patch.object(AccessService, 'assert_warehouse_access', new=AsyncMock()), + patch.object(OperationDao, 'get_filling_by_idempotency_key', new=AsyncMock(return_value=None)), + patch.object(InventoryDao, 'get_pallet_by_qr_token', new=AsyncMock(return_value=self.pallet)), + patch.object(InventoryDao, 'get_inventory_for_update', new=AsyncMock(return_value=self.inventory)), + patch.object(InventoryDao, 'get_pallets_by_ids', new=AsyncMock(return_value=[self.pallet])), + patch.object(MasterDao, 'get_warehouse', new=AsyncMock(return_value=self.warehouse)), + patch.object(MasterDao, 'get_strain', new=AsyncMock(return_value=self.strain)), + patch.object(OperationDao, 'add_filling_record', new=AsyncMock(side_effect=RuntimeError('flush failed'))), + self.assertRaises(RuntimeError), + ): + await FillingService.fill_pallet_services(self.db, self.page, self.user) + self.db.rollback.assert_awaited_once() + self.db.commit.assert_not_awaited() + + async def test_request_hash_failure_also_rolls_back(self) -> None: + with ( + patch.object(FillingService, '_build_request_hash', side_effect=RuntimeError('hash failed')), + self.assertRaises(RuntimeError), + ): + await FillingService.fill_pallet_services(self.db, self.page, self.user) + self.db.rollback.assert_awaited_once() + self.db.commit.assert_not_awaited() + + async def test_integrity_error_recovers_only_exact_authorized_idempotent_result(self) -> None: + recovered = make_filling() + key_reads = AsyncMock(side_effect=[None, recovered]) + access = AsyncMock() + with ( + patch.object(AccessDao, 'list_user_warehouse_ids', new=AsyncMock(return_value=[7])), + patch.object(AccessService, 'assert_warehouse_access', new=access), + patch.object(OperationDao, 'get_filling_by_idempotency_key', new=key_reads), + patch.object(InventoryDao, 'get_pallet_by_qr_token', new=AsyncMock(return_value=self.pallet)), + patch.object(InventoryDao, 'get_inventory_for_update', new=AsyncMock(return_value=self.inventory)), + patch.object(InventoryDao, 'get_pallets_by_ids', new=AsyncMock(return_value=[self.pallet])), + patch.object(MasterDao, 'get_warehouse', new=AsyncMock(return_value=self.warehouse)), + patch.object(MasterDao, 'get_strain', new=AsyncMock(return_value=self.strain)), + patch.object(OperationDao, 'add_filling_record', new=AsyncMock(side_effect=integrity_error())), + ): + result = await FillingService.fill_pallet_services(self.db, self.page, self.user) + + self.assertEqual(2, key_reads.await_count) + self.assertEqual(2, self.db.rollback.await_count) + access.assert_any_await(self.db, self.user, 7) + self.assertTrue(result.is_success) + + async def test_integrity_error_without_scoped_idempotent_result_is_hidden_and_never_succeeds(self) -> None: + with ( + patch.object(AccessDao, 'list_user_warehouse_ids', new=AsyncMock(return_value=[7])), + patch.object(AccessService, 'assert_warehouse_access', new=AsyncMock()), + patch.object( + OperationDao, + 'get_filling_by_idempotency_key', + new=AsyncMock(side_effect=[None, None]), + ), + patch.object(InventoryDao, 'get_pallet_by_qr_token', new=AsyncMock(return_value=self.pallet)), + patch.object(InventoryDao, 'get_inventory_for_update', new=AsyncMock(return_value=self.inventory)), + patch.object(InventoryDao, 'get_pallets_by_ids', new=AsyncMock(return_value=[self.pallet])), + patch.object(MasterDao, 'get_warehouse', new=AsyncMock(return_value=self.warehouse)), + patch.object(MasterDao, 'get_strain', new=AsyncMock(return_value=self.strain)), + patch.object(OperationDao, 'add_filling_record', new=AsyncMock(side_effect=integrity_error())), + self.assertRaises(PermissionException), + ): + await FillingService.fill_pallet_services(self.db, self.page, self.user) + self.assertEqual(2, self.db.rollback.await_count) + self.db.commit.assert_not_awaited() + + +class TestFillingOperationDaoScope(unittest.IsolatedAsyncioTestCase): + @staticmethod + def _compiled(statement: Any) -> str: + return str(statement.compile(compile_kwargs={'literal_binds': True})).lower() + + async def test_filling_list_pushes_warehouse_scope_into_sql(self) -> None: + query = FillingPageQueryModel() + paginate = AsyncMock(return_value=[]) + db = MagicMock(spec=AsyncSession) + with patch.object(OperationDao, '_paginate', new=paginate): + await OperationDao.get_filling_list(db, query, False, warehouse_ids=[9, 7, 7]) + sql = self._compiled(paginate.await_args.args[1]) + self.assertIn('biz_filling_record.warehouse_id in (7, 9)', sql) + + async def test_filling_list_empty_scope_compiles_to_always_false_sql(self) -> None: + query = FillingPageQueryModel() + paginate = AsyncMock(return_value=[]) + db = MagicMock(spec=AsyncSession) + with patch.object(OperationDao, '_paginate', new=paginate): + await OperationDao.get_filling_list(db, query, False, warehouse_ids=[]) + sql = self._compiled(paginate.await_args.args[1]) + self.assertTrue('false' in sql or '0 = 1' in sql, sql) + + async def test_filling_detail_and_idempotency_reads_push_scope_into_sql(self) -> None: + for method, key in ( + (OperationDao.get_filling_by_id, 31), + (OperationDao.get_filling_by_idempotency_key, 'fill-key'), + ): + with self.subTest(method=method.__name__): + result = MagicMock() + result.scalar_one_or_none.return_value = None + db = MagicMock(spec=AsyncSession) + db.execute.return_value = result + await method(db, key, warehouse_ids=[7]) + sql = self._compiled(db.execute.await_args.args[0]) + self.assertIn('biz_filling_record.warehouse_id in (7)', sql) + + +class TestFillingController(unittest.TestCase): + @staticmethod + def _routes() -> dict[tuple[str, str], APIRoute]: + return { + (route.path, next(iter(route.methods))): route + for route in filling_controller_module.filling_controller.routes + if isinstance(route, APIRoute) + } + + def test_openapi_has_exact_five_operations_and_response_models(self) -> None: + app = FastAPI() + app.include_router(filling_controller_module.filling_controller) + schema = app.openapi() + expected = { + ('/business/filling/list', 'get'), + ('/business/filling/strain-options', 'get'), + ('/business/filling/scan/{qr_token}', 'get'), + ('/business/filling/{filling_id}', 'get'), + ('/business/filling', 'post'), + } + self.assertEqual(expected, {(path, method) for path, methods in schema['paths'].items() for method in methods}) + scan_parameters = schema['paths']['/business/filling/scan/{qr_token}']['get']['parameters'] + self.assertEqual({'qr_token'}, {parameter['name'] for parameter in scan_parameters}) + + routes = self._routes() + self.assertEqual(PageResponseModel[FillingModel], routes[('/business/filling/list', 'GET')].response_model) + self.assertEqual( + DataResponseModel[list[BusinessOptionModel]], + routes[('/business/filling/strain-options', 'GET')].response_model, + ) + self.assertEqual( + DataResponseModel[PalletScanResponseModel], + routes[('/business/filling/scan/{qr_token}', 'GET')].response_model, + ) + self.assertEqual( + DataResponseModel[FillingModel], + routes[('/business/filling/{filling_id}', 'GET')].response_model, + ) + self.assertEqual(ResponseBaseModel, routes[('/business/filling', 'POST')].response_model) + + def test_static_routes_are_declared_before_dynamic_detail(self) -> None: + paths = [route.path for route in filling_controller_module.filling_controller.routes] + detail_index = paths.index('/business/filling/{filling_id}') + self.assertLess(paths.index('/business/filling/strain-options'), detail_index) + self.assertLess(paths.index('/business/filling/scan/{qr_token}'), detail_index) + + def test_each_route_has_exact_permission_and_three_unique_values(self) -> None: + expected = { + ('/business/filling/list', 'GET'): 'business:filling:list', + ('/business/filling/strain-options', 'GET'): 'business:filling:query', + ('/business/filling/scan/{qr_token}', 'GET'): 'business:filling:query', + ('/business/filling/{filling_id}', 'GET'): 'business:filling:query', + ('/business/filling', 'POST'): 'business:filling:add', + } + actual = {} + for key, route in self._routes().items(): + permissions = [ + dependency.dependency + for dependency in route.dependencies + if isinstance(dependency.dependency, CheckUserInterfaceAuth) + ] + self.assertEqual(1, len(permissions), key) + self.assertTrue(any(isinstance(dependency.dependency, PreAuth) for dependency in route.dependencies)) + actual[key] = permissions[0].perm + self.assertEqual(expected, actual) + self.assertEqual( + {'business:filling:list', 'business:filling:query', 'business:filling:add'}, + set(actual.values()), + ) + + def test_all_routes_inject_current_user_and_only_post_has_log(self) -> None: + for key, route in self._routes().items(): + self.assertTrue( + any( + dependency.name == 'current_user' and dependency.call == LoginService.get_current_user + for dependency in route.dependant.dependencies + ), + key, + ) + self.assertEqual(key == ('/business/filling', 'POST'), hasattr(route.endpoint, '__wrapped__')) + + def test_controller_does_not_add_scan_export_or_draft_permissions(self) -> None: + source = inspect.getsource(filling_controller_module) + for forbidden in ( + 'business:filling:scan', + 'business:filling:export', + 'business:filling:edit', + 'business:filling:remove', + 'business:filling:confirm', + ): + self.assertNotIn(forbidden, 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 ae5ae73..5f1aa1a 100644 --- a/shierfeng-fastapi-backend/tests/module_business/test_inventory_dao.py +++ b/shierfeng-fastapi-backend/tests/module_business/test_inventory_dao.py @@ -127,7 +127,7 @@ class TestDaoPublicContract(unittest.TestCase): 'get_active_filling_for_update': ('db', 'filling_id'), 'get_inbound_by_confirm_idempotency_key': ('db', 'idempotency_key'), 'get_shipment_by_confirm_idempotency_key': ('db', 'idempotency_key'), - 'get_filling_by_idempotency_key': ('db', 'idempotency_key'), + 'get_filling_by_idempotency_key': ('db', 'idempotency_key', 'warehouse_ids'), 'get_usage_by_idempotency_key': ('db', 'idempotency_key'), 'get_return_by_request_idempotency_key': ('db', 'idempotency_key'), 'get_return_by_confirm_idempotency_key': ('db', 'idempotency_key'), @@ -136,7 +136,7 @@ class TestDaoPublicContract(unittest.TestCase): 'update_inbound': ('db', 'inbound'), 'delete_inbound': ('db', 'inbound_id'), 'replace_inbound_items': ('db', 'inbound_id', 'items'), - 'get_filling_by_id': ('db', 'filling_id'), + 'get_filling_by_id': ('db', 'filling_id', 'warehouse_ids'), 'add_filling_record': ('db', 'filling'), 'update_filling_record': ('db', 'filling'), 'get_shipment_by_id': ('db', 'shipment_id'), @@ -164,7 +164,12 @@ class TestDaoPublicContract(unittest.TestCase): self.assertIs(True, inbound_list_signature.parameters['is_page'].default) self.assertIsNone(inbound_list_signature.parameters['warehouse_ids'].default) - for method_name in ('get_inbound_for_update', 'get_inbound_by_id'): + for method_name in ( + 'get_inbound_for_update', + 'get_inbound_by_id', + 'get_filling_by_idempotency_key', + 'get_filling_by_id', + ): with self.subTest(method=method_name): signature = inspect.signature(getattr(OperationDao, method_name)) warehouse_ids = signature.parameters.get('warehouse_ids') @@ -172,12 +177,15 @@ class TestDaoPublicContract(unittest.TestCase): if warehouse_ids is not None: self.assertIsNone(warehouse_ids.default) - for method_name in ( - 'get_filling_list', - 'get_shipment_list', - 'get_usage_list', - 'get_return_list', - ): + filling_list_signature = inspect.signature(OperationDao.get_filling_list) + self.assertEqual( + ('db', 'query_object', 'is_page', 'warehouse_ids'), + tuple(filling_list_signature.parameters), + ) + self.assertIs(True, filling_list_signature.parameters['is_page'].default) + self.assertIsNone(filling_list_signature.parameters['warehouse_ids'].default) + + for method_name in ('get_shipment_list', 'get_usage_list', 'get_return_list'): with self.subTest(method=method_name): signature = inspect.signature(getattr(OperationDao, method_name)) self.assertEqual(('db', 'query_object', 'is_page'), tuple(signature.parameters))