功能：隔离五类测试业务链路

shierfeng-fastapi-backend/module_business/service/inbound_service.py

@@ -8,7 +8,7 @@ from sqlalchemy.exc import IntegrityError
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from common.vo import CrudResponseModel, PageModel
-from exceptions.exception import PermissionException, ServiceException
+from exceptions.exception import PermissionException, ServiceException, ServiceWarning
 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
@@ -16,6 +16,7 @@ from module_business.dao.master_dao import MasterDao
 from module_business.dao.operation_dao import OperationDao
 from module_business.domain.enums import DocumentStatus, PalletEventType, PalletFlowState
 from module_business.domain.pallet_state_machine import PalletStateMachine
+from module_business.domain.test_batch_scope import TestBatchScope
 from module_business.entity.do.master_do import BizPallet
 from module_business.entity.do.operation_do import (
     BizInboundItem,
@@ -35,6 +36,7 @@ from module_business.entity.vo.operation_vo import (
     PalletScanResponseModel,
 )
 from module_business.service.access_service import AccessService
+from module_business.service.test_batch_access_service import TestBatchAccessService
 
 
 class InboundService:
@@ -95,9 +97,10 @@ class InboundService:
         query_db: AsyncSession,
         warehouse_id: int,
         current_user: CurrentUserModel,
+        test_scope: TestBatchScope,
     ) -> Any:
         await AccessService.assert_warehouse_access(query_db, current_user, warehouse_id)
-        warehouse = await MasterDao.get_warehouse(query_db, warehouse_id)
+        warehouse = await MasterDao.get_warehouse(query_db, warehouse_id, test_scope)
         if warehouse is None:
             raise ServiceException(message='目标仓库不存在')
         if warehouse.status != '0':
@@ -113,6 +116,7 @@ class InboundService:
         warehouse_ids: set[int],
         hidden_warehouse_ids: set[int],
         require_active: bool,
+        test_scope: TestBatchScope,
     ) -> None:
         """锁定仓库及授权映射，并基于锁后最新状态重新校验写权限。"""
         warehouses: dict[int, Any] = {}
@@ -121,6 +125,7 @@ class InboundService:
             warehouses[warehouse_id] = await MasterDao.get_warehouse(
                 query_db,
                 warehouse_id,
+                test_scope,
                 for_update=True,
             )
             try:
@@ -152,6 +157,7 @@ class InboundService:
         inbound: BizInboundOrder,
         current_user: CurrentUserModel,
         *,
+        test_scope: TestBatchScope,
         require_active: bool = True,
     ) -> None:
         await cls._lock_and_authorize_warehouses(
@@ -160,6 +166,7 @@ class InboundService:
             warehouse_ids={inbound.warehouse_id},
             hidden_warehouse_ids={inbound.warehouse_id},
             require_active=require_active,
+            test_scope=test_scope,
         )
 
     @staticmethod
@@ -183,10 +190,11 @@ class InboundService:
         cls,
         query_db: AsyncSession,
         pallet_ids: list[int],
+        test_scope: TestBatchScope,
     ) -> tuple[list[BizPalletInventory], list[BizPallet]]:
         ordered_ids = sorted(pallet_ids)
-        inventories = await InventoryDao.get_inventories_for_update(query_db, ordered_ids)
-        pallets = await InventoryDao.get_pallets_by_ids(query_db, ordered_ids)
+        inventories = await InventoryDao.get_inventories_for_update(query_db, ordered_ids, test_scope)
+        pallets = await InventoryDao.get_pallets_by_ids(query_db, ordered_ids, test_scope)
         expected_ids = set(ordered_ids)
         inventory_by_id = {inventory.pallet_id: inventory for inventory in inventories}
         pallet_by_id = {pallet.pallet_id: pallet for pallet in pallets}
@@ -205,6 +213,7 @@ class InboundService:
         pallet_ids: list[int],
         user_name: str,
         current_time: datetime,
+        test_batch_id: int | None,
     ) -> list[BizInboundItem]:
         return [
             BizInboundItem(
@@ -212,6 +221,7 @@ class InboundService:
                 pallet_id=pallet_id,
                 create_by=user_name,
                 create_time=current_time,
+                test_batch_id=test_batch_id,
             )
             for pallet_id in sorted(pallet_ids)
         ]
