From bc774c74e6386c682b12bf9f0d3be2eb00ce2ea3 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 21:14:01 +0800 Subject: [PATCH] =?UTF-8?q?=E5=8A=9F=E8=83=BD=EF=BC=9A=E5=AE=9E=E7=8E=B0?= =?UTF-8?q?=E5=AE=A2=E6=88=B7=E7=8E=B0=E5=9C=BA=E4=BD=BF=E7=94=A8=E8=AE=B0?= =?UTF-8?q?=E5=BD=95?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../controller/usage_controller.py | 124 ++ .../module_business/dao/master_dao.py | 16 +- .../module_business/dao/operation_dao.py | 16 +- .../module_business/service/usage_service.py | 524 +++++++ .../module_business/test_inventory_dao.py | 89 +- .../module_business/test_master_service.py | 12 +- .../module_business/test_usage_service.py | 1201 +++++++++++++++++ 7 files changed, 1968 insertions(+), 14 deletions(-) create mode 100644 shierfeng-fastapi-backend/module_business/controller/usage_controller.py create mode 100644 shierfeng-fastapi-backend/module_business/service/usage_service.py create mode 100644 shierfeng-fastapi-backend/tests/module_business/test_usage_service.py diff --git a/shierfeng-fastapi-backend/module_business/controller/usage_controller.py b/shierfeng-fastapi-backend/module_business/controller/usage_controller.py new file mode 100644 index 0000000..a010bb9 --- /dev/null +++ b/shierfeng-fastapi-backend/module_business/controller/usage_controller.py @@ -0,0 +1,124 @@ +"""客户现场使用接口。""" + +from typing import Annotated + +from fastapi import Path, Query, Request, Response +from sqlalchemy.ext.asyncio import AsyncSession + +from common.annotation.log_annotation import Log +from common.aspect.db_seesion import DBSessionDependency +from common.aspect.interface_auth import UserInterfaceAuthDependency +from common.aspect.pre_auth import CurrentUserDependency, PreAuthDependency +from common.enums import BusinessType +from common.router import APIRouterPro +from common.vo import DataResponseModel, PageResponseModel, ResponseBaseModel +from module_admin.entity.vo.user_vo import CurrentUserModel +from module_business.entity.vo.operation_vo import ( + BusinessOptionModel, + PalletScanResponseModel, + UsageCreateModel, + UsageModel, + UsagePageQueryModel, +) +from module_business.service.usage_service import UsageService +from utils.response_util import ResponseUtil + +usage_controller = APIRouterPro( + prefix='/business/usage', + tags=['业务管理-客户现场使用'], + dependencies=[PreAuthDependency()], +) + + +@usage_controller.get( + '/list', + summary='获取客户现场使用记录分页列表接口', + description='按当前用户客户现场数据范围获取使用记录', + response_model=PageResponseModel[UsageModel], + dependencies=[UserInterfaceAuthDependency('business:usage:list')], +) +async def get_business_usage_list( + request: Request, + usage_page_query: Annotated[UsagePageQueryModel, Query()], + query_db: Annotated[AsyncSession, DBSessionDependency()], + current_user: Annotated[CurrentUserModel, CurrentUserDependency()], +) -> Response: + result = await UsageService.get_usage_list_services( + query_db, + usage_page_query, + current_user, + is_page=True, + ) + return ResponseUtil.success(model_content=result) + + +@usage_controller.get( + '/line-options', + summary='获取客户现场启用产线选项接口', + description='仅返回当前用户有权访问的指定客户现场启用产线', + response_model=DataResponseModel[list[BusinessOptionModel]], + dependencies=[UserInterfaceAuthDependency('business:usage:query')], +) +async def get_business_usage_line_options( + request: Request, + site_id: Annotated[int, Query(alias='siteId', gt=0, description='客户现场ID')], + query_db: Annotated[AsyncSession, DBSessionDependency()], + current_user: Annotated[CurrentUserModel, CurrentUserDependency()], +) -> Response: + result = await UsageService.get_line_options_services(query_db, site_id, current_user) + return ResponseUtil.success(data=result) + + +@usage_controller.get( + '/scan/{qr_token}', + summary='扫描客户现场待使用托盘接口', + description='按托盘当前客户现场数据范围判断是否允许记录使用', + response_model=DataResponseModel[PalletScanResponseModel], + dependencies=[UserInterfaceAuthDependency('business:usage:query')], +) +async def scan_business_usage_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 UsageService.scan_usage_pallet_services(query_db, qr_token, current_user) + return ResponseUtil.success(data=result) + + +@usage_controller.get( + '/{usage_id}', + summary='获取客户现场使用记录详情接口', + description='按客户现场数据范围获取指定使用记录', + response_model=DataResponseModel[UsageModel], + dependencies=[UserInterfaceAuthDependency('business:usage:query')], +) +async def query_detail_business_usage( + request: Request, + usage_id: Annotated[int, Path(gt=0, description='使用记录ID')], + query_db: Annotated[AsyncSession, DBSessionDependency()], + current_user: Annotated[CurrentUserModel, CurrentUserDependency()], +) -> Response: + result = await UsageService.usage_detail_services(query_db, usage_id, current_user) + return ResponseUtil.success(data=result) + + +@usage_controller.post( + '', + summary='记录客户现场使用接口', + description='扫描托盘并原子记录客户现场使用包数', + response_model=ResponseBaseModel, + dependencies=[UserInterfaceAuthDependency('business:usage:add')], +) +@Log(title='客户现场使用', business_type=BusinessType.INSERT) +async def add_business_usage( + request: Request, + add_usage: UsageCreateModel, + query_db: Annotated[AsyncSession, DBSessionDependency()], + current_user: Annotated[CurrentUserModel, CurrentUserDependency()], +) -> Response: + result = await UsageService.record_usage_services(query_db, add_usage, current_user) + return ResponseUtil.success(msg=result.message) diff --git a/shierfeng-fastapi-backend/module_business/dao/master_dao.py b/shierfeng-fastapi-backend/module_business/dao/master_dao.py index 6366d22..b39445c 100644 --- a/shierfeng-fastapi-backend/module_business/dao/master_dao.py +++ b/shierfeng-fastapi-backend/module_business/dao/master_dao.py @@ -337,8 +337,20 @@ class MasterDao: return await cls._paginate(db, statement, query_object, is_page) @classmethod - async def get_line(cls, db: AsyncSession, line_id: int) -> BizProductionLine | None: - return await cls._get(db, BizProductionLine, BizProductionLine.line_id, line_id) + async def get_line( + cls, + db: AsyncSession, + line_id: int, + *, + for_update: bool = False, + ) -> BizProductionLine | None: + return await cls._get( + db, + BizProductionLine, + BizProductionLine.line_id, + line_id, + for_update=for_update, + ) @classmethod async def find_line_by_code( diff --git a/shierfeng-fastapi-backend/module_business/dao/operation_dao.py b/shierfeng-fastapi-backend/module_business/dao/operation_dao.py index fb699ae..ef329b2 100644 --- a/shierfeng-fastapi-backend/module_business/dao/operation_dao.py +++ b/shierfeng-fastapi-backend/module_business/dao/operation_dao.py @@ -69,6 +69,15 @@ class OperationDao: scoped_warehouse_ids = sorted(set(warehouse_ids)) return [BizFillingRecord.warehouse_id.in_(scoped_warehouse_ids) if scoped_warehouse_ids else false()] + @staticmethod + def _usage_scope_conditions( + site_ids: list[int] | None, + ) -> list[ColumnElement[bool]]: + if site_ids is None: + return [] + scoped_site_ids = sorted(set(site_ids)) + return [BizUsageRecord.site_id.in_(scoped_site_ids) if scoped_site_ids else false()] + @staticmethod def _shipment_scope_conditions( warehouse_ids: list[int] | None, @@ -363,12 +372,14 @@ class OperationDao: cls, db: AsyncSession, idempotency_key: str, + site_ids: list[int] | None = None, ) -> BizUsageRecord | None: return await cls._get_by_key( db, BizUsageRecord, BizUsageRecord.idempotency_key, idempotency_key, + *cls._usage_scope_conditions(site_ids), ) @classmethod @@ -651,8 +662,9 @@ class OperationDao: db: AsyncSession, query_object: UsagePageQueryModel, is_page: bool = True, + site_ids: list[int] | None = None, ) -> PageModel | list[dict[str, Any]]: - conditions = [] + conditions = cls._usage_scope_conditions(site_ids) if query_object.usage_no: conditions.append(BizUsageRecord.usage_no.like(f'%{query_object.usage_no}%')) if query_object.site_id: @@ -682,12 +694,14 @@ class OperationDao: cls, db: AsyncSession, usage_id: int, + site_ids: list[int] | None = None, ) -> BizUsageRecord | None: return await cls._get_by_id( db, BizUsageRecord, BizUsageRecord.usage_id, usage_id, + *cls._usage_scope_conditions(site_ids), ) @classmethod diff --git a/shierfeng-fastapi-backend/module_business/service/usage_service.py b/shierfeng-fastapi-backend/module_business/service/usage_service.py new file mode 100644 index 0000000..e25d7bb --- /dev/null +++ b/shierfeng-fastapi-backend/module_business/service/usage_service.py @@ -0,0 +1,524 @@ +"""客户现场使用服务。""" + +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.operation_do import BizPalletEvent, BizUsageRecord +from module_business.entity.vo.master_vo import LinePageQueryModel +from module_business.entity.vo.operation_vo import ( + BusinessOptionModel, + PalletScanResponseModel, + UsageCreateModel, + UsageModel, + UsagePageQueryModel, +) +from module_business.service.access_service import AccessService + + +class UsageService: + """客户现场使用服务层。""" + + @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) + + @classmethod + async def _site_scope( + cls, + query_db: AsyncSession, + current_user: CurrentUserModel, + ) -> list[int] | None: + if cls._is_admin(current_user): + return None + user_id, _ = cls._current_identity(current_user) + return sorted(set(await AccessDao.list_user_site_ids(query_db, user_id))) + + @classmethod + def _raise_pallet_not_accessible( + cls, + current_user: CurrentUserModel, + admin_message: str = '托盘不存在', + ) -> NoReturn: + if cls._is_admin(current_user): + raise ServiceException(message=admin_message) + raise PermissionException(data='', message='无权访问该托盘') + + @classmethod + async def _assert_pallet_site_access( + cls, + query_db: AsyncSession, + current_user: CurrentUserModel, + site_id: int, + *, + for_update: bool = False, + ) -> None: + try: + if for_update: + await AccessService.assert_site_access( + query_db, + current_user, + site_id, + for_update=True, + ) + else: + await AccessService.assert_site_access(query_db, current_user, site_id) + except PermissionException: + if cls._is_admin(current_user): + raise + cls._raise_pallet_not_accessible(current_user) + + @classmethod + def _raise_usage_not_accessible(cls, current_user: CurrentUserModel) -> NoReturn: + if cls._is_admin(current_user): + raise ServiceException(message='使用记录不存在') + raise PermissionException(data='', message='无权访问该使用记录') + + @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: UsageCreateModel) -> str: + payload = json.dumps( + { + 'qr_token': page_object.qr_token.strip(), + 'line_id': page_object.line_id, + 'used_packages': page_object.used_packages, + }, + sort_keys=True, + separators=(',', ':'), + ).encode() + return hashlib.sha256(payload).hexdigest() + + @staticmethod + def _is_exact_idempotent_result( + usage: BizUsageRecord, + page_object: UsageCreateModel, + request_hash: str, + ) -> bool: + return usage.idempotency_key == page_object.idempotency_key and usage.request_hash == request_hash + + @staticmethod + def _success(usage_id: int) -> CrudResponseModel: + return CrudResponseModel(is_success=True, message='使用成功', result=usage_id) + + @classmethod + async def _lock_and_reauthorize_usage_site( + cls, + query_db: AsyncSession, + site_id: int, + current_user: CurrentUserModel, + ) -> None: + site = await MasterDao.get_site(query_db, site_id, for_update=True) + try: + await AccessService.assert_site_access( + query_db, + current_user, + site_id, + for_update=True, + ) + except PermissionException: + if cls._is_admin(current_user): + raise + cls._raise_usage_not_accessible(current_user) + if site is None: + raise ServiceException(message='客户现场不存在') + + @classmethod + async def _recover_integrity_error( + cls, + query_db: AsyncSession, + page_object: UsageCreateModel, + request_hash: str, + current_user: CurrentUserModel, + expected_site_id: int, + ) -> CrudResponseModel: + try: + site_ids = await cls._site_scope(query_db, current_user) + if site_ids is not None and expected_site_id not in site_ids: + cls._raise_pallet_not_accessible(current_user) + recovered = await OperationDao.get_usage_by_idempotency_key( + query_db, + page_object.idempotency_key, + site_ids=site_ids, + ) + if recovered is None: + raise ServiceException(message='使用失败，数据冲突') + if recovered.site_id != expected_site_id or not cls._is_exact_idempotent_result( + recovered, page_object, request_hash + ): + raise ServiceException(message='使用失败，幂等键已被其他请求使用') + await cls._lock_and_reauthorize_usage_site( + query_db, + recovered.site_id, + current_user, + ) + return cls._success(recovered.usage_id) + finally: + await query_db.rollback() + + @classmethod + async def get_usage_list_services( + cls, + query_db: AsyncSession, + query_object: UsagePageQueryModel, + current_user: CurrentUserModel, + is_page: bool = True, + ) -> PageModel | list[UsageModel]: + site_ids = await cls._site_scope(query_db, current_user) + if site_ids is not None and query_object.site_id is not None and query_object.site_id not in site_ids: + raise PermissionException(data='', message='无权访问该客户现场') + if site_ids == []: + if is_page: + return PageModel( + rows=[], + pageNum=query_object.page_num, + pageSize=query_object.page_size, + total=0, + hasNext=False, + ) + return [] + result = await OperationDao.get_usage_list( + query_db, + query_object, + is_page, + site_ids=site_ids, + ) + rows = result.rows if isinstance(result, PageModel) else result + models = [UsageModel.model_validate(record) for record in rows] + if isinstance(result, PageModel): + return result.model_copy(update={'rows': models}) + return models + + @classmethod + async def usage_detail_services( + cls, + query_db: AsyncSession, + usage_id: int, + current_user: CurrentUserModel, + ) -> UsageModel: + site_ids = await cls._site_scope(query_db, current_user) + usage = await OperationDao.get_usage_by_id( + query_db, + usage_id, + site_ids=site_ids, + ) + if usage is None: + cls._raise_usage_not_accessible(current_user) + return UsageModel.model_validate(usage) + + @classmethod + async def get_line_options_services( + cls, + query_db: AsyncSession, + site_id: int, + current_user: CurrentUserModel, + ) -> list[BusinessOptionModel]: + cls._current_identity(current_user) + await AccessService.assert_site_access(query_db, current_user, site_id) + site = await MasterDao.get_site(query_db, site_id) + if site is None: + raise ServiceException(message='客户现场不存在') + if site.status != '0': + raise ServiceException(message='客户现场已停用') + lines = await MasterDao.list_line( + query_db, + LinePageQueryModel(siteId=site_id, status='0'), + False, + ) + options = [ + BusinessOptionModel( + id=cls._record_value(line, 'line_id'), + code=cls._record_value(line, 'line_code'), + name=cls._record_value(line, 'line_name'), + ) + for line in lines + if cls._record_value(line, 'site_id') == site_id and cls._record_value(line, 'status') == '0' + ] + return sorted(options, key=lambda option: option.id) + + @classmethod + async def scan_usage_pallet_services( + cls, + query_db: AsyncSession, + qr_token: str, + current_user: CurrentUserModel, + ) -> PalletScanResponseModel: + normalized_qr = qr_token.strip() + pallet = await InventoryDao.get_pallet_by_qr_token(query_db, normalized_qr) + 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.site_id is None: + cls._raise_pallet_not_accessible(current_user, '托盘不在客户现场') + await cls._assert_pallet_site_access(query_db, current_user, inventory.site_id) + + if pallet.qr_token != normalized_qr or pallet.pallet_id != inventory.pallet_id: + raise ServiceException(message='托盘二维码与库存不一致') + if pallet.asset_status != '0': + raise ServiceException(message='托盘资产已停用，不允许使用') + if ( + inventory.flow_state + not in { + PalletFlowState.AT_SITE_UNUSED.value, + PalletFlowState.AT_SITE_PARTIAL.value, + } + or inventory.warehouse_id is not None + or inventory.active_filling_id is None + or type(inventory.remaining_packages) is not int + or inventory.remaining_packages <= 0 + or inventory.remaining_packages > pallet.capacity_packages + ): + raise ServiceException(message='托盘当前状态不允许使用') + filling = await OperationDao.get_filling_by_id(query_db, inventory.active_filling_id) + if filling is None: + raise ServiceException(message='托盘活动装填周期不存在') + if ( + filling.filling_id != inventory.active_filling_id + or filling.pallet_id != pallet.pallet_id + or filling.active_slot != 'ACTIVE' + or filling.closed_at is not None + ): + raise ServiceException(message='托盘活动装填周期无效') + return PalletScanResponseModel( + pallet_id=pallet.pallet_id, + pallet_code=pallet.pallet_code, + flow_state=PalletFlowState(inventory.flow_state), + warehouse_id=None, + site_id=inventory.site_id, + current_line_id=inventory.current_line_id, + active_filling_id=inventory.active_filling_id, + strain_id=filling.strain_id, + strain_code=filling.strain_code_snapshot, + strain_name=filling.strain_name_snapshot, + batch_no=filling.batch_no, + package_spec=filling.package_spec, + capacity_packages=pallet.capacity_packages, + remaining_packages=inventory.remaining_packages, + allowed_actions=['usage'], + ) + + @classmethod + async def record_usage_services( # noqa: PLR0912, PLR0915 + cls, + query_db: AsyncSession, + page_object: UsageCreateModel, + current_user: CurrentUserModel, + ) -> CrudResponseModel: + expected_site_id: int | None = None + try: + request_hash = cls._build_request_hash(page_object) + user_id, user_name = cls._current_identity(current_user) + site_ids = await cls._site_scope(query_db, current_user) + existing = await OperationDao.get_usage_by_idempotency_key( + query_db, + page_object.idempotency_key, + site_ids=site_ids, + ) + if existing is not None: + if not cls._is_exact_idempotent_result(existing, page_object, request_hash): + raise ServiceException(message='使用失败，幂等键已被其他请求使用') + if site_ids is not None and existing.site_id not in site_ids: + cls._raise_usage_not_accessible(current_user) + await cls._lock_and_reauthorize_usage_site( + query_db, + existing.site_id, + current_user, + ) + usage_id = existing.usage_id + await query_db.rollback() + return cls._success(usage_id) + + scanned_pallet = await InventoryDao.get_pallet_by_qr_token( + query_db, + page_object.qr_token.strip(), + ) + 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.site_id is None: + cls._raise_pallet_not_accessible(current_user, '托盘不在客户现场') + expected_site_id = inventory.site_id + await cls._assert_pallet_site_access(query_db, current_user, inventory.site_id) + + locked_pallets = await InventoryDao.get_pallets_by_ids( + query_db, + [inventory.pallet_id], + ) + if len(locked_pallets) != 1 or locked_pallets[0].pallet_id != inventory.pallet_id: + cls._raise_pallet_not_accessible(current_user) + pallet = locked_pallets[0] + + if inventory.active_filling_id is None: + raise ServiceException(message='托盘不存在活动装填周期') + filling = await OperationDao.get_active_filling_for_update( + query_db, + inventory.active_filling_id, + ) + if filling is None: + raise ServiceException(message='托盘活动装填周期不存在或已关闭') + + site = await MasterDao.get_site(query_db, inventory.site_id, for_update=True) + await cls._assert_pallet_site_access( + query_db, + current_user, + inventory.site_id, + for_update=True, + ) + if site is None: + raise ServiceException(message='客户现场不存在') + if site.status != '0': + raise ServiceException(message='客户现场已停用') + line = await MasterDao.get_line(query_db, page_object.line_id, for_update=True) + if line is None: + raise ServiceException(message='产线不存在') + if line.status != '0': + raise ServiceException(message='产线已停用') + if line.site_id != inventory.site_id: + raise ServiceException(message='产线不属于托盘当前客户现场') + + if pallet.pallet_id != inventory.pallet_id or pallet.qr_token != page_object.qr_token.strip(): + raise ServiceException(message='托盘二维码与库存不一致') + if pallet.asset_status != '0': + raise ServiceException(message='托盘资产已停用，不允许使用') + if inventory.flow_state not in { + PalletFlowState.AT_SITE_UNUSED.value, + PalletFlowState.AT_SITE_PARTIAL.value, + }: + raise ServiceException(message='托盘当前状态不允许使用') + if inventory.warehouse_id is not None: + raise ServiceException(message='托盘当前位置无效') + if type(inventory.remaining_packages) is not int or inventory.remaining_packages <= 0: + raise ServiceException(message='托盘当前余量无效') + if inventory.remaining_packages > pallet.capacity_packages: + raise ServiceException(message='托盘当前余量超过容量') + if type(inventory.version) is not int or inventory.version < 0: + raise ServiceException(message='托盘库存版本无效') + if ( + filling.filling_id != inventory.active_filling_id + or filling.pallet_id != pallet.pallet_id + or filling.active_slot != 'ACTIVE' + or filling.closed_at is not None + ): + raise ServiceException(message='托盘活动装填周期无效') + if inventory.current_line_id is not None and inventory.current_line_id != line.line_id: + raise ServiceException(message='托盘本次驻场周期已绑定其他产线') + if type(page_object.used_packages) is not int or page_object.used_packages <= 0: + raise ServiceException(message='使用包数必须为正整数') + if page_object.used_packages > inventory.remaining_packages: + raise ServiceException(message='使用包数不能超过当前余量') + + remaining_after = inventory.remaining_packages - page_object.used_packages + target_state = PalletFlowState.AT_SITE_EMPTY if remaining_after == 0 else PalletFlowState.AT_SITE_PARTIAL + PalletStateMachine.ensure_transition( + PalletFlowState(inventory.flow_state), + target_state, + ) + + current_time = datetime.now() + usage = BizUsageRecord( + usage_no=f'US-{secrets.token_hex(16)}', + pallet_id=pallet.pallet_id, + active_filling_id=filling.filling_id, + site_id=inventory.site_id, + line_id=line.line_id, + used_packages=page_object.used_packages, + remaining_packages_after=remaining_after, + idempotency_key=page_object.idempotency_key, + request_hash=request_hash, + operator_id=user_id, + occurred_at=current_time, + ) + usage = await OperationDao.add_usage_record(query_db, usage) + usage_id = usage.usage_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.current_line_id = line.line_id + inventory.remaining_packages = remaining_after + inventory.flow_state = target_state.value + if target_state is PalletFlowState.AT_SITE_EMPTY: + filling.active_slot = None + filling.closed_at = current_time + filling.update_by = user_name + filling.update_time = current_time + await OperationDao.update_filling_record(query_db, filling) + inventory.active_filling_id = None + inventory.version += 1 + inventory.updated_at = current_time + event = BizPalletEvent( + pallet_id=inventory.pallet_id, + event_type=PalletEventType.USAGE_RECORDED.value, + source_type='USAGE', + source_id=usage_id, + before_state=before_state, + after_state=target_state.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=remaining_after, + 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(usage_id) + except IntegrityError: + await query_db.rollback() + if expected_site_id is None: + raise ServiceException(message='使用失败，数据冲突') from None + return await cls._recover_integrity_error( + query_db, + page_object, + request_hash, + current_user, + expected_site_id, + ) + except Exception: + await query_db.rollback() + raise diff --git a/shierfeng-fastapi-backend/tests/module_business/test_inventory_dao.py b/shierfeng-fastapi-backend/tests/module_business/test_inventory_dao.py index d899d26..c7503ac 100644 --- a/shierfeng-fastapi-backend/tests/module_business/test_inventory_dao.py +++ b/shierfeng-fastapi-backend/tests/module_business/test_inventory_dao.py @@ -133,7 +133,7 @@ class TestDaoPublicContract(unittest.TestCase): 'site_ids', ), 'get_filling_by_idempotency_key': ('db', 'idempotency_key', 'warehouse_ids'), - 'get_usage_by_idempotency_key': ('db', 'idempotency_key'), + 'get_usage_by_idempotency_key': ('db', 'idempotency_key', 'site_ids'), 'get_return_by_request_idempotency_key': ('db', 'idempotency_key'), 'get_return_by_confirm_idempotency_key': ('db', 'idempotency_key'), 'get_inbound_by_id': ('db', 'inbound_id', 'warehouse_ids'), @@ -149,7 +149,7 @@ class TestDaoPublicContract(unittest.TestCase): 'update_shipment': ('db', 'shipment'), 'delete_shipment': ('db', 'shipment_id'), 'replace_shipment_items': ('db', 'shipment_id', 'items'), - 'get_usage_by_id': ('db', 'usage_id'), + 'get_usage_by_id': ('db', 'usage_id', 'site_ids'), 'add_usage_record': ('db', 'usage'), 'get_return_by_id': ('db', 'return_id'), 'add_return_order': ('db', 'return_order'), @@ -209,11 +209,25 @@ class TestDaoPublicContract(unittest.TestCase): self.assertIsNone(shipment_list_signature.parameters['warehouse_ids'].default) self.assertIsNone(shipment_list_signature.parameters['site_ids'].default) - for method_name in ('get_usage_list', 'get_return_list'): + usage_list_signature = inspect.signature(OperationDao.get_usage_list) + self.assertEqual( + ('db', 'query_object', 'is_page', 'site_ids'), + tuple(usage_list_signature.parameters), + ) + self.assertIs(True, usage_list_signature.parameters['is_page'].default) + self.assertIsNone(usage_list_signature.parameters['site_ids'].default) + + for method_name in ( + 'get_usage_by_idempotency_key', + 'get_usage_by_id', + ): with self.subTest(method=method_name): signature = inspect.signature(getattr(OperationDao, method_name)) - self.assertEqual(('db', 'query_object', 'is_page'), tuple(signature.parameters)) - self.assertIs(True, signature.parameters['is_page'].default) + self.assertIsNone(signature.parameters['site_ids'].default) + + return_list_signature = inspect.signature(OperationDao.get_return_list) + self.assertEqual(('db', 'query_object', 'is_page'), tuple(return_list_signature.parameters)) + self.assertIs(True, return_list_signature.parameters['is_page'].default) def test_dao_source_never_controls_transaction_boundary(self) -> None: for module in (inventory_dao_module, operation_dao_module): @@ -864,6 +878,71 @@ class TestOperationDaoReads(unittest.IsolatedAsyncioTestCase): self.assertIn("biz_shipment_order.confirm_idempotency_key = 'confirm-key'", key_sql) self.assertNotIn('FOR UPDATE', key_sql) + async def test_usage_reads_push_site_scope_and_empty_scope_is_always_false(self) -> None: + query = UsagePageQueryModel(page_num=2, page_size=7) + marker = object() + with patch.object(PageUtil, 'paginate', new=AsyncMock(return_value=marker)) as paginate: + actual = await OperationDao.get_usage_list( + self.db, + query, + is_page=True, + site_ids=[11, 9, 11], + ) + + self.assertIs(marker, actual) + list_statement = paginate.await_args.args[1] + + scalar_result = MagicMock() + scalar_result.scalar_one_or_none.return_value = None + self.db.execute.return_value = scalar_result + await OperationDao.get_usage_by_id(self.db, 15, site_ids=[11, 9, 11]) + detail_statement = self.db.execute.await_args.args[0] + + self.db.reset_mock() + self.db.execute.return_value = scalar_result + await OperationDao.get_usage_by_idempotency_key( + self.db, + 'usage-key', + site_ids=[11, 9, 11], + ) + key_statement = self.db.execute.await_args.args[0] + + for dialect in (mysql.dialect(), postgresql.dialect()): + with self.subTest(dialect=dialect.name): + for statement in (list_statement, detail_statement, key_statement): + sql = compile_sql(statement, dialect) + self.assertIn('biz_usage_record.site_id IN (9, 11)', sql) + + with patch.object(PageUtil, 'paginate', new=AsyncMock(return_value=marker)) as empty_paginate: + await OperationDao.get_usage_list(self.db, query, site_ids=[]) + empty_list = empty_paginate.await_args.args[1] + + self.db.reset_mock() + self.db.execute.return_value = scalar_result + await OperationDao.get_usage_by_id(self.db, 15, site_ids=[]) + empty_detail = self.db.execute.await_args.args[0] + + self.db.reset_mock() + self.db.execute.return_value = scalar_result + await OperationDao.get_usage_by_idempotency_key(self.db, 'usage-key', site_ids=[]) + empty_key = self.db.execute.await_args.args[0] + + for dialect in (mysql.dialect(), postgresql.dialect()): + with self.subTest(dialect=f'{dialect.name}-empty'): + for statement in (empty_list, empty_detail, empty_key): + sql = compile_sql(statement, dialect) + self.assertTrue('false' in sql.lower() or '0 = 1' in sql, sql) + + async def test_usage_none_scope_adds_no_authorization_predicate(self) -> None: + query = UsagePageQueryModel() + marker = object() + with patch.object(PageUtil, 'paginate', new=AsyncMock(return_value=marker)) as paginate: + await OperationDao.get_usage_list(self.db, query, site_ids=None) + + sql = compile_sql(paginate.await_args.args[1]) + self.assertNotIn('site_id IN', sql) + self.assertNotIn('WHERE false', sql) + def test_time_conditions_include_fractional_last_second_with_exclusive_next_day(self) -> None: lower, upper = OperationDao._time_conditions( BizUsageRecord.occurred_at, diff --git a/shierfeng-fastapi-backend/tests/module_business/test_master_service.py b/shierfeng-fastapi-backend/tests/module_business/test_master_service.py index a927755..0c7c30e 100644 --- a/shierfeng-fastapi-backend/tests/module_business/test_master_service.py +++ b/shierfeng-fastapi-backend/tests/module_business/test_master_service.py @@ -503,13 +503,13 @@ class TestMasterDao(unittest.IsolatedAsyncioTestCase): self.assertIn(f'{table}.{id_field} = 9', sql) async def test_locking_master_gets_use_current_reads_and_refresh_identity_map(self) -> None: - strain_lock_parameter = inspect.signature(MasterDao.get_strain).parameters.get('for_update') - self.assertIsNotNone(strain_lock_parameter) - self.assertIs(inspect.Parameter.KEYWORD_ONLY, strain_lock_parameter.kind) - self.assertFalse(strain_lock_parameter.default) - - for resource in ('warehouse', 'strain'): + for resource in ('warehouse', 'strain', 'line'): with self.subTest(resource=resource): + lock_parameter = inspect.signature(getattr(MasterDao, f'get_{resource}')).parameters.get('for_update') + self.assertIsNotNone(lock_parameter) + self.assertIs(inspect.Parameter.KEYWORD_ONLY, lock_parameter.kind) + self.assertFalse(lock_parameter.default) + result = MagicMock() result.scalar_one_or_none.return_value = None self.db.execute = AsyncMock(return_value=result) diff --git a/shierfeng-fastapi-backend/tests/module_business/test_usage_service.py b/shierfeng-fastapi-backend/tests/module_business/test_usage_service.py new file mode 100644 index 0000000..cbf75bb --- /dev/null +++ b/shierfeng-fastapi-backend/tests/module_business/test_usage_service.py @@ -0,0 +1,1201 @@ +import hashlib +import importlib +import importlib.util +import inspect +import json +import unittest +from contextlib import ExitStack +from datetime import datetime +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.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 BizCustomerSite, BizPallet, BizProductionLine +from module_business.entity.do.operation_do import ( + BizFillingRecord, + BizPalletEvent, + BizPalletInventory, + BizUsageRecord, +) +from module_business.entity.vo.master_vo import LinePageQueryModel +from module_business.entity.vo.operation_vo import ( + BusinessOptionModel, + PalletScanResponseModel, + UsageCreateModel, + UsageModel, + UsagePageQueryModel, +) +from module_business.service.access_service import AccessService + +SERVICE_SPEC = importlib.util.find_spec('module_business.service.usage_service') +CONTROLLER_SPEC = importlib.util.find_spec('module_business.controller.usage_controller') +UsageService = None +if SERVICE_SPEC is not None: + UsageService = getattr( + importlib.import_module('module_business.service.usage_service'), + 'UsageService', + None, + ) +usage_controller_module = None +usage_controller = None +if CONTROLLER_SPEC is not None: + usage_controller_module = importlib.import_module('module_business.controller.usage_controller') + usage_controller = getattr(usage_controller_module, 'usage_controller', None) + + +def make_user(user_id: int = 7) -> CurrentUserModel: + return CurrentUserModel( + permissions=[], + roles=[], + user=UserInfoModel(userId=user_id, userName='admin' if user_id == 1 else f'user-{user_id}'), + ) + + +def integrity_error() -> IntegrityError: + return IntegrityError('statement', {}, RuntimeError('constraint')) + + +def make_page( + *, + qr_token: str = 'qr-2', + line_id: int = 4, + used_packages: int = 3, + idempotency_key: str = 'usage-key', +) -> UsageCreateModel: + return UsageCreateModel( + qrToken=qr_token, + lineId=line_id, + usedPackages=used_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, + *, + site_id: int | None = 9, + current_line_id: int | None = None, + active_filling_id: int | None = 31, + remaining_packages: int = 8, + flow_state: str = PalletFlowState.AT_SITE_UNUSED.value, + version: int = 3, +) -> BizPalletInventory: + return BizPalletInventory( + pallet_id=pallet_id, + active_filling_id=active_filling_id, + warehouse_id=None, + site_id=site_id, + current_line_id=current_line_id, + remaining_packages=remaining_packages, + flow_state=flow_state, + last_event_id=40, + version=version, + updated_at=datetime(2026, 7, 25, 9, 0), + ) + + +def make_filling( + filling_id: int = 31, + *, + pallet_id: int = 2, + active_slot: str | None = 'ACTIVE', +) -> BizFillingRecord: + return BizFillingRecord( + filling_id=filling_id, + filling_no=f'FL-{filling_id}', + pallet_id=pallet_id, + warehouse_id=7, + strain_id=11, + strain_code_snapshot='S-011', + strain_name_snapshot='香菇 11', + batch_no='B-20260726', + package_spec='20kg', + initial_packages=20, + active_slot=active_slot, + closed_at=None, + idempotency_key='fill-key', + request_hash='f' * 64, + operator_id=7, + operated_at=datetime(2026, 7, 25, 8, 0), + ) + + +def make_site(site_id: int = 9, *, status: str = '0') -> BizCustomerSite: + return BizCustomerSite( + site_id=site_id, + customer_id=3, + site_code=f'SITE-{site_id}', + site_name=f'现场 {site_id}', + status=status, + ) + + +def make_line( + line_id: int = 4, + *, + site_id: int = 9, + status: str = '0', +) -> BizProductionLine: + return BizProductionLine( + line_id=line_id, + site_id=site_id, + line_code=f'L-{line_id}', + line_name=f'产线 {line_id}', + status=status, + ) + + +def make_usage( + usage_id: int = 61, + *, + site_id: int = 9, + line_id: int = 4, + idempotency_key: str = 'usage-key', + request_hash: str = 'a' * 64, +) -> BizUsageRecord: + return BizUsageRecord( + usage_id=usage_id, + usage_no=f'US-{usage_id}', + pallet_id=2, + active_filling_id=31, + site_id=site_id, + line_id=line_id, + used_packages=3, + remaining_packages_after=5, + idempotency_key=idempotency_key, + request_hash=request_hash, + operator_id=7, + occurred_at=datetime(2026, 7, 26, 11, 30), + ) + + +class TestUsageBootstrap(unittest.TestCase): + def test_usage_service_module_exists(self) -> None: + self.assertIsNotNone(SERVICE_SPEC, 'UsageService 尚未实现') + + def test_usage_service_class_exists(self) -> None: + self.assertIsNotNone(UsageService, 'UsageService 尚未实现') + + def test_usage_controller_module_exists(self) -> None: + self.assertIsNotNone(CONTROLLER_SPEC, 'usage_controller 尚未实现') + + def test_usage_controller_router_exists(self) -> None: + self.assertIsNotNone(usage_controller, 'usage_controller 路由尚未实现') + + +@unittest.skipUnless(UsageService is not None, 'UsageService 尚未实现') +class TestUsagePublicContract(unittest.TestCase): + def test_service_exposes_complete_usage_workflow(self) -> None: + expected = { + 'get_usage_list_services', + 'usage_detail_services', + 'get_line_options_services', + 'scan_usage_pallet_services', + 'record_usage_services', + } + self.assertEqual(expected, {name for name in expected if hasattr(UsageService, name)}) + + def test_record_signature_and_types_are_exact(self) -> None: + signature = inspect.signature(UsageService.record_usage_services) + self.assertEqual(('query_db', 'page_object', 'current_user'), tuple(signature.parameters)) + hints = get_type_hints(UsageService.record_usage_services) + self.assertIs(AsyncSession, hints['query_db']) + self.assertIs(UsageCreateModel, hints['page_object']) + self.assertIs(CurrentUserModel, hints['current_user']) + self.assertIs(CrudResponseModel, hints['return']) + + def test_request_hash_is_stable_normalized_and_excludes_idempotency_key(self) -> None: + page = make_page(qr_token=' qr-2 ', idempotency_key='key-a') + expected = hashlib.sha256( + json.dumps( + {'line_id': 4, 'qr_token': 'qr-2', 'used_packages': 3}, + sort_keys=True, + separators=(',', ':'), + ).encode() + ).hexdigest() + + self.assertEqual(expected, UsageService._build_request_hash(page)) + self.assertEqual( + expected, + UsageService._build_request_hash(page.model_copy(update={'idempotency_key': 'key-b'})), + ) + self.assertNotEqual( + expected, + UsageService._build_request_hash(page.model_copy(update={'used_packages': 2})), + ) + + 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(used_packages=invalid) + + +@unittest.skipUnless(UsageService is not None, 'UsageService 尚未实现') +class TestUsageReadServices(unittest.IsolatedAsyncioTestCase): + def setUp(self) -> None: + self.db = MagicMock(spec=AsyncSession) + self.user = make_user() + + async def test_list_pushes_current_site_scope_and_converts_rows(self) -> None: + query = UsagePageQueryModel(pageNum=1, pageSize=10) + page = PageModel( + rows=[make_usage()], + pageNum=1, + pageSize=10, + total=1, + hasNext=False, + ) + get_list = AsyncMock(return_value=page) + with ( + patch.object(AccessDao, 'list_user_site_ids', new=AsyncMock(return_value=[11, 9, 9])), + patch.object(OperationDao, 'get_usage_list', new=get_list), + ): + result = await UsageService.get_usage_list_services(self.db, query, self.user) + + get_list.assert_awaited_once_with(self.db, query, True, site_ids=[9, 11]) + self.assertEqual(1, result.total) + self.assertIsInstance(result.rows[0], UsageModel) + + async def test_empty_scope_returns_empty_without_global_query_or_total(self) -> None: + for is_page in (True, False): + with self.subTest(is_page=is_page): + get_list = AsyncMock() + with ( + patch.object(AccessDao, 'list_user_site_ids', new=AsyncMock(return_value=[])), + patch.object(OperationDao, 'get_usage_list', new=get_list), + ): + result = await UsageService.get_usage_list_services( + self.db, + UsagePageQueryModel(pageNum=2, pageSize=7), + self.user, + is_page=is_page, + ) + + get_list.assert_not_awaited() + if is_page: + self.assertEqual([], result.rows) + self.assertEqual(0, result.total) + self.assertEqual(2, result.page_num) + self.assertEqual(7, result.page_size) + else: + self.assertEqual([], result) + + async def test_list_rejects_explicit_site_outside_scope_before_dao(self) -> None: + get_list = AsyncMock() + with ( + patch.object(AccessDao, 'list_user_site_ids', new=AsyncMock(return_value=[9])), + patch.object(OperationDao, 'get_usage_list', new=get_list), + self.assertRaises(PermissionException) as raised, + ): + await UsageService.get_usage_list_services( + self.db, + UsagePageQueryModel(siteId=10), + self.user, + ) + + self.assertEqual('无权访问该客户现场', raised.exception.message) + get_list.assert_not_awaited() + + async def test_admin_list_is_unscoped(self) -> None: + query = UsagePageQueryModel() + get_scope = AsyncMock() + get_list = AsyncMock(return_value=[]) + with ( + patch.object(AccessDao, 'list_user_site_ids', new=get_scope), + patch.object(OperationDao, 'get_usage_list', new=get_list), + ): + result = await UsageService.get_usage_list_services( + self.db, + query, + make_user(1), + is_page=False, + ) + + get_scope.assert_not_awaited() + get_list.assert_awaited_once_with(self.db, query, False, site_ids=None) + self.assertEqual([], result) + + async def test_detail_is_scoped_and_hides_missing_but_admin_gets_precise_error(self) -> None: + for user, expected_exception, expected_message, expected_scope in ( + (self.user, PermissionException, '无权访问该使用记录', [9]), + (make_user(1), ServiceException, '使用记录不存在', None), + ): + get_detail = AsyncMock(return_value=None) + with ( + self.subTest(admin=user.user.admin), + patch.object(AccessDao, 'list_user_site_ids', new=AsyncMock(return_value=[9])), + patch.object(OperationDao, 'get_usage_by_id', new=get_detail), + self.assertRaises(expected_exception) as raised, + ): + await UsageService.usage_detail_services(self.db, 61, user) + + get_detail.assert_awaited_once_with(self.db, 61, site_ids=expected_scope) + self.assertEqual(expected_message, raised.exception.message) + + async def test_line_options_require_site_access_and_only_return_enabled_lines_for_site(self) -> None: + rows = [ + {'lineId': 6, 'siteId': 9, 'lineCode': 'L-6', 'lineName': '六线', 'status': '0'}, + {'lineId': 5, 'siteId': 10, 'lineCode': 'L-5', 'lineName': '五线', 'status': '0'}, + {'lineId': 4, 'siteId': 9, 'lineCode': 'L-4', 'lineName': '四线', 'status': '1'}, + {'lineId': 3, 'siteId': 9, 'lineCode': 'L-3', 'lineName': '三线', 'status': '0'}, + ] + access = AsyncMock() + list_line = AsyncMock(return_value=rows) + with ( + patch.object(AccessService, 'assert_site_access', new=access), + patch.object(MasterDao, 'get_site', new=AsyncMock(return_value=make_site())), + patch.object(MasterDao, 'list_line', new=list_line), + ): + result = await UsageService.get_line_options_services(self.db, 9, self.user) + + access.assert_awaited_once_with(self.db, self.user, 9) + query = list_line.await_args.args[1] + self.assertIsInstance(query, LinePageQueryModel) + self.assertEqual(9, query.site_id) + self.assertEqual('0', query.status) + self.assertFalse(list_line.await_args.args[2]) + self.assertEqual([3, 6], [option.id for option in result]) + self.assertTrue(all(isinstance(option, BusinessOptionModel) for option in result)) + + async def test_line_options_denial_happens_before_master_query(self) -> None: + get_site = AsyncMock() + list_line = AsyncMock() + with ( + patch.object( + AccessService, + 'assert_site_access', + new=AsyncMock(side_effect=PermissionException(data='', message='无权访问该客户现场')), + ), + patch.object(MasterDao, 'get_site', new=get_site), + patch.object(MasterDao, 'list_line', new=list_line), + self.assertRaises(PermissionException), + ): + await UsageService.get_line_options_services(self.db, 9, self.user) + + get_site.assert_not_awaited() + list_line.assert_not_awaited() + + async def test_scan_is_read_only_and_returns_only_usage_action(self) -> None: + pallet = make_pallet() + inventory = make_inventory( + current_line_id=4, + remaining_packages=5, + flow_state=PalletFlowState.AT_SITE_PARTIAL.value, + ) + filling = make_filling() + access = AsyncMock() + inventory_lock = AsyncMock() + pallet_lock = AsyncMock() + filling_lock = AsyncMock() + get_filling = AsyncMock(return_value=filling) + 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_site_access', new=access), + patch.object(OperationDao, 'get_filling_by_id', new=get_filling), + patch.object(InventoryDao, 'get_inventory_for_update', new=inventory_lock), + patch.object(InventoryDao, 'get_pallets_by_ids', new=pallet_lock), + patch.object(OperationDao, 'get_active_filling_for_update', new=filling_lock), + ): + result = await UsageService.scan_usage_pallet_services(self.db, 'qr-2', self.user) + + access.assert_awaited_once_with(self.db, self.user, 9) + get_filling.assert_awaited_once_with(self.db, 31) + inventory_lock.assert_not_awaited() + pallet_lock.assert_not_awaited() + filling_lock.assert_not_awaited() + self.db.commit.assert_not_awaited() + self.assertIsInstance(result, PalletScanResponseModel) + self.assertEqual(PalletFlowState.AT_SITE_PARTIAL, result.flow_state) + self.assertEqual(9, result.site_id) + self.assertEqual(4, result.current_line_id) + self.assertEqual(31, result.active_filling_id) + self.assertEqual('S-011', result.strain_code) + self.assertEqual('B-20260726', result.batch_no) + self.assertEqual(5, result.remaining_packages) + self.assertEqual(['usage'], result.allowed_actions) + + async def test_scan_hides_missing_non_site_and_unauthorized_qr_for_normal_user(self) -> None: + with ( + patch.object(InventoryDao, 'get_pallet_by_qr_token', new=AsyncMock(return_value=None)), + self.assertRaises(PermissionException) as missing, + ): + await UsageService.scan_usage_pallet_services(self.db, 'missing', self.user) + self.assertEqual('无权访问该托盘', missing.exception.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(site_id=None))), + self.assertRaises(PermissionException) as non_site, + ): + await UsageService.scan_usage_pallet_services(self.db, 'qr-2', self.user) + self.assertEqual('无权访问该托盘', non_site.exception.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(site_id=10))), + patch.object( + AccessService, + 'assert_site_access', + new=AsyncMock(side_effect=PermissionException(data='', message='无权访问该客户现场')), + ), + self.assertRaises(PermissionException) as unauthorized, + ): + await UsageService.scan_usage_pallet_services(self.db, 'qr-2', self.user) + self.assertEqual('无权访问该托盘', unauthorized.exception.message) + + async def test_scan_admin_keeps_precise_missing_error(self) -> None: + with ( + patch.object(InventoryDao, 'get_pallet_by_qr_token', new=AsyncMock(return_value=None)), + self.assertRaises(ServiceException) as raised, + ): + await UsageService.scan_usage_pallet_services(self.db, 'missing', make_user(1)) + self.assertEqual('托盘不存在', raised.exception.message) + + +@unittest.skipUnless(usage_controller is not None, 'usage_controller 尚未实现') +class TestUsageControllerContract(unittest.TestCase): + @staticmethod + def _routes() -> dict[tuple[str, str], APIRoute]: + return { + (route.path, next(iter(route.methods))): route + for route in usage_controller.routes + if isinstance(route, APIRoute) + } + + def test_openapi_has_exact_five_operations_aliases_and_response_models(self) -> None: + app = FastAPI() + app.include_router(usage_controller) + schema = app.openapi() + expected = { + ('/business/usage/list', 'get'), + ('/business/usage/line-options', 'get'), + ('/business/usage/scan/{qr_token}', 'get'), + ('/business/usage/{usage_id}', 'get'), + ('/business/usage', 'post'), + } + self.assertEqual(expected, {(path, method) for path, methods in schema['paths'].items() for method in methods}) + self.assertEqual( + {'siteId'}, + {parameter['name'] for parameter in schema['paths']['/business/usage/line-options']['get']['parameters']}, + ) + self.assertEqual( + {'qr_token'}, + { + parameter['name'] + for parameter in schema['paths']['/business/usage/scan/{qr_token}']['get']['parameters'] + }, + ) + + routes = self._routes() + self.assertEqual(PageResponseModel[UsageModel], routes[('/business/usage/list', 'GET')].response_model) + self.assertEqual( + DataResponseModel[list[BusinessOptionModel]], + routes[('/business/usage/line-options', 'GET')].response_model, + ) + self.assertEqual( + DataResponseModel[PalletScanResponseModel], + routes[('/business/usage/scan/{qr_token}', 'GET')].response_model, + ) + self.assertEqual( + DataResponseModel[UsageModel], + routes[('/business/usage/{usage_id}', 'GET')].response_model, + ) + self.assertEqual(ResponseBaseModel, routes[('/business/usage', 'POST')].response_model) + + def test_static_routes_precede_dynamic_detail(self) -> None: + paths = [route.path for route in usage_controller.routes] + detail_index = paths.index('/business/usage/{usage_id}') + self.assertLess(paths.index('/business/usage/line-options'), detail_index) + self.assertLess(paths.index('/business/usage/scan/{qr_token}'), detail_index) + + def test_routes_have_exact_three_permissions_current_user_and_only_post_log(self) -> None: + expected = { + ('/business/usage/list', 'GET'): 'business:usage:list', + ('/business/usage/line-options', 'GET'): 'business:usage:query', + ('/business/usage/scan/{qr_token}', 'GET'): 'business:usage:query', + ('/business/usage/{usage_id}', 'GET'): 'business:usage:query', + ('/business/usage', 'POST'): 'business:usage: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)) + 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/usage', 'POST'), hasattr(route.endpoint, '__wrapped__')) + actual[key] = permissions[0].perm + + self.assertEqual(expected, actual) + self.assertEqual( + {'business:usage:list', 'business:usage:query', 'business:usage:add'}, + set(actual.values()), + ) + + def test_controller_does_not_add_forbidden_permissions(self) -> None: + source = inspect.getsource(usage_controller_module) + for forbidden in ( + 'business:usage:scan', + 'business:usage:export', + 'business:usage:edit', + 'business:usage:remove', + 'business:usage:confirm', + ): + self.assertNotIn(forbidden, source) + + +@unittest.skipUnless(UsageService is not None, 'UsageService 尚未实现') +class TestRecordUsageCore(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.filling = make_filling() + self.site = make_site() + self.line = make_line() + self.fixed_time = datetime(2026, 7, 26, 11, 30) + + def _patch_record_path( # noqa: PLR0913 + self, + *, + site_scope: list[int] | None | AsyncMock = None, + key_read: BizUsageRecord | None | AsyncMock = None, + pallet: BizPallet | None = None, + inventory: BizPalletInventory | None = None, + filling: BizFillingRecord | None = None, + site: BizCustomerSite | None = None, + line: BizProductionLine | None = None, + access: AsyncMock | None = None, + add_usage: AsyncMock | None = None, + update_filling: AsyncMock | None = None, + add_event: AsyncMock | None = None, + ) -> tuple[ExitStack, dict[str, AsyncMock]]: + stack = ExitStack() + if isinstance(site_scope, AsyncMock): + scope_read = site_scope + else: + scope_read = AsyncMock(return_value=[9] if site_scope is None else site_scope) + usage_key_read = key_read if isinstance(key_read, AsyncMock) else AsyncMock(return_value=key_read) + selected_pallet = pallet or self.pallet + selected_inventory = inventory or self.inventory + selected_filling = filling or self.filling + selected_site = self.site if site is None else site + selected_line = self.line if line is None else line + site_access = access or AsyncMock() + + async def assign_usage_id(_db: AsyncSession, usage: BizUsageRecord) -> BizUsageRecord: + usage.usage_id = 61 + return usage + + usage_write = add_usage or AsyncMock(side_effect=assign_usage_id) + filling_write = update_filling or AsyncMock() + event_write = add_event or AsyncMock() + inventory_lock = AsyncMock(return_value=selected_inventory) + pallet_lock = AsyncMock(return_value=[selected_pallet]) + filling_lock = AsyncMock(return_value=selected_filling) + site_lock = AsyncMock(return_value=selected_site) + line_lock = AsyncMock(return_value=selected_line) + + stack.enter_context(patch.object(AccessDao, 'list_user_site_ids', new=scope_read)) + stack.enter_context(patch.object(AccessService, 'assert_site_access', new=site_access)) + stack.enter_context(patch.object(OperationDao, 'get_usage_by_idempotency_key', new=usage_key_read)) + stack.enter_context( + patch.object(InventoryDao, 'get_pallet_by_qr_token', new=AsyncMock(return_value=selected_pallet)) + ) + stack.enter_context(patch.object(InventoryDao, 'get_inventory_for_update', new=inventory_lock)) + stack.enter_context(patch.object(InventoryDao, 'get_pallets_by_ids', new=pallet_lock)) + stack.enter_context(patch.object(OperationDao, 'get_active_filling_for_update', new=filling_lock)) + stack.enter_context(patch.object(MasterDao, 'get_site', new=site_lock)) + stack.enter_context(patch.object(MasterDao, 'get_line', new=line_lock)) + stack.enter_context(patch.object(OperationDao, 'add_usage_record', new=usage_write)) + stack.enter_context(patch.object(OperationDao, 'update_filling_record', new=filling_write)) + stack.enter_context(patch.object(InventoryDao, 'add_event', new=event_write)) + mocked_datetime = stack.enter_context(patch('module_business.service.usage_service.datetime')) + mocked_datetime.now.return_value = self.fixed_time + return stack, { + 'scope_read': scope_read, + 'key_read': usage_key_read, + 'access': site_access, + 'inventory_lock': inventory_lock, + 'pallet_lock': pallet_lock, + 'filling_lock': filling_lock, + 'site_lock': site_lock, + 'line_lock': line_lock, + 'add_usage': usage_write, + 'update_filling': filling_write, + 'add_event': event_write, + } + + async def test_first_partial_usage_locks_authoritative_rows_then_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 lock_filling(*_args: Any) -> BizFillingRecord: + actions.append('filling-lock') + return self.filling + + async def lock_site(*_args: Any, **_kwargs: Any) -> BizCustomerSite: + actions.append('site-lock') + return self.site + + async def lock_line(*_args: Any, **_kwargs: Any) -> BizProductionLine: + actions.append('line-lock') + return self.line + + async def assert_site_access( + _db: AsyncSession, + _user: CurrentUserModel, + _site_id: int, + *, + for_update: bool = False, + ) -> None: + if for_update: + actions.append('site-mapping-lock') + + async def add_usage(_db: AsyncSession, usage: BizUsageRecord) -> BizUsageRecord: + actions.append('usage-flush') + usage.usage_id = 61 + captured['usage'] = usage + return usage + + 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['usage'].__dict__.pop('usage_id', None) + + self.db.commit.side_effect = commit + access = AsyncMock(side_effect=assert_site_access) + update_filling = AsyncMock() + with ( + patch.object(AccessDao, 'list_user_site_ids', new=AsyncMock(return_value=[9])), + patch.object(AccessService, 'assert_site_access', new=access), + patch.object(OperationDao, 'get_usage_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(OperationDao, 'get_active_filling_for_update', new=AsyncMock(side_effect=lock_filling)), + patch.object(MasterDao, 'get_site', new=AsyncMock(side_effect=lock_site)), + patch.object(MasterDao, 'get_line', new=AsyncMock(side_effect=lock_line)), + patch.object(OperationDao, 'add_usage_record', new=AsyncMock(side_effect=add_usage)), + patch.object(OperationDao, 'update_filling_record', new=update_filling), + patch.object(InventoryDao, 'add_event', new=AsyncMock(side_effect=add_event)), + patch('module_business.service.usage_service.datetime') as mocked_datetime, + ): + mocked_datetime.now.return_value = self.fixed_time + result = await UsageService.record_usage_services(self.db, self.page, self.user) + + self.assertEqual( + [ + 'inventory-lock', + 'pallet-lock', + 'filling-lock', + 'site-lock', + 'site-mapping-lock', + 'line-lock', + 'usage-flush', + 'event-flush', + 'commit', + ], + actions, + ) + self.assertEqual([False, True], [call.kwargs.get('for_update', False) for call in access.await_args_list]) + self.assertEqual([9, 9], [call.args[2] for call in access.await_args_list]) + self.assertEqual(4, self.inventory.current_line_id) + self.assertEqual(5, self.inventory.remaining_packages) + self.assertEqual(PalletFlowState.AT_SITE_PARTIAL.value, self.inventory.flow_state) + self.assertEqual(31, self.inventory.active_filling_id) + self.assertEqual(4, self.inventory.version) + self.assertEqual(self.fixed_time, self.inventory.updated_at) + self.assertEqual('ACTIVE', self.filling.active_slot) + self.assertIsNone(self.filling.closed_at) + update_filling.assert_not_awaited() + + usage = captured['usage'] + self.assertTrue(usage.usage_no.startswith('US-')) + self.assertEqual(2, usage.pallet_id) + self.assertEqual(31, usage.active_filling_id) + self.assertEqual(9, usage.site_id) + self.assertEqual(4, usage.line_id) + self.assertEqual(3, usage.used_packages) + self.assertEqual(5, usage.remaining_packages_after) + self.assertEqual('usage-key', usage.idempotency_key) + self.assertEqual(UsageService._build_request_hash(self.page), usage.request_hash) + self.assertEqual(7, usage.operator_id) + self.assertEqual(self.fixed_time, usage.occurred_at) + + event = captured['event'] + self.assertEqual(PalletEventType.USAGE_RECORDED.value, event.event_type) + self.assertEqual('USAGE', event.source_type) + self.assertEqual(61, event.source_id) + self.assertEqual(PalletFlowState.AT_SITE_UNUSED.value, event.before_state) + self.assertEqual(PalletFlowState.AT_SITE_PARTIAL.value, event.after_state) + self.assertIsNone(event.before_warehouse_id) + self.assertIsNone(event.after_warehouse_id) + self.assertEqual(9, event.before_site_id) + self.assertEqual(9, event.after_site_id) + self.assertIsNone(event.before_line_id) + self.assertEqual(4, event.after_line_id) + self.assertEqual(8, event.before_remaining_packages) + self.assertEqual(5, event.after_remaining_packages) + self.assertEqual(7, event.operator_id) + self.assertEqual(self.fixed_time, event.occurred_at) + self.assertEqual('usage-key', event.idempotency_key) + self.assertEqual(usage.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(61, result.result) + + async def test_zero_remaining_closes_filling_but_keeps_line_until_return(self) -> None: + inventory = make_inventory( + current_line_id=4, + remaining_packages=3, + flow_state=PalletFlowState.AT_SITE_PARTIAL.value, + ) + update_filling = AsyncMock() + add_event = AsyncMock() + stack, mocks = self._patch_record_path( + inventory=inventory, + update_filling=update_filling, + add_event=add_event, + ) + with stack: + result = await UsageService.record_usage_services( + self.db, + make_page(used_packages=3), + self.user, + ) + + self.assertTrue(result.is_success) + self.assertEqual(PalletFlowState.AT_SITE_EMPTY.value, inventory.flow_state) + self.assertEqual(0, inventory.remaining_packages) + self.assertIsNone(inventory.active_filling_id) + self.assertEqual(4, inventory.current_line_id) + self.assertEqual(4, inventory.version) + self.assertIsNone(self.filling.active_slot) + self.assertEqual(self.fixed_time, self.filling.closed_at) + self.assertEqual('user-7', self.filling.update_by) + self.assertEqual(self.fixed_time, self.filling.update_time) + update_filling.assert_awaited_once_with(self.db, self.filling) + event = add_event.await_args.args[2] + self.assertEqual(PalletFlowState.AT_SITE_EMPTY.value, event.after_state) + self.assertEqual(0, event.after_remaining_packages) + self.assertEqual(4, event.after_line_id) + self.db.commit.assert_awaited_once_with() + self.db.rollback.assert_not_awaited() + mocks['add_usage'].assert_awaited_once() + + async def test_same_line_can_continue_but_different_line_is_rejected(self) -> None: + same_inventory = make_inventory( + current_line_id=4, + flow_state=PalletFlowState.AT_SITE_PARTIAL.value, + ) + stack, _ = self._patch_record_path(inventory=same_inventory) + with stack: + result = await UsageService.record_usage_services( + self.db, + make_page(used_packages=2), + self.user, + ) + self.assertTrue(result.is_success) + self.assertEqual(4, same_inventory.current_line_id) + self.assertEqual(6, same_inventory.remaining_packages) + + self.db.reset_mock() + other_inventory = make_inventory( + current_line_id=5, + flow_state=PalletFlowState.AT_SITE_PARTIAL.value, + ) + stack, mocks = self._patch_record_path(inventory=other_inventory) + with stack, self.assertRaises(ServiceException) as raised: + await UsageService.record_usage_services(self.db, self.page, self.user) + self.assertIn('已绑定其他产线', raised.exception.message) + mocks['add_usage'].assert_not_awaited() + mocks['add_event'].assert_not_awaited() + self.db.commit.assert_not_awaited() + self.db.rollback.assert_awaited_once() + + async def test_line_must_exist_be_enabled_and_belong_to_locked_site(self) -> None: + for line, message in ( + (None, '产线不存在'), + (make_line(status='1'), '产线已停用'), + (make_line(site_id=10), '产线不属于托盘当前客户现场'), + ): + with self.subTest(message=message): + self.db.reset_mock() + stack, mocks = self._patch_record_path(line=line) + if line is None: + mocks['line_lock'].return_value = None + with stack, self.assertRaises(ServiceException) as raised: + await UsageService.record_usage_services(self.db, self.page, self.user) + self.assertEqual(message, raised.exception.message) + mocks['add_usage'].assert_not_awaited() + mocks['add_event'].assert_not_awaited() + self.db.commit.assert_not_awaited() + self.db.rollback.assert_awaited_once() + + async def test_only_site_usage_states_and_strict_current_quantity_are_allowed(self) -> None: + cases = ( + ( + make_inventory(flow_state=PalletFlowState.AT_SITE_EMPTY.value, remaining_packages=0), + self.page, + '托盘当前状态不允许使用', + ), + ( + make_inventory(remaining_packages=8), + UsageCreateModel.model_construct( + qr_token='qr-2', + line_id=4, + used_packages=True, + idempotency_key='usage-key', + ), + '使用包数必须为正整数', + ), + (make_inventory(remaining_packages=8), make_page(used_packages=9), '使用包数不能超过当前余量'), + ) + for inventory, page, message in cases: + with self.subTest(message=message): + self.db.reset_mock() + stack, mocks = self._patch_record_path(inventory=inventory) + with stack, self.assertRaises(ServiceException) as raised: + await UsageService.record_usage_services(self.db, page, self.user) + self.assertEqual(message, raised.exception.message) + mocks['add_usage'].assert_not_awaited() + mocks['add_event'].assert_not_awaited() + self.db.commit.assert_not_awaited() + self.db.rollback.assert_awaited_once() + + async def test_initial_actual_site_denial_hides_qr_before_secondary_locks(self) -> None: + access = AsyncMock(side_effect=PermissionException(data='', message='无权访问该客户现场')) + stack, mocks = self._patch_record_path(access=access) + with stack, self.assertRaises(PermissionException) as raised: + await UsageService.record_usage_services(self.db, self.page, self.user) + + self.assertEqual('无权访问该托盘', raised.exception.message) + mocks['inventory_lock'].assert_awaited_once() + mocks['pallet_lock'].assert_not_awaited() + mocks['filling_lock'].assert_not_awaited() + mocks['site_lock'].assert_not_awaited() + mocks['line_lock'].assert_not_awaited() + mocks['add_usage'].assert_not_awaited() + self.db.rollback.assert_awaited_once() + + async def test_current_mapping_denial_wins_over_disabled_site_without_leaking_status(self) -> None: + async def access_check( + _db: AsyncSession, + _user: CurrentUserModel, + _site_id: int, + *, + for_update: bool = False, + ) -> None: + if for_update: + raise PermissionException(data='', message='无权访问该客户现场') + + access = AsyncMock(side_effect=access_check) + stack, mocks = self._patch_record_path(site=make_site(status='1'), access=access) + with stack, self.assertRaises(PermissionException) as raised: + await UsageService.record_usage_services(self.db, self.page, self.user) + + self.assertEqual('无权访问该托盘', raised.exception.message) + mocks['site_lock'].assert_awaited_once_with(self.db, 9, for_update=True) + access.assert_awaited_with(self.db, self.user, 9, for_update=True) + mocks['line_lock'].assert_not_awaited() + mocks['add_usage'].assert_not_awaited() + self.db.rollback.assert_awaited_once() + + async def test_admin_bypasses_mapping_but_gets_precise_site_errors(self) -> None: + for site, message in ((None, '客户现场不存在'), (make_site(status='1'), '客户现场已停用')): + with self.subTest(message=message): + self.db.reset_mock() + stack, mocks = self._patch_record_path(site=site) + if site is None: + mocks['site_lock'].return_value = None + with stack, self.assertRaises(ServiceException) as raised: + await UsageService.record_usage_services(self.db, self.page, make_user(1)) + self.assertEqual(message, raised.exception.message) + mocks['add_usage'].assert_not_awaited() + self.db.rollback.assert_awaited_once() + + async def test_locked_qr_asset_and_active_filling_invariants_fail_before_usage_write(self) -> None: + cases = ( + {'pallet': make_pallet(qr_token='other-qr')}, + {'pallet': make_pallet(asset_status='1')}, + {'inventory': make_inventory(active_filling_id=None)}, + {'filling': make_filling(pallet_id=3)}, + {'filling': make_filling(active_slot=None)}, + ) + for overrides in cases: + with self.subTest(overrides=tuple(overrides)): + self.db.reset_mock() + stack, mocks = self._patch_record_path(**overrides) + with stack, self.assertRaises(ServiceException): + await UsageService.record_usage_services(self.db, self.page, self.user) + mocks['add_usage'].assert_not_awaited() + mocks['update_filling'].assert_not_awaited() + mocks['add_event'].assert_not_awaited() + self.db.commit.assert_not_awaited() + self.db.rollback.assert_awaited_once() + + async def test_invalid_locked_inventory_version_fails_before_usage_write(self) -> None: + inventory = make_inventory(version=-1) + stack, mocks = self._patch_record_path(inventory=inventory) + with stack, self.assertRaises(ServiceException) as raised: + await UsageService.record_usage_services(self.db, self.page, self.user) + + self.assertEqual('托盘库存版本无效', raised.exception.message) + mocks['add_usage'].assert_not_awaited() + mocks['add_event'].assert_not_awaited() + self.db.rollback.assert_awaited_once() + + async def test_record_hides_missing_and_non_site_qr_but_admin_gets_precise_error(self) -> None: + with ( + patch.object(AccessDao, 'list_user_site_ids', new=AsyncMock(return_value=[9])), + patch.object(OperationDao, 'get_usage_by_idempotency_key', new=AsyncMock(return_value=None)), + patch.object(InventoryDao, 'get_pallet_by_qr_token', new=AsyncMock(return_value=None)), + self.assertRaises(PermissionException) as normal_missing, + ): + await UsageService.record_usage_services(self.db, self.page, self.user) + self.assertEqual('无权访问该托盘', normal_missing.exception.message) + self.db.rollback.assert_awaited_once() + + self.db.reset_mock() + with ( + patch.object(OperationDao, 'get_usage_by_idempotency_key', new=AsyncMock(return_value=None)), + patch.object(InventoryDao, 'get_pallet_by_qr_token', new=AsyncMock(return_value=None)), + self.assertRaises(ServiceException) as admin_missing, + ): + await UsageService.record_usage_services(self.db, self.page, make_user(1)) + self.assertEqual('托盘不存在', admin_missing.exception.message) + self.db.rollback.assert_awaited_once() + + self.db.reset_mock() + stack, mocks = self._patch_record_path(inventory=make_inventory(site_id=None)) + with stack, self.assertRaises(PermissionException) as non_site: + await UsageService.record_usage_services(self.db, self.page, self.user) + self.assertEqual('无权访问该托盘', non_site.exception.message) + mocks['pallet_lock'].assert_not_awaited() + self.db.rollback.assert_awaited_once() + + +@unittest.skipUnless(UsageService is not None, 'UsageService 尚未实现') +class TestUsageIdempotency(unittest.IsolatedAsyncioTestCase): + def setUp(self) -> None: + self.db = MagicMock(spec=AsyncSession) + self.user = make_user() + self.page = make_page() + self.core = TestRecordUsageCore(methodName='runTest') + self.core.setUp() + self.core.db = self.db + self.core.user = self.user + self.core.page = self.page + + def _exact_usage(self, *, request_hash: str | None = None) -> BizUsageRecord: + return make_usage( + request_hash=request_hash or UsageService._build_request_hash(self.page), + ) + + async def test_exact_retry_rechecks_current_mapping_and_returns_without_mutation(self) -> None: + existing = self._exact_usage() + inventory_lock = AsyncMock() + site_lock = AsyncMock(return_value=make_site(status='1')) + access = AsyncMock() + with ( + patch.object(AccessDao, 'list_user_site_ids', new=AsyncMock(return_value=[9])), + patch.object( + OperationDao, + 'get_usage_by_idempotency_key', + new=AsyncMock(return_value=existing), + ) as get_key, + patch.object(MasterDao, 'get_site', new=site_lock), + patch.object(AccessService, 'assert_site_access', new=access), + patch.object(InventoryDao, 'get_inventory_for_update', new=inventory_lock), + patch.object(OperationDao, 'add_usage_record', new=AsyncMock()) as add_usage, + ): + result = await UsageService.record_usage_services(self.db, self.page, self.user) + + get_key.assert_awaited_once_with(self.db, 'usage-key', site_ids=[9]) + site_lock.assert_awaited_once_with(self.db, 9, for_update=True) + access.assert_awaited_once_with(self.db, self.user, 9, for_update=True) + inventory_lock.assert_not_awaited() + add_usage.assert_not_awaited() + self.db.commit.assert_not_awaited() + self.db.rollback.assert_awaited_once() + self.assertEqual(61, result.result) + + async def test_same_key_with_different_hash_is_stable_conflict(self) -> None: + existing = self._exact_usage(request_hash='different-hash') + with ( + patch.object(AccessDao, 'list_user_site_ids', new=AsyncMock(return_value=[9])), + patch.object(OperationDao, 'get_usage_by_idempotency_key', new=AsyncMock(return_value=existing)), + patch.object(InventoryDao, 'get_inventory_for_update', new=AsyncMock()) as inventory_lock, + self.assertRaises(ServiceException) as raised, + ): + await UsageService.record_usage_services(self.db, self.page, self.user) + + self.assertIn('幂等键', raised.exception.message) + inventory_lock.assert_not_awaited() + self.db.rollback.assert_awaited_once() + + async def test_exact_retry_never_reuses_success_after_current_mapping_revocation(self) -> None: + existing = self._exact_usage() + access = AsyncMock(side_effect=PermissionException(data='', message='无权访问该客户现场')) + with ( + patch.object(AccessDao, 'list_user_site_ids', new=AsyncMock(return_value=[9])), + patch.object(OperationDao, 'get_usage_by_idempotency_key', new=AsyncMock(return_value=existing)), + patch.object(MasterDao, 'get_site', new=AsyncMock(return_value=make_site())), + patch.object(AccessService, 'assert_site_access', new=access), + self.assertRaises(PermissionException), + ): + await UsageService.record_usage_services(self.db, self.page, self.user) + + access.assert_awaited_once_with(self.db, self.user, 9, for_update=True) + self.db.commit.assert_not_awaited() + self.db.rollback.assert_awaited_once() + + async def test_integrity_error_recovers_only_fresh_scoped_exact_result(self) -> None: + recovered = self._exact_usage() + scope = AsyncMock(side_effect=[[9], [9]]) + key_read = AsyncMock(side_effect=[None, recovered]) + add_usage = AsyncMock(side_effect=integrity_error()) + stack, mocks = self.core._patch_record_path( + site_scope=scope, + key_read=key_read, + add_usage=add_usage, + ) + with stack: + result = await UsageService.record_usage_services(self.db, self.page, self.user) + + self.assertEqual(61, result.result) + self.assertEqual(2, scope.await_count) + self.assertEqual(2, key_read.await_count) + self.assertEqual([9], key_read.await_args_list[1].kwargs['site_ids']) + self.assertEqual(2, self.db.rollback.await_count) + self.db.commit.assert_not_awaited() + mocks['add_event'].assert_not_awaited() + + async def test_integrity_recovery_rejects_different_hash_or_missing_result(self) -> None: + for recovered in (self._exact_usage(request_hash='different-hash'), None): + with self.subTest(recovered=recovered is not None): + self.db.reset_mock() + key_read = AsyncMock(side_effect=[None, recovered]) + stack, _ = self.core._patch_record_path( + key_read=key_read, + add_usage=AsyncMock(side_effect=integrity_error()), + ) + with stack, self.assertRaises(ServiceException) as raised: + await UsageService.record_usage_services(self.db, self.page, self.user) + self.assertTrue('幂等' in raised.exception.message or '冲突' in raised.exception.message) + self.assertEqual(2, self.db.rollback.await_count) + self.db.commit.assert_not_awaited() + + async def test_integrity_recovery_rechecks_fresh_scope_and_rejects_revoked_site(self) -> None: + scope = AsyncMock(side_effect=[[9], []]) + key_read = AsyncMock(return_value=None) + stack, mocks = self.core._patch_record_path( + site_scope=scope, + key_read=key_read, + add_usage=AsyncMock(side_effect=integrity_error()), + ) + with stack, self.assertRaises(PermissionException): + await UsageService.record_usage_services(self.db, self.page, self.user) + + self.assertEqual(2, scope.await_count) + self.assertEqual(1, key_read.await_count) + self.assertEqual(2, self.db.rollback.await_count) + mocks['add_event'].assert_not_awaited() + self.db.commit.assert_not_awaited() + + async def test_unexpected_write_failure_rolls_back_once_and_never_commits(self) -> None: + stack, _ = self.core._patch_record_path( + add_event=AsyncMock(side_effect=RuntimeError('event failed')), + ) + with stack, self.assertRaises(RuntimeError): + await UsageService.record_usage_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(UsageService, '_build_request_hash', side_effect=RuntimeError('hash failed')), + self.assertRaises(RuntimeError), + ): + await UsageService.record_usage_services(self.db, self.page, self.user) + + self.db.rollback.assert_awaited_once() + self.db.commit.assert_not_awaited() + + +if __name__ == '__main__': + unittest.main()