@@ -226,6 +236,7 @@ class InboundService:
         query_db: AsyncSession,
         query_object: InboundPageQueryModel,
         current_user: CurrentUserModel,
+        test_scope: TestBatchScope,
         is_page: bool = True,
     ) -> PageModel | list[InboundModel]:
         warehouse_ids = await cls._warehouse_scope(query_db, current_user)
@@ -239,8 +250,9 @@ class InboundService:
         result = await OperationDao.get_inbound_list(
             query_db,
             query_object,
-            is_page,
             warehouse_ids=warehouse_ids,
+            scope=test_scope,
+            is_page=is_page,
         )
         rows = result.rows if isinstance(result, PageModel) else result
         models = [InboundModel.model_validate(record) for record in rows]
@@ -254,11 +266,13 @@ class InboundService:
         query_db: AsyncSession,
         inbound_id: int,
         current_user: CurrentUserModel,
+        test_scope: TestBatchScope,
     ) -> InboundModel:
         warehouse_ids = await cls._warehouse_scope(query_db, current_user)
         inbound = await OperationDao.get_inbound_by_id(
             query_db,
             inbound_id,
+            test_scope,
             warehouse_ids=warehouse_ids,
         )
         if inbound is None:
@@ -270,12 +284,14 @@ class InboundService:
         cls,
         query_db: AsyncSession,
         current_user: CurrentUserModel,
+        test_scope: TestBatchScope,
     ) -> list[BusinessOptionModel]:
         warehouse_ids = await cls._warehouse_scope(query_db, current_user)
         allowed_ids = None if warehouse_ids is None else set(warehouse_ids)
         warehouses = await MasterDao.list_warehouse(
             query_db,
             WarehousePageQueryModel(status='0'),
+            test_scope,
             False,
         )
         options = []
@@ -301,22 +317,23 @@ class InboundService:
         qr_token: str,
         query_object: InboundScanQueryModel,
         current_user: CurrentUserModel,
+        test_scope: TestBatchScope,
     ) -> PalletScanResponseModel:
         await AccessService.assert_warehouse_access(
             query_db,
             current_user,
             query_object.warehouse_id,
         )
-        warehouse = await MasterDao.get_warehouse(query_db, query_object.warehouse_id)
+        warehouse = await MasterDao.get_warehouse(query_db, query_object.warehouse_id, test_scope)
         if warehouse is None:
             raise ServiceException(message='目标仓库不存在')
         if warehouse.status != '0':
             raise ServiceException(message='目标仓库已停用')
 
-        pallet = await InventoryDao.get_pallet_by_qr_token(query_db, qr_token)
+        pallet = await InventoryDao.get_pallet_by_qr_token(query_db, qr_token, test_scope)
         if pallet is None:
             raise ServiceException(message='托盘不存在')
-        inventory = await MasterDao.get_pallet_inventory(query_db, pallet.pallet_id)
+        inventory = await MasterDao.get_pallet_inventory(query_db, pallet.pallet_id, test_scope)
         if inventory is None:
             raise ServiceException(message='托盘库存不存在')
         cls._validate_pallet_pair(pallet, inventory)
@@ -344,22 +361,26 @@ class InboundService:
         query_db: AsyncSession,
         page_object: InboundCreateModel,
         current_user: CurrentUserModel,
+        test_scope: TestBatchScope,
     ) -> CrudResponseModel:
         try:
+            await TestBatchAccessService.lock_active_batch(query_db, test_scope)
             _, user_name = cls._current_identity(current_user)
             await cls._validate_active_warehouse(
                 query_db,
                 page_object.warehouse_id,
                 current_user,
+                test_scope,
             )
             ordered_ids = sorted(page_object.pallet_ids)
-            await cls._lock_and_validate_pallets(query_db, ordered_ids)
+            await cls._lock_and_validate_pallets(query_db, ordered_ids, test_scope)
             await cls._lock_and_authorize_warehouses(
                 query_db,
                 current_user,
                 warehouse_ids={page_object.warehouse_id},
                 hidden_warehouse_ids=set(),
                 require_active=True,
+                test_scope=test_scope,
             )
 
             current_time = datetime.now()
@@ -371,12 +392,13 @@ class InboundService:
                 create_time=current_time,
                 update_by=user_name,
                 update_time=current_time,
+                test_batch_id=test_scope.batch_id,
             )
             inbound = await OperationDao.add_inbound(query_db, inbound)
             await OperationDao.replace_inbound_items(
                 query_db,
                 inbound.inbound_id,
-                cls._build_items(inbound.inbound_id, ordered_ids, user_name, current_time),
+                cls._build_items(inbound.inbound_id, ordered_ids, user_name, current_time, test_scope.batch_id),
             )
             await query_db.commit()
             return cls._success('新增成功')
@@ -393,13 +415,16 @@ class InboundService:
         query_db: AsyncSession,
         page_object: InboundUpdateModel,
         current_user: CurrentUserModel,
+        test_scope: TestBatchScope,
     ) -> CrudResponseModel:
         try:
+            await TestBatchAccessService.lock_active_batch(query_db, test_scope)
             _, user_name = cls._current_identity(current_user)
             warehouse_ids = await cls._warehouse_scope(query_db, current_user)
             inbound = await OperationDao.get_inbound_for_update(
                 query_db,
                 page_object.inbound_id,
+                test_scope,
                 warehouse_ids=warehouse_ids,
             )
             if inbound is None:
@@ -410,15 +435,17 @@ class InboundService:
                 query_db,
                 page_object.warehouse_id,
                 current_user,
+                test_scope,
             )
             ordered_ids = sorted(page_object.pallet_ids)
-            await cls._lock_and_validate_pallets(query_db, ordered_ids)
+            await cls._lock_and_validate_pallets(query_db, ordered_ids, test_scope)
             await cls._lock_and_authorize_warehouses(
                 query_db,
                 current_user,
                 warehouse_ids={inbound.warehouse_id, page_object.warehouse_id},
                 hidden_warehouse_ids={inbound.warehouse_id},
                 require_active=True,
+                test_scope=test_scope,
             )
 
             current_time = datetime.now()
@@ -429,7 +456,7 @@ class InboundService:
             await OperationDao.replace_inbound_items(
                 query_db,
                 inbound.inbound_id,
-                cls._build_items(inbound.inbound_id, ordered_ids, user_name, current_time),
+                cls._build_items(inbound.inbound_id, ordered_ids, user_name, current_time, test_scope.batch_id),
             )
             await query_db.commit()
             return cls._success('修改成功')
@@ -446,47 +473,9 @@ class InboundService:
         query_db: AsyncSession,
         inbound_ids: list[int],
         current_user: CurrentUserModel,
+        test_scope: TestBatchScope,
     ) -> CrudResponseModel:
-        try:
-            if not inbound_ids or any(inbound_id <= 0 for inbound_id in inbound_ids):
-                raise ServiceException(message='入库单ID必须为正整数')
-            if len(inbound_ids) != len(set(inbound_ids)):
-                raise ServiceException(message='入库单ID不能重复')
-
-            ordered_ids = sorted(inbound_ids)
-            warehouse_ids = await cls._warehouse_scope(query_db, current_user)
-            locked_inbounds = []
-            for inbound_id in ordered_ids:
-                inbound = await OperationDao.get_inbound_for_update(
-                    query_db,
-                    inbound_id,
-                    warehouse_ids=warehouse_ids,
-                )
-                if inbound is None:
-                    cls._raise_inbound_not_accessible(current_user)
-                if inbound.status != DocumentStatus.DRAFT.value:
-                    raise ServiceException(message='已确认入库单不允许删除')
-                locked_inbounds.append(inbound)
-
-            locked_warehouse_ids = {inbound.warehouse_id for inbound in locked_inbounds}
-            await cls._lock_and_authorize_warehouses(
-                query_db,
-                current_user,
-                warehouse_ids=locked_warehouse_ids,
-                hidden_warehouse_ids=locked_warehouse_ids,
-                require_active=False,
-            )
-
-            for inbound_id in ordered_ids:
-                await OperationDao.delete_inbound(query_db, inbound_id)
-            await query_db.commit()
-            return cls._success('删除成功')
-        except IntegrityError:
-            await query_db.rollback()
-            raise ServiceException(message='删除入库单失败，单据状态或关联数据已变化') from None
-        except Exception:
-            await query_db.rollback()
-            raise
+        raise ServiceWarning(message='不支持单条物理删除，请通过测试数据批次清理')
 
     @staticmethod
     def _is_exact_confirmed(
@@ -510,12 +499,14 @@ class InboundService:
         idempotency_key: str,
         request_hash: str,
         current_user: CurrentUserModel,
+        test_scope: TestBatchScope,
     ) -> CrudResponseModel:
         try:
             warehouse_ids = await cls._warehouse_scope(query_db, current_user)
             current = await OperationDao.get_inbound_by_id(
                 query_db,
                 inbound_id,
+                test_scope,
                 warehouse_ids=warehouse_ids,
             )
             if current is None:
@@ -523,6 +514,7 @@ class InboundService:
             recovered = await OperationDao.get_inbound_by_confirm_idempotency_key(
                 query_db,
                 idempotency_key,
+                test_scope,
                 warehouse_ids=[current.warehouse_id],
             )
             if recovered is None:
@@ -537,6 +529,7 @@ class InboundService:
             locked_recovered = await OperationDao.get_inbound_for_update(
                 query_db,
                 inbound_id,
+                test_scope,
                 warehouse_ids=warehouse_ids,
             )
             if locked_recovered is None:
@@ -552,6 +545,7 @@ class InboundService:
                 query_db,
                 locked_recovered,
                 current_user,
+                test_scope=test_scope,
                 require_active=False,
             )
             return cls._success('确认成功')
@@ -565,13 +559,16 @@ class InboundService:
         inbound_id: int,
         page_object: ConfirmCommandModel,
         current_user: CurrentUserModel,
+        test_scope: TestBatchScope,
     ) -> CrudResponseModel:
         try:
+            await TestBatchAccessService.lock_active_batch(query_db, test_scope)
             user_id, user_name = cls._current_identity(current_user)
             warehouse_ids = await cls._warehouse_scope(query_db, current_user)
             inbound = await OperationDao.get_inbound_by_id(
                 query_db,
                 inbound_id,
+                test_scope,
                 warehouse_ids=warehouse_ids,
             )
             if inbound is None:
@@ -582,6 +579,7 @@ class InboundService:
             existing = await OperationDao.get_inbound_by_confirm_idempotency_key(
                 query_db,
                 idempotency_key,
+                test_scope,
                 warehouse_ids=[inbound.warehouse_id],
             )
             if existing is not None:
@@ -601,6 +599,7 @@ class InboundService:
             locked_inbound = await OperationDao.get_inbound_for_update(
                 query_db,
                 inbound_id,
+                test_scope,
                 warehouse_ids=warehouse_ids,
             )
             if locked_inbound is None:
@@ -616,6 +615,7 @@ class InboundService:
                         query_db,
                         locked_inbound,
                         current_user,
+                        test_scope=test_scope,
                         require_active=False,
                     )
                     await query_db.rollback()
@@ -634,11 +634,12 @@ class InboundService:
             if len(pallet_ids) != len(set(pallet_ids)):
                 raise ServiceException(message='入库单托盘不能重复')
             ordered_ids = sorted(pallet_ids)
-            inventories, _ = await cls._lock_and_validate_pallets(query_db, ordered_ids)
+            inventories, _ = await cls._lock_and_validate_pallets(query_db, ordered_ids, test_scope)
             await cls._lock_and_reauthorize_warehouse(
                 query_db,
                 locked_inbound,
                 current_user,
+                test_scope=test_scope,
             )
 
             for inventory in inventories:
@@ -691,6 +692,7 @@ class InboundService:
                     occurred_at=current_time,
                     idempotency_key=idempotency_key,
                     request_hash=request_hash,
+                    test_batch_id=test_scope.batch_id,
                 )
                 await InventoryDao.add_event(query_db, inventory, event)
 
@@ -704,6 +706,7 @@ class InboundService:
                 page_object.idempotency_key,
                 cls._build_confirm_request_hash(inbound_id),
                 current_user,
+                test_scope,
             )
         except Exception:
             await query_db.rollback